#!/usr/bin/env python3
"""
Minecraft Username Sniper GUI v9 - NUCLEAR PACKAGE
- Multiple OAuth tokens with connection pooling
- NAMEMC (windowed) + LABYNAMES (exact) timing
- Pre-warmed HTTP/2 connections
- Smart retry logic with exponential backoff
- Instant success detection + auto-stop
- Sound + desktop alerts on success
- Token health validation
- DNS pre-resolution
- Adaptive thread control
"""
import sys
import time
import httpx
import threading
import json
import gc
import socket
import platform
import subprocess
from datetime import datetime, timezone
from PyQt5.QtWidgets import (
QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QTabWidget, QLabel, QLineEdit, QTextEdit, QPushButton, QSpinBox,
QGroupBox, QFormLayout, QMessageBox, QCheckBox, QFileDialog,
QProgressBar, QComboBox, QSlider
)
from PyQt5.QtCore import Qt, QTimer, pyqtSignal, QThread
from PyQt5.QtGui import QFont, QColor, QIcon
# === Disable garbage collection for precision timing ===
gc.disable()
# === Pre-resolve DNS for speed ===
MINECRAFT_API_HOST = "api.minecraftservices.com"
MINECRAFT_API_IP = None
# === FEATURE #6: FINGERPRINT ROTATION ===
# Pool of realistic browser User-Agent strings
USER_AGENTS = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 14_2) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Safari/605.1.15",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Edge/120.0.0.0",
]
# Track per-endpoint rate limit cooldowns
endpoint_cooldowns = {} # endpoint -> (expires_at_timestamp, reason)
def get_random_user_agent():
"""Get a random User-Agent from the pool"""
import random
return random.choice(USER_AGENTS)
def is_endpoint_cooldown(endpoint):
"""Check if endpoint is in cooldown due to 429 Retry-After"""
if endpoint in endpoint_cooldowns:
expires_at, reason = endpoint_cooldowns[endpoint]
if time.time() < expires_at:
remaining = int(expires_at - time.time())
return True, remaining, reason
return False, 0, None
def set_endpoint_cooldown(endpoint, retry_after_seconds, reason="429"):
"""Set a cooldown for an endpoint based on Retry-After header"""
expires_at = time.time() + retry_after_seconds
endpoint_cooldowns[endpoint] = (expires_at, reason)
# === FEATURE #8: TOKEN ROTATION ON RATE LIMIT ===
# Track which tokens have been rate limited to avoid reusing them
token_rate_limits = {} # token_hash -> (expires_at, count)
def is_token_rate_limited(token):
"""Check if a specific token has been rate limited"""
token_hash = hash(token) % 1000000
if token_hash in token_rate_limits:
expires_at, count = token_rate_limits[token_hash]
if time.time() < expires_at:
remaining = int(expires_at - time.time())
return True, remaining, count
return False, 0, 0
def set_token_rate_limit(token, duration=60):
"""Mark a token as rate limited for a duration"""
token_hash = hash(token) % 1000000
expires_at = time.time() + duration
current_count = token_rate_limits.get(token_hash, (0, 0))[1]
token_rate_limits[token_hash] = (expires_at, current_count + 1)
def get_usable_tokens(tokens):
"""Filter out rate-limited tokens and return usable ones"""
usable = []
skipped = 0
for token in tokens:
is_limited, remaining, count = is_token_rate_limited(token)
if is_limited:
skipped += 1
else:
usable.append(token)
return usable, skipped
def pre_resolve_dns():
"""Resolve DNS before drop time for speed"""
global MINECRAFT_API_IP
try:
MINECRAFT_API_IP = socket.gethostbyname(MINECRAFT_API_HOST)
return True, MINECRAFT_API_IP
except Exception as e:
return False, str(e)
# ============================================================================
# FEATURE #9: ACCOUNT LOADING HELPERS
# ============================================================================
def load_accounts_from_file(filepath: str) -> list:
"""
Load accounts from file (email:password or bearer tokens)
Format: One account per line
- email:password (triggers OAuth flow)
- eyJ... (raw bearer token, length > 200)
"""
accounts = []
try:
with open(filepath, 'r') as f:
for line in f:
line = line.strip()
if not line:
continue
# Detect bearer token (MCsniperGO pattern)
if len(line) > 200 and ':' not in line and line.startswith('eyJ'):
accounts.append({'type': 'token', 'token': line})
elif ':' in line:
# Email:password format
parts = line.split(':', 1)
if len(parts) == 2 and parts[0] and parts[1]:
accounts.append({'type': 'credential', 'email': parts[0], 'password': parts[1]})
else:
# Treat as plain bearer token
accounts.append({'type': 'token', 'token': line})
except Exception as e:
print(f"[ERROR] Failed to load accounts from {filepath}: {e}")
return accounts
def authenticate_accounts(accounts: list, log_callback=None) -> dict:
"""
Authenticate a list of accounts and return bearer tokens
Returns: {email_or_hash: token} dict
"""
log = log_callback or print
tokens = {}
for idx, account in enumerate(accounts):
if account['type'] == 'token':
# Already a bearer token
token_hash = account['token'][:8] + '...'
tokens[f"token_{idx}"] = account['token']
log(f"[AUTH] Loaded token {token_hash} (skipping OAuth)")
else:
# Email:password - authenticate via OAuth
email = account['email']
password = account['password']
try:
auth = MicrosoftAuth(email, password, log_callback=log)
token = auth.authenticate()
tokens[email] = token
log(f"[AUTH] ✓ Authenticated {email}")
except Exception as e:
log(f"[AUTH] ✗ Failed to authenticate {email}: {e}")
return tokens
def save_tokens_to_file(tokens_dict: dict, filepath: str):
"""Save authenticated tokens to file (one per line)"""
try:
with open(filepath, 'w') as f:
for token in tokens_dict.values():
f.write(token + '\n')
except Exception as e:
print(f"[ERROR] Failed to save tokens: {e}")
# ============================================================================
# FEATURE #9: AUTO-AUTH SYSTEM (Microsoft OAuth Device Flow)
# ============================================================================
# Authenticates email:password → generates Minecraft bearer tokens
# Based on MCsniperGO's flow: Microsoft OAuth → XBL → XSTS → Minecraft
# ============================================================================
MICROSOFT_CLIENT_ID = "00000000441cc96b" # Minecraft for Nintendo Switch
class MicrosoftAuth:
"""Handles full Microsoft OAuth device flow to Minecraft bearer tokens"""
def __init__(self, email: str, password: str, log_callback=None):
self.email = email
self.password = password
self.log = log_callback or print
self.session = httpx.Client(
headers={"User-Agent": get_random_user_agent()},
timeout=30.0
)
def log_step(self, step: str, status: str = "✓"):
"""Log authentication step"""
self.log(f"[AUTH] {step} {status}")
def authenticate(self) -> str:
"""
Full auth flow: email/password → Microsoft OAuth → XBL → XSTS → Minecraft
Returns: Minecraft bearer token (or raises exception)
"""
self.log_step("Starting authentication", "→")
# Step 1: Get Microsoft OAuth access token
self.log_step("1. Microsoft OAuth login", "...")
ms_token = self._get_microsoft_token()
if not ms_token:
raise Exception("Microsoft OAuth failed")
self.log_step("1. Microsoft OAuth login", "✓")
# Step 2: Get XBL (Xbox Live) token
self.log_step("2. Xbox Live token", "...")
xbl_token = self._get_xbl_token(ms_token)
if not xbl_token:
raise Exception("XBL token failed")
self.log_step("2. Xbox Live token", "✓")
# Step 3: Get XSTS token (Xbox Live Sandbox Security Token Service)
self.log_step("3. XSTS token", "...")
xsts_token = self._get_xsts_token(xbl_token)
if not xsts_token:
raise Exception("XSTS token failed")
self.log_step("3. XSTS token", "✓")
# Step 4: Get Minecraft bearer token
self.log_step("4. Minecraft bearer token", "...")
mc_token = self._get_minecraft_token(xsts_token)
if not mc_token:
raise Exception("Minecraft token failed")
self.log_step("4. Minecraft bearer token", "✓")
self.log_step("Authentication complete!", "✓✓✓")
return mc_token
def _get_microsoft_token(self) -> str:
"""Step 1: Microsoft OAuth device flow"""
# Request device code
url = "https://login.live.com/oauth20_token.srf"
data = {
"client_id": MICROSOFT_CLIENT_ID,
"scope": "service::user.auth.xboxlive.com::MBI_SSL",
"grant_type": "device_code"
}
response = self.session.post(url, data=data)
if response.status_code != 200:
self.log_step(f"Microsoft OAuth failed: {response.status_code}", "✗")
return ""
result = response.json()
device_code = result.get("device_code")
user_code = result.get("user_code")
verification_url = result.get("verification_url")
interval = result.get("interval", 5) # polling interval in seconds
if not device_code or not user_code:
self.log_step("No device code received", "✗")
return ""
# Show user the verification info
self.log(f"\n{'='*60}")
self.log(f"🔐 AUTHENTICATION REQUIRED")
self.log(f"{'='*60}")
self.log(f"1. Go to: {verification_url}")
self.log(f"2. Enter code: {user_code}")
self.log(f"3. Sign in with: {self.email}")
self.log(f"4. Wait for authorization...")
self.log(f"{'='*60}\n")
# Poll for token
poll_url = "https://login.live.com/oauth20_token.srf"
poll_data = {
"client_id": MICROSOFT_CLIENT_ID,
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
"device_code": device_code
}
max_attempts = 60 # 5 minutes max
for attempt in range(max_attempts):
time.sleep(interval)
response = self.session.post(poll_url, data=poll_data)
if response.status_code == 200:
result = response.json()
return result.get("access_token", "")
elif response.status_code == 400:
result = response.json()
error = result.get("error", "")
if error == "authorization_pending":
continue # Still waiting
elif error == "access_denied":
self.log_step("User denied authorization", "✗")
return ""
else:
self.log_step(f"Polling error: {error}", "✗")
return ""
self.log_step("Authentication timeout", "✗")
return ""
def _get_xbl_token(self, ms_token: str) -> dict:
"""Step 2: Get Xbox Live token"""
url = "https://user.auth.xboxlive.com/user/authenticate"
headers = {
"Authorization": f"Bearer {ms_token}",
"Content-Type": "application/json",
"X-Xbl-Contract": "1"
}
data = {
"Properties": {
"AuthMethod": "RPS",
"SiteName": "user.auth.xboxlive.com",
"RpsTicket": f"d={ms_token}"
},
"RelyingParty": "http://auth.xboxlive.com",
"TokenType": "JWT"
}
response = self.session.post(url, headers=headers, json=data)
if response.status_code != 200:
self.log_step(f"XBL failed: {response.status_code}", "✗")
return {}
return response.json()
def _get_xsts_token(self, xbl_data: dict) -> dict:
"""Step 3: Get XSTS token"""
url = "https://xsts.auth.xboxlive.com/xsts/authorize"
headers = {
"Content-Type": "application/json",
"X-Xbl-Contract": "1"
}
# Extract XBL token and user info
xbl_token = xbl_data.get("Token", "")
user_hwid = xbl_data.get("DisplayClaims", {}).get("xui", [{}])[0].get("userHWId", "")
data = {
"Properties": {
"SandboxId": "RETAIL",
"UserTokens": [xbl_token],
"UserTileId": user_hwid # Some accounts need this
},
"RelyingParty": "rp://api.minecraftservices.com/",
"TokenType": "JWT"
}
response = self.session.post(url, headers=headers, json=data)
if response.status_code != 200:
self.log_step(f"XSTS failed: {response.status_code}", "✗")
return {}
return response.json()
def _get_minecraft_token(self, xsts_data: dict) -> str:
"""Step 4: Get Minecraft bearer token"""
url = "https://api.minecraftservices.com/authentication/login_with_xbox"
headers = {
"Content-Type": "application/json"
}
xsts_token = xsts_data.get("Token", "")
data = {
"identity": {
"issuer": "xp",
"token": xsts_token
}
}
response = self.session.post(url, headers=headers, json=data)
if response.status_code != 200:
self.log_step(f"Minecraft token failed: {response.status_code}", "✗")
return ""
return response.json().get("access_token", "")
def authenticate_accounts(account_list: list, log_callback=None) -> dict:
"""
Authenticate a list of email:password accounts
Returns: {email: bearer_token} dict
"""
log = log_callback or print
results = {}
for account in account_list:
if ":" not in account:
# Already a bearer token
log(f"[AUTH] Using pre-existing token for: {account[:30]}...")
results[account[:20]] = account
continue
email, password = account.split(":", 1)
log(f"[AUTH] Authenticating: {email}")
try:
auth = MicrosoftAuth(email, password, log)
token = auth.authenticate()
results[email] = token
log(f"[AUTH] ✓ Success for {email}\n")
except Exception as e:
log(f"[AUTH] ✗ Failed for {email}: {e}\n")
return results
def load_accounts_from_file(filepath: str) -> list:
"""Load accounts from file (email:password or bearer tokens)"""
try:
with open(filepath, 'r') as f:
accounts = [line.strip() for line in f if line.strip() and not line.startswith('#')]
return accounts
except FileNotFoundError:
return []
def save_tokens_to_file(tokens: dict, filepath: str):
"""Save authenticated tokens to file for reuse"""
with open(filepath, 'w') as f:
for email, token in tokens.items():
f.write(f"{email}:{token}\n")
# === Pre-warmed connection pool for NAMEMC ===
namemc_client = httpx.Client(
limits=httpx.Limits(max_connections=100, max_keepalive_connections=50)
)
# === Global success flag to stop all threads ===
success_achieved = threading.Event()
def play_success_sound():
"""Play sound alert on success"""
try:
system = platform.system()
if system == "Linux":
# Try multiple sound files
for sound_file in ["/usr/share/sounds/freedesktop/stereo/complete.oga",
"/usr/share/sounds/ubuntu-submitted/stereo/alarm-clock.oga",
"/usr/share/sounds/freedesktop/stereo/message.oga"]:
try:
if subprocess.call(["which", "aplay"], stdout=subprocess.DEVNULL) == 0:
subprocess.run(["aplay", "-q", sound_file], timeout=2)
return
else:
subprocess.run(["play", sound_file], timeout=2)
return
except:
continue
elif system == "Darwin": # macOS
subprocess.run(["afplay", "/System/Library/Sounds/Glass.aiff"], timeout=2)
elif system == "Windows":
subprocess.run(["powershell", "-Command", "[System.Media.SystemSounds]::Beep.Play()"], timeout=2)
except Exception as e:
pass # Silent fail if sound not available
def send_desktop_notification(title, message):
"""Send desktop notification on success"""
try:
system = platform.system()
if system == "Linux":
subprocess.run(["notify-send", "-u", "critical", "-t", "10000", title, message], timeout=2)
elif system == "Darwin": # macOS
subprocess.run(["osascript", "-e", f'display notification "{message}" with title "{title}"'], timeout=2)
elif system == "Windows":
# Windows notification would require a library, skip for now
pass
except Exception as e:
pass # Silent fail
class SniperWorker(QThread):
"""Worker thread for firing concurrent requests"""
log_signal = pyqtSignal(str)
progress_signal = pyqtSignal(int)
def __init__(self, names, tokens, drop_time_utc, threads_count, timing_mode="exact", priority_mode="fresh_first", warmup_enabled=True, warmup_duration=30, adaptive_enabled=True, aggressiveness=7, fingerprint_rotation=True, token_rotation=True, temporal_jitter=True, response_analysis=True, multi_name_mode="sequential", stagger_seconds=0):
super().__init__()
self.names = names if isinstance(names, list) else [names] # Support single or multiple names
self.tokens = tokens
self.drop_time_utc = drop_time_utc
self.threads_count = threads_count
self.timing_mode = timing_mode
self.priority_mode = priority_mode
self.warmup_enabled = warmup_enabled
self.warmup_duration = warmup_duration # seconds before drop to start warming
self.stop_event = threading.Event()
self.success_event = threading.Event()
self.requests_fired = 0
self.requests_success = 0
self.requests_failed = 0
self.connection_warmed = False
self.warmup_requests_sent = 0
self.warmup_success = 0
self.warmup_failures = 0
# === MULTI-NAME MODE ===
self.multi_name_mode = multi_name_mode # "sequential", "parallel", or "staggered"
self.stagger_seconds = stagger_seconds # seconds between names (for staggered mode)
self.name_results = {} # Track results per name: {name: {"success": bool, "message": str}}
self.current_name_index = 0
self.names_completed = 0
self.total_names = len(self.names)
# === AUTO-TOKEN REFRESH ===
self.auto_refresh_tokens = False # Will be set by GUI if auto-refresh enabled
self.token_refresh_window = 300 # 5 minutes before drop (default, can be overridden)
# === FEATURE #5: ADAPTIVE THROTTLING SETTINGS ===
self.adaptive_enabled = adaptive_enabled
self.aggressiveness = aggressiveness # 1-10 scale, affects how aggressively we adapt
# === FEATURE #6: FINGERPRINT ROTATION ===
# Rotate User-Agent per request to avoid fingerprint-based rate limiting
self.fingerprint_rotation = fingerprint_rotation
self.current_user_agent = get_random_user_agent()
self.fingerprint_rotations = 0 # track how many times we've rotated
# === FEATURE #4: ERROR LEARNING (AUTO-BLACKLIST) ===
# Track errors per token and blacklist consistently failing ones
self.token_error_counts = {} # token_hash -> error_count
self.token_error_types = {} # token_hash -> {status_code: count}
self.blacklisted_tokens = set() # tokens that exceeded error threshold
self.ERROR_THRESHOLD = 5 # blacklist after 5 consecutive errors
self.BLACKLIST_THRESHOLD = 0.8 # blacklist if >80% of errors are same type
# === FEATURE #8: TOKEN ROTATION ON RATE LIMIT ===
# Rotate to different tokens when one gets rate limited
self.token_rotation_enabled = token_rotation
self.token_rotation_count = 0
self.rate_limited_tokens = set() # tokens currently rate limited
self.token_rotation_duration = 60 # seconds to skip a rate-limited token
# === FEATURE #13: TEMPORAL JITTER ===
# Add microsecond-level randomization to evade detection
self.temporal_jitter_enabled = temporal_jitter
self.jitter_range_ms = 3 # spread requests over ±3ms window
self.jitter_applied_count = 0
# === FEATURE #15: RESPONSE ANALYSIS & EARLY ABORT ===
# Monitor responses in real-time and abort if outcome determined
self.response_analysis_enabled = response_analysis
self.early_abort_on_success = True # Always abort on success
self.failure_patterns = ["name taken", "already registered", "invalid token", "authentication failed", "token invalid"]
self.consecutive_failures = 0
self.success_indicators = ["success", "created", "ok", "claimed"]
self.failure_count = 0
self.success_count = 0
self.max_consecutive_failures = 50 # abort after 50 same-type failures
# === FEATURE #5: ADAPTIVE THROTTLING ===
# Dynamically adjust request rate based on competition and error rates
self.competition_level = "LOW" # LOW, MEDIUM, HIGH, EXTREME
self.last_100_requests = [] # track recent request outcomes
self.rate_limit_spike = False # detect sudden 429 increase
self.adaptive_threads = self.threads_count # current thread count (can be reduced)
self.adaptive_delay = 0.0 # current delay between bursts (can be increased)
self.COMPETITION_WINDOW = 100 # analyze last 100 requests
self.RATE_LIMIT_THRESHOLD = 0.3 # >30% 429s = rate limit spike
self.COMPETITION_HIGH_THRESHOLD = 0.4 # >40% failures = high competition
self.COMPETITION_EXTREME_THRESHOLD = 0.6 # >60% failures = extreme competition
self.MIN_THREADS = max(1, self.threads_count // 4) # never go below 25%
self.MAX_DELAY = 2.0 # max delay between bursts (seconds)
def run(self):
"""Main sniper logic - supports multi-name mode"""
# === FEATURE #7: MULTI-NAME MODE ===
if len(self.names) > 1:
self.log_signal.emit(f"[*] Starting multi-name sniper: {len(self.names)} name(s)")
self.log_signal.emit(f"[*] Mode: {self.multi_name_mode}, Stagger: {self.stagger_seconds}s")
self._run_multi_name_mode()
else:
# Single name mode (backward compatible)
self.log_signal.emit(f"[*] Starting sniper for: {self.names[0]}")
self.log_signal.emit(f"[*] Mode: {self.timing_mode}, Tokens: {len(self.tokens)}, Threads: {self.threads_count}")
# Pre-resolve DNS
self.log_signal.emit("[*] Pre-resolving DNS...")
dns_success, dns_result = pre_resolve_dns()
if dns_success:
self.log_signal.emit(f"[*] DNS resolved: {MINECRAFT_API_HOST} → {dns_result}")
else:
self.log_signal.emit(f"[!] DNS warning: {dns_result}")
# Validate tokens
self.log_signal.emit("[*] Validating tokens...")
valid_tokens = self._validate_tokens()
if len(valid_tokens) == 0:
self.log_signal.emit("[ERROR] No valid tokens found!")
return
self.tokens = valid_tokens
self.log_signal.emit(f"[*] {len(valid_tokens)} valid tokens ready")
# Pre-warm connections
self.log_signal.emit("[*] Pre-warming HTTP/2 connections...")
self._warm_connections()
self.log_signal.emit("[✓] Connections warmed and ready!")
if self.timing_mode == "exact":
self._exact_timing_snipe()
else:
self._windowed_timing_snipe()
def _run_multi_name_mode(self):
"""Handle multiple usernames with staggered/sequential/parallel execution"""
self.log_signal.emit(f"[*] Multi-name queue: {', '.join(self.names)}")
for idx, name in enumerate(self.names):
if self.stop_event.is_set():
break
self.log_signal.emit(f"\n{'='*60}")
self.log_signal.emit(f"[*] Processing name {idx+1}/{len(self.names)}: {name}")
self.log_signal.emit(f"{'='*60}\n")
# Set current name
self.name = name
# === FEATURE #9: AUTO-REFRESH TOKENS ===
if self.auto_refresh_tokens:
self.log_signal.emit(f"🔄 Auto-refresh: Authenticating new tokens for {name}")
self._auto_authenticate_tokens()
# Pre-resolve DNS
dns_success, dns_result = pre_resolve_dns()
if dns_success:
self.log_signal.emit(f"[*] DNS resolved: {MINECRAFT_API_HOST} → {dns_result}")
# Validate tokens
valid_tokens = self._validate_tokens()
if len(valid_tokens) == 0:
self.log_signal.emit(f"[ERROR] No valid tokens for {name}!")
self.name_results[name] = {"success": False, "reason": "No valid tokens"}
continue
self.tokens = valid_tokens
# Pre-warm connections
self._warm_connections()
# Snipe this name
if self.timing_mode == "exact":
self._exact_timing_snipe()
else:
self._windowed_timing_snipe()
# Stagger delay
if self.multi_name_mode == "staggered" and idx < len(self.names) - 1:
self.log_signal.emit(f"[*] Stagger delay: {self.stagger_seconds}s before next name")
for _ in range(self.stagger_seconds):
if self.stop_event.is_set():
break
time.sleep(1)
# Print summary
if len(self.names) > 1:
self.log_signal.emit(f"\n{'='*60}")
self.log_signal.emit(f"📊 MULTI-NAME SUMMARY:")
for name, result in self.name_results.items():
status = "✅ SUCCESS" if result.get("success") else "❌ FAILED"
self.log_signal.emit(f" {name}: {status}")
self.log_signal.emit(f"{'='*60}\n")
def _auto_authenticate_tokens(self):
"""Auto-authenticate from stored credentials (FEATURE #9)"""
# This would load credentials from a secure storage and re-authenticate
# For now, we'll just log the placeholder
self.log_signal.emit("⚠️ Auto-auth placeholder: Implement credential storage")
# TODO: Implement OAuth flow with stored credentials
def _validate_tokens(self):
"""Check which tokens are valid before drop time"""
valid_tokens = []
test_url = "https://api.minecraftservices.com/minecraft/profile"
for i, token in enumerate(self.tokens):
if self.stop_event.is_set():
break
client = httpx.Client(
http2=True,
timeout=httpx.Timeout(3.0, connect=1.5),
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
},
)
try:
r = client.get(test_url, timeout=2)
if r.status_code in [200, 201, 204]:
valid_tokens.append(token)
self.log_signal.emit(f"[✓] Token {i+1} validated")
else:
self.log_signal.emit(f"[✗] Token {i+1} invalid (status {r.status_code})")
except Exception as e:
self.log_signal.emit(f"[✗] Token {i+1} failed: {str(e)[:30]}")
finally:
client.close()
# Small delay between validation requests
time.sleep(0.1)
return valid_tokens
def _warm_connections(self):
"""Pre-warm HTTP/2 connections for instant firing"""
warm_url = "https://api.minecraftservices.com/minecraft/profile"
for i in range(min(3, len(self.tokens))): # Warm with first 3 tokens
if self.stop_event.is_set():
break
token = self.tokens[i]
client = httpx.Client(
transport=httpx.HTTPTransport(pool=connection_pool),
http2=True,
timeout=httpx.Timeout(3.0, connect=1.5),
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
},
)
try:
client.get(warm_url, timeout=2)
# Keep connection alive by not closing immediately
time.sleep(0.05)
except:
pass
finally:
client.close()
self.connection_warmed = True
def _token_warmup_request(self, token, username=None):
"""
Send a harmless profile request to keep token fresh in Minecraft's cache.
Optional: can use GET /profile/{username} for variety to avoid rate limits.
Returns: (success: bool, status_code: int, response_time_ms: float)
"""
# Alternate between different harmless endpoints to avoid detection
import random
endpoint_type = random.choice(["profile", "profile", "profile", "sketch"]) # 75% profile, 25% sketch
if endpoint_type == "profile":
url = "https://api.minecraftservices.com/minecraft/profile"
else:
# Use sketch endpoint (harmless, just returns skin data)
url = "https://api.minecraftservices.com/minecraft/profile/skin"
client = httpx.Client(
transport=httpx.HTTPTransport(pool=connection_pool),
http2=True,
timeout=httpx.Timeout(2.0, connect=1.0),
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
},
)
start_time = time.time()
try:
r = client.get(url, timeout=1.5)
response_time = (time.time() - start_time) * 1000 # ms
# Success: 200, 201, 204 are all valid
if r.status_code in [200, 201, 204]:
return True, r.status_code, response_time
else:
return False, r.status_code, response_time
except httpx.TimeoutException:
return False, 0, (time.time() - start_time) * 1000
except Exception as e:
return False, 0, (time.time() - start_time) * 1000
finally:
client.close()
def _run_token_warmup_phase(self, target_ts):
"""
FEATURE #1: Token Warm-Up System
Run continuous harmless requests starting warmup_duration seconds before drop time.
This keeps tokens "fresh" in Minecraft's session cache, reducing first-request latency.
"""
if not self.warmup_enabled:
self.log_signal.emit("[*] Token warm-up: DISABLED")
return
warmup_start_ts = target_ts - self.warmup_duration
self.log_signal.emit(f"[🔥] TOKEN WARM-UP SYSTEM: ENABLED")
self.log_signal.emit(f"[🔥] Duration: {self.warmup_duration}s before drop (starts in {self.warmup_duration}s)")
self.log_signal.emit(f"[🔥] Strategy: Continuous profile requests to keep cache fresh")
# Wait until warmup start time
wait_start = time.time()
while not self.stop_event.is_set():
now = time.time()
delta_to_warmup = warmup_start_ts - now
if delta_to_warmup <= 0:
break
elif delta_to_warmup > 5:
time.sleep(5)
else:
time.sleep(0.5)
# Update progress bar during wait
delta_to_drop = target_ts - now
if delta_to_drop > 0:
progress = min(99, int((1 - delta_to_drop / (target_ts - wait_start)) * 100))
self.progress_signal.emit(max(0, progress))
if self.stop_event.is_set():
return
# === WARM-UP PHASE STARTED ===
self.log_signal.emit(f"[🔥] 🔥🔥🔥 TOKEN WARM-UP PHASE STARTED! 🔥🔥🔥")
self.log_signal.emit(f"[🔥] Dropping in {self.warmup_duration}s - firing harmless requests now...")
warmup_cycle = 0
last_status_update = time.time()
# Continue until drop time
while not self.stop_event.is_set():
now = time.time()
if now >= target_ts:
break
# Fire warmup requests with all tokens every 2-4 seconds
import random
for token in self.tokens:
if self.stop_event.is_set() or now >= target_ts:
break
success, status, resp_time = self._token_warmup_request(token)
self.warmup_requests_sent += 1
if success:
self.warmup_success += 1
else:
self.warmup_failures += 1
if status in [401, 403]:
self.log_signal.emit(f"[⚠️] Token returned {status} - may be invalid!")
# Tiny delay between tokens to avoid thundering herd
time.sleep(0.02)
warmup_cycle += 1
# Status update every 10 seconds
if time.time() - last_status_update > 10:
remaining = int(target_ts - time.time())
avg_resp = (self.warmup_success + self.warmup_failures) / max(1, self.warmup_requests_sent) * 100
self.log_signal.emit(f"[🔥] Warm-up status: {self.warmup_requests_sent} requests, {self.warmup_success}✓ {self.warmup_failures}✗ | {remaining}s until drop")
last_status_update = time.time()
# Wait 2-4 seconds before next cycle (randomized to avoid patterns)
wait_time = random.uniform(2.0, 4.0)
time.sleep(min(wait_time, target_ts - time.time()))
# === WARM-UP PHASE COMPLETE ===
self.log_signal.emit(f"[🔥] ✅ TOKEN WARM-UP COMPLETE!")
self.log_signal.emit(f"[🔥] Stats: {self.warmup_requests_sent} requests | {self.warmup_success} success | {self.warmup_failures} failed")
if self.warmup_failures > self.warmup_success * 0.3: # More than 30% failure rate
self.log_signal.emit(f"[⚠️] High warm-up failure rate detected! Some tokens may be cold or invalid.")
else:
self.log_signal.emit(f"[✓] All tokens warmed and ready for snipe!")
def _exact_timing_snipe(self):
"""Fire at exact drop time, then continue firing every 2-5 seconds"""
try:
target = datetime.fromisoformat(self.drop_time_utc.replace("Z", "+00:00"))
target_ts = target.timestamp()
self.log_signal.emit(f"[*] Waiting for exact drop time: {self.drop_time_utc} UTC")
# === RUN TOKEN WARM-UP PHASE ===
self._run_token_warmup_phase(target_ts)
if self.stop_event.is_set():
self.log_signal.emit("[!] Sniper stopped by user")
return
self.progress_signal.emit(100)
self.log_signal.emit(f"[!] 🎯🎯🎯 EXACT DROP TIME! FIRING NOW! 🎯🎯🎯")
# === FEATURE #2: NUCLEAR MODE - FIRST BURST ===
# Fire immediately at exact time with ZERO DELAY (nuclear mode)
self.log_signal.emit(f"[!] ⚛️ ACTIVATING NUCLEAR MODE FOR INITIAL BURST!")
self._fire_all_tokens_burst(nuclear_mode=True)
# Check if we succeeded on the first nuclear burst
if self.success_event.is_set():
return
self.log_signal.emit(f"[!] ⚠️ Nuclear burst complete, switching to controlled bursts...")
self.log_signal.emit(f"[!] ⚠️ Continuing to fire every 2-5 seconds until success...")
self.log_signal.emit(f"[!] ⚠️ Press STOP to quit anytime!")
# Continue firing periodically (2-5 seconds) after the exact time
import random
burst_count = 1
while not self.stop_event.is_set():
if self.success_event.is_set():
self.log_signal.emit("[✓] Success detected! Stopping exact mode")
break
burst_count += 1
wait_time = random.uniform(2.0, 5.0)
self.log_signal.emit(f"[*] ⏱️ Waiting {wait_time:.1f}s before next burst...")
# Check stop event during wait
for _ in range(int(wait_time * 10)):
if self.stop_event.is_set() or self.success_event.is_set():
break
time.sleep(0.1)
if self.stop_event.is_set() or self.success_event.is_set():
break
self.log_signal.emit(f"[!] 🔥 Follow-up Burst #{burst_count} firing...")
self._fire_all_tokens_burst()
if not self.success_event.is_set():
self.log_signal.emit("[!] Sniper stopped by user")
except Exception as e:
self.log_signal.emit(f"[ERROR] Timing error: {e}")
def _windowed_timing_snipe(self):
"""Fire requests continuously every 1-3 seconds starting at drop time (NO END)"""
try:
target = datetime.fromisoformat(self.drop_time_utc.replace("Z", "+00:00"))
target_ts = target.timestamp()
self.log_signal.emit(f"[*] Waiting for drop time: {self.drop_time_utc} UTC")
# Wait until the exact drop time
while not self.stop_event.is_set():
now = time.time()
delta = target_ts - now
if delta <= 0:
break
elif delta > 1:
time.sleep(delta - 1)
else:
time.sleep(0.1)
if self.stop_event.is_set():
self.log_signal.emit("[!] Sniper stopped by user")
return
self.log_signal.emit(f"[!] 🎯🎯🎯 DROP TIME REACHED! Firing continuously every 1-3 seconds...")
self.log_signal.emit(f"[!] ⚠️ This will run until success or manual stop!")
# Fire continuously every 1-3 seconds (NO TIME LIMIT)
import random
burst_count = 0
while not self.stop_event.is_set():
if self.success_event.is_set():
self.log_signal.emit("[✓] Success detected! Stopping windowed mode")
break
burst_count += 1
# === FEATURE #2: NUCLEAR MODE - FIRST BURST ONLY ===
if burst_count == 1:
self.log_signal.emit(f"[!] ⚛️ ACTIVATING NUCLEAR MODE FOR INITIAL BURST!")
self._fire_all_tokens_burst(nuclear_mode=True)
else:
self.log_signal.emit(f"[!] 🔥 Burst #{burst_count} firing...")
self._fire_all_tokens_burst()
# Adaptive delay based on success rate (1-3 seconds)
if self.requests_fired > 0:
success_rate = self.requests_success / self.requests_fired
if success_rate > 0.3: # High success rate, slow down
wait_time = random.uniform(2.0, 3.0)
else:
wait_time = random.uniform(1.0, 2.5)
else:
wait_time = random.uniform(1.0, 3.0)
self.log_signal.emit(f"[*] ⏱️ Next burst in {wait_time:.1f}s... (Press STOP to quit)")
# Check stop event during wait
for _ in range(int(wait_time * 10)):
if self.stop_event.is_set() or self.success_event.is_set():
break
time.sleep(0.1)
if not self.success_event.is_set():
self.log_signal.emit("[!] Sniper stopped by user")
except Exception as e:
self.log_signal.emit(f"[ERROR] Windowed timing error: {e}")
def _fire_all_tokens_burst(self, nuclear_mode=False):
"""Fire concurrent requests with all tokens in controlled bursts
Args:
nuclear_mode: If True, fires ALL threads simultaneously with 0ms delay (maximum aggression)
"""
global success_achieved
# Sort tokens by priority
tokens_to_use = self.tokens.copy()
if self.priority_mode == "fresh_first":
# Assume newer tokens (later in list) are less likely rate-limited
tokens_to_use = tokens_to_use[::-1]
# === FEATURE #4: FILTER OUT BLACKLISTED TOKENS=***
active_tokens = [t for t in tokens_to_use if not self._is_token_blacklisted(t)]
filtered_count = len(tokens_to_use) - len(active_tokens)
if filtered_count > 0:
self.log_signal.emit(f"[⚫] Filtering {filtered_count} blacklisted tokens. Active: {len(active_tokens)}/{len(tokens_to_use)}")
# === FEATURE #5: USE ADAPTIVE THREAD COUNT ===
threads_to_use = self.adaptive_threads if not nuclear_mode else self.threads_count
threads_list = []
burst_size = len(active_tokens) * threads_to_use
if nuclear_mode:
self.log_signal.emit(f"[!] ⚛️⚛️⚛️ NUCLEAR BURST! Firing {burst_size} requests SIMULTANEOUSLY! ⚛️⚛️⚛️")
else:
self.log_signal.emit(f"[!] Firing {burst_size} requests across {len(active_tokens)} tokens (Threads: {threads_to_use}, Competition: {self.competition_level})")
if nuclear_mode:
# === NUCLEAR MODE: ZERO DELAY ===
# Fire ALL threads instantly - maximum aggression
for i, token in enumerate(active_tokens):
if self.success_event.is_set() or self.stop_event.is_set():
break
for j in range(self.threads_count):
if self.success_event.is_set():
break
# === FEATURE #13: APPLY TEMPORAL JITTER ===
if self.temporal_jitter_enabled:
# Add random jitter between -3ms and +3ms
jitter_ms = random.uniform(-self.jitter_range_ms, self.jitter_range_ms)
jitter_sec = jitter_ms / 1000.0
time.sleep(max(0, jitter_sec)) # only sleep if positive
self.jitter_applied_count += 1
t = threading.Thread(
target=self._send_request_with_retry,
args=(token,),
daemon=True # Daemon threads for nuclear mode
)
t.start()
threads_list.append(t)
# No staggering in nuclear mode - pure speed
else:
# === STANDARD MODE: CONTROLLED BURST ===
# Fire in controlled bursts to avoid overwhelming
for i, token in enumerate(active_tokens):
if self.success_event.is_set() or self.stop_event.is_set():
break
for j in range(threads_to_use):
if self.success_event.is_set():
break
# === FEATURE #13: APPLY TEMPORAL JITTER ===
if self.temporal_jitter_enabled:
# Add random jitter between -3ms and +3ms
jitter_ms = random.uniform(-self.jitter_range_ms, self.jitter_range_ms)
jitter_sec = jitter_ms / 1000.0
time.sleep(max(0, jitter_sec)) # only sleep if positive
self.jitter_applied_count += 1
t = threading.Thread(
target=self._send_request_with_retry,
args=(token,)
)
t.start()
threads_list.append(t)
# Small stagger between token groups (10ms)
time.sleep(0.01)
# Wait for all threads to complete
for t in threads_list:
t.join(timeout=5.0)
# === FEATURE #13: LOG JITTER STATS ===
if self.temporal_jitter_enabled and self.jitter_applied_count > 0:
self.log_signal.emit(f"[⚡] Temporal jitter applied to {self.jitter_applied_count} requests (±{self.jitter_range_ms}ms)")
# === FEATURE #15: LOG RESPONSE ANALYSIS STATS ===
if self.response_analysis_enabled:
self.log_signal.emit(f"[📊] Response analysis: {self.success_count} successes, {self.failure_count} failures detected")
if self.consecutive_failures > 0:
self.log_signal.emit(f"[📊] Consecutive failure streak: {self.consecutive_failures}")
# Check if we achieved success
if self.success_event.is_set():
self.log_signal.emit(f"[🏆] SUCCESS ACHIEVED! Total requests: {self.requests_fired}")
self._trigger_success_alerts()
def _send_request_with_retry(self, token):
"""Send request with smart retry logic, fingerprint rotation, and Retry-After detection"""
global success_achieved
url = f"https://api.minecraftservices.com/minecraft/profile/name/{self.name}"
endpoint = "minecraft/profile/name"
# === FEATURE #3: FAST-FAIL TIMEOUT ===
# Fast-fail threshold: if request takes > 3 seconds, abort immediately
FAST_FAIL_TIMEOUT = 3.0 # seconds
max_retries = 3
retry_delays = [0.1, 0.3, 0.6] # Exponential backoff
for attempt in range(max_retries):
if self.success_event.is_set() or self.stop_event.is_set():
return
# === FEATURE #6: CHECK ENDPOINT COOLDOWN ===
# Respect Retry-After headers from previous 429 responses
in_cooldown, remaining, reason = is_endpoint_cooldown(endpoint)
if in_cooldown:
self.log_signal.emit(f"[🚫] {endpoint} in cooldown ({remaining}s remaining) - {reason}")
return # Abort this request, endpoint is on cooldown
# === FEATURE #6: ROTATE FINGERPRINT ===
# Get a fresh User-Agent for this request to avoid fingerprint-based rate limiting
if self.fingerprint_rotation:
self.current_user_agent = get_random_user_agent()
self.fingerprint_rotations += 1
client = httpx.Client(
transport=httpx.HTTPTransport(pool=connection_pool),
http2=True,
# Use fast-fail timeout for connect, longer for read
timeout=httpx.Timeout(FAST_FAIL_TIMEOUT, connect=FAST_FAIL_TIMEOUT),
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"User-Agent": self.current_user_agent, # === FEATURE #6: ROTATING USER-AGENT ===
},
)
timestamp = datetime.now(timezone.utc).isoformat()
try:
# === FEATURE #3: FAST-FAIL WITH TIMER ===
start_time = time.time()
r = client.put(url, json={})
elapsed = time.time() - start_time
# Fast-fail check: if request took too long, log and abort
if elapsed > FAST_FAIL_TIMEOUT:
self.requests_failed += 1
self.log_signal.emit(f"[⚠️] FAST-FAIL: Request took {elapsed:.2f}s (> {FAST_FAIL_TIMEOUT}s) - Token {hash(token) % 10000}")
return # Abort this token immediately
try:
body = r.json()
except:
body = r.text
# Update stats
self.requests_fired += 1
# Check for success
if r.status_code in [200, 201, 204]:
self.requests_success += 1
self.log_signal.emit(f"[🏆🏆🏆 SUCCESS! {r.status_code} ({elapsed:.3f}s) - Token {hash(token) % 10000}]")
self.success_event.set()
success_achieved.set()
# === FEATURE #15: COUNT SUCCESS ===
self.success_count += 1
# === FEATURE #5: TRACK OUTCOME ===
self._track_request_outcome("SUCCESS", elapsed)
return
# Check for duplicate (you already own it!)
if r.status_code == 403 and isinstance(body, dict) and body.get('details', {}).get('status') == 'DUPLICATE':
self.requests_success += 1
self.log_signal.emit(f"[🏆 ALREADY OWNED! 403 DUPLICATE - You have the name!]")
self.success_event.set()
success_achieved.set()
# === FEATURE #15: COUNT SUCCESS ===
self.success_count += 1
# === FEATURE #5: TRACK OUTCOME ===
self._track_request_outcome("DUPLICATE", elapsed)
return
# === FEATURE #6: HANDLE 429 WITH RETRY-AFTER DETECTION ===
if r.status_code == 429:
# Check for Retry-After header (seconds until we can retry)
retry_after = r.headers.get("Retry-After")
# === FEATURE #8: MARK TOKEN AS RATE LIMITED ===
if self.token_rotation_enabled:
set_token_rate_limit(token, self.token_rotation_duration)
self.rate_limited_tokens.add(token)
self.token_rotation_count += 1
usable, skipped = get_usable_tokens(self.tokens)
self.log_signal.emit(f"[🔄] Token rate limited - Rotated #{self.token_rotation_count} | Usable: {len(usable)}/{len(self.tokens)} | Token {hash(token) % 10000}")
if retry_after:
try:
retry_after_seconds = int(retry_after)
# Set endpoint cooldown to respect the server's rate limit
set_endpoint_cooldown(endpoint, retry_after_seconds, "429")
self.log_signal.emit(f"[🚫] RATE LIMITED (429) - Cooldown: {retry_after_seconds}s - {endpoint}")
# Track this as a rate limit event for adaptive throttling
self._track_request_outcome("RATE_LIMITED", elapsed)
return # Abort all requests to this endpoint until cooldown expires
except ValueError:
# Retry-After might be a date, fall through to standard retry
pass
# No Retry-After or couldn't parse it, use standard retry logic
if attempt < max_retries - 1:
wait_time = retry_delays[attempt]
self.log_signal.emit(f"[⏱️] Rate limited (429), retrying in {wait_time}s...")
time.sleep(wait_time)
continue
# Max retries exceeded
self.requests_failed += 1
self._track_request_outcome("FAILURE", elapsed)
self.log_signal.emit(f"[{timestamp}] ❌ 429 MAX RETRIES - Token {hash(token) % 10000}")
return
# Failed request
self.requests_failed += 1
# === FEATURE #15: RESPONSE ANALYSIS ===
if self.response_analysis_enabled:
# Check response body for failure patterns
response_text = str(body).lower() if body else ""
matched_pattern = None
for pattern in self.failure_patterns:
if pattern in response_text:
matched_pattern = pattern
self.failure_count += 1
self.consecutive_failures += 1
break
else:
self.consecutive_failures = 0
# Check for success indicators in body
if any(indicator in response_text for indicator in self.success_indicators):
self.success_count += 1
# Early abort: too many consecutive same-type failures
if self.consecutive_failures >= self.max_consecutive_failures:
self.log_signal.emit(f"[🛑] Early abort triggered: {self.consecutive_failures} consecutive '{matched_pattern}' responses")
self.stop_event.set()
# Early abort: success already achieved (redundancy check)
if self.early_abort_on_success and self.success_count > 0:
self.stop_event.set()
# === FEATURE #4: ERROR LEARNING - TRACK ERRORS ===
token_hash = hash(token) % 10000
self._track_error(token_hash, r.status_code)
# === FEATURE #5: TRACK OUTCOME ===
self._track_request_outcome("FAILURE", elapsed)
status = "❌"
self.log_signal.emit(f"[{timestamp}] {status} {r.status_code} - Token {token_hash}")
except httpx.TimeoutException:
self.requests_failed += 1
# === FEATURE #4: TRACK TIMEOUT ERRORS ===
token_hash = hash(token) % 10000
self._track_error(token_hash, "TIMEOUT")
# === FEATURE #5: TRACK OUTCOME ===
self._track_request_outcome("TIMEOUT", elapsed)
if attempt < max_retries - 1:
wait_time = retry_delays[attempt]
self.log_signal.emit(f"[⏱️] TIMEOUT, retrying in {wait_time}s...")
time.sleep(wait_time)
continue
else:
self.log_signal.emit(f"[{timestamp}] TIMEOUT - Token {token_hash}")
except Exception as e:
self.requests_failed += 1
# === FEATURE #4: TRACK EXCEPTION ERRORS ===
token_hash = hash(token) % 10000
self._track_error(token_hash, "EXCEPTION")
# === FEATURE #5: TRACK OUTCOME ===
self._track_request_outcome("EXCEPTION", elapsed)
self.log_signal.emit(f"[{timestamp}] ERROR - {str(e)[:50]}")
finally:
client.close()
# If we got here, either succeeded or max retries reached
if self.success_event.is_set():
return
def _track_error(self, token_hash, error_type):
"""Track errors per token and blacklist consistently failing ones"""
# Initialize tracking for this token if needed
if token_hash not in self.token_error_counts:
self.token_error_counts[token_hash] = 0
self.token_error_types[token_hash] = {}
# Increment error count
self.token_error_counts[token_hash] += 1
# Track error type distribution
if error_type not in self.token_error_types[token_hash]:
self.token_error_types[token_hash][error_type] = 0
self.token_error_types[token_hash][error_type] += 1
# Check if token should be blacklisted
self._check_blacklist(token_hash)
def _check_blacklist(self, token_hash):
"""Check if token should be blacklisted based on error patterns"""
if token_hash in self.blacklisted_tokens:
return # Already blacklisted
error_count = self.token_error_counts.get(token_hash, 0)
# Check if error count exceeds threshold
if error_count >= self.ERROR_THRESHOLD:
error_types = self.token_error_types.get(token_hash, {})
total_errors = sum(error_types.values())
# Find most common error type
if total_errors > 0:
most_common_error = max(error_types.items(), key=lambda x: x[1])
dominant_error_rate = most_common_error[1] / total_errors
# Blacklist if dominant error rate exceeds threshold
if dominant_error_rate >= self.BLACKLIST_THRESHOLD:
self.blacklisted_tokens.add(token_hash)
self.log_signal.emit(f"[⚫] BLACKLISTED: Token {token_hash} ({error_count} errors, {most_common_error[0]}: {most_common_error[1]})")
self.log_signal.emit(f" Pattern: {most_common_error[1]}/{total_errors} ({dominant_error_rate*100:.0f}%) were {most_common_error[0]}")
def _is_token_blacklisted(self, token):
"""Check if a token is blacklisted"""
return hash(token) % 10000 in self.blacklisted_tokens
# === FEATURE #5: ADAPTIVE THROTTLING METHODS ===
def _track_request_outcome(self, outcome, elapsed_time):
"""Track request outcome for adaptive analysis"""
self.last_100_requests.append({
"outcome": outcome,
"elapsed": elapsed_time,
"timestamp": time.time()
})
# Keep only last N requests
if len(self.last_100_requests) > self.COMPETITION_WINDOW:
self.last_100_requests.pop(0)
# Analyze and adapt if we have enough data
if len(self.last_100_requests) >= 20:
self._analyze_and_adapt()
def _analyze_and_adapt(self):
"""Analyze recent requests and adapt strategy"""
if len(self.last_100_requests) < 20:
return
# Calculate metrics
total = len(self.last_100_requests)
successes = sum(1 for r in self.last_100_requests if r["outcome"] in ["SUCCESS", "DUPLICATE"])
failures = sum(1 for r in self.last_100_requests if r["outcome"] in ["FAILURE", "TIMEOUT", "EXCEPTION"])
rate_limits = sum(1 for r in self.last_100_requests if r["outcome"] == "FAILURE" and hasattr(self, 'last_status') and self.last_status == 429)
failure_rate = failures / total
avg_response_time = sum(r["elapsed"] for r in self.last_100_requests) / total
# Detect rate limit spike
recent_429s = sum(1 for r in self.last_100_requests[-10:] if r["outcome"] == "FAILURE")
current_rate_limit_spike = recent_429s / 10 > self.RATE_LIMIT_THRESHOLD
# Determine competition level
old_competition = self.competition_level
if failure_rate > self.COMPETITION_EXTREME_THRESHOLD:
self.competition_level = "EXTREME"
elif failure_rate > self.COMPETITION_HIGH_THRESHOLD:
self.competition_level = "HIGH"
elif failure_rate > 0.2:
self.competition_level = "MEDIUM"
else:
self.competition_level = "LOW"
# Adapt threads and delay based on competition
old_threads = self.adaptive_threads
old_delay = self.adaptive_delay
if self.competition_level == "EXTREME":
# Slow down significantly
self.adaptive_threads = max(self.MIN_THREADS, self.threads_count // 2)
self.adaptive_delay = min(self.MAX_DELAY, self.adaptive_delay + 0.5)
self.rate_limit_spike = True
elif self.competition_level == "HIGH":
# Moderate slowdown
self.adaptive_threads = max(self.MIN_THREADS, self.threads_count * 3 // 4)
self.adaptive_delay = min(self.MAX_DELAY, self.adaptive_delay + 0.3)
self.rate_limit_spike = current_rate_limit_spike
elif self.competition_level == "MEDIUM":
# Slight adjustment
self.adaptive_threads = max(self.MIN_THREADS, self.threads_count * 7 // 8)
self.adaptive_delay = min(0.5, self.adaptive_delay + 0.1)
self.rate_limit_spike = False
else: # LOW
# Ramp back up to full power
self.adaptive_threads = min(self.threads_count, self.adaptive_threads + 2)
self.adaptive_delay = max(0.0, self.adaptive_delay - 0.1)
self.rate_limit_spike = False
# Log changes
if old_competition != self.competition_level or old_threads != self.adaptive_threads or abs(old_delay - self.adaptive_delay) > 0.01:
emoji = {"LOW": "🟢", "MEDIUM": "🟡", "HIGH": "🟠", "EXTREME": "🔴"}.get(self.competition_level, "⚪")
self.log_signal.emit(f"\\n{emoji} [ADAPTIVE] Competition: {self.competition_level} | Threads: {old_threads}→{self.adaptive_threads} | Delay: {old_delay:.2f}s→{self.adaptive_delay:.2f}s")
self.log_signal.emit(f" Stats: {successes} success / {failures} failures ({failure_rate*100:.0f}%) | Avg response: {avg_response_time*1000:.0f}ms")
if self.rate_limit_spike:
self.log_signal.emit(f" ⚠️ RATE LIMIT SPIKE DETECTED - Backing off!")
def _trigger_success_alerts(self):
"""Trigger all success alerts"""
self.log_signal.emit(f"\n{'='*60}")
self.log_signal.emit(f"[🎉🎉🎉 USERNAME SECURED: {self.name} 🎉🎉🎉]")
self.log_signal.emit(f"{'='*60}\n")
# Play sound
play_success_sound()
# Send desktop notification
send_desktop_notification(
"🏆 MINECRAFT NAME SECURED!",
f"Username '{self.name}' has been claimed successfully!"
)
# Additional log
self.log_signal.emit(f"[✓] Total requests fired: {self.requests_fired}")
self.log_signal.emit(f"[✓] Successful: {self.requests_success}")
self.log_signal.emit(f"[✓] Failed: {self.requests_failed}")
# === FEATURE #4: BLACKLIST SUMMARY ===
if self.blacklisted_tokens:
self.log_signal.emit(f"\n[⚫] BLACKLIST SUMMARY: {len(self.blacklisted_tokens)} tokens blacklisted")
for token_hash in sorted(self.blacklisted_tokens):
error_count = self.token_error_counts.get(token_hash, 0)
error_types = self.token_error_types.get(token_hash, {})
if error_types:
most_common = max(error_types.items(), key=lambda x: x[1])
self.log_signal.emit(f" Token {token_hash}: {error_count} errors ({most_common[0]}: {most_common[1]})")
else:
self.log_signal.emit(f"[✓] No tokens blacklisted during this run")
# === FEATURE #5: ADAPTIVE SUMMARY ===
self.log_signal.emit(f"\n[📊] ADAPTIVE THROTTLING SUMMARY:")
self.log_signal.emit(f" Final competition level: {self.competition_level}")
self.log_signal.emit(f" Thread count: {self.threads_count} → {self.adaptive_threads}")
self.log_signal.emit(f" Final delay between bursts: {self.adaptive_delay:.2f}s")
if self.last_100_requests:
total = len(self.last_100_requests)
avg_time = sum(r["elapsed"] for r in self.last_100_requests) / total
self.log_signal.emit(f" Avg response time: {avg_time*1000:.0f}ms")
class SniperGUI(QMainWindow):
"""Main GUI application"""
def __init__(self):
super().__init__()
self.workers = []
self.init_ui()
def init_ui(self):
"""Initialize the user interface"""
self.setWindowTitle("⚡ Minecraft Username Sniper GUI v9 - NUCLEAR")
self.setGeometry(100, 100, 1100, 900)
# Main widget and layout
main_widget = QWidget()
self.setCentralWidget(main_widget)
main_layout = QVBoxLayout(main_widget)
# Modern dark theme stylesheet
self.setStyleSheet("""
QMainWindow {
background-color: #1a1a2e;
}
QWidget {
background-color: #1a1a2e;
color: #eaeaea;
font-family: 'Segoe UI', 'Arial', sans-serif;
font-size: 11px;
}
QLabel {
color: #eaeaea;
}
QLineEdit, QTextEdit, QSpinBox, QComboBox {
background-color: #16213e;
border: 1px solid #0f3460;
border-radius: 6px;
padding: 8px 12px;
color: #eaeaea;
selection-background-color: #e94560;
}
QLineEdit:focus, QTextEdit:focus, QSpinBox:focus, QComboBox:focus {
border: 1px solid #e94560;
}
QGroupBox {
font-weight: bold;
font-size: 11px;
color: #eaeaea;
border: 1px solid #0f3460;
border-radius: 8px;
margin-top: 12px;
padding-top: 12px;
background-color: #16213e;
}
QGroupBox::title {
subcontrol-origin: margin;
left: 12px;
padding: 0 8px;
color: #e94560;
}
QPushButton {
background-color: #0f3460;
color: #eaeaea;
border: none;
border-radius: 6px;
padding: 10px 20px;
font-weight: bold;
min-width: 80px;
}
QPushButton:hover {
background-color: #e94560;
}
QPushButton:pressed {
background-color: #c73e54;
}
QCheckBox {
color: #eaeaea;
spacing: 8px;
}
QCheckBox::indicator {
width: 18px;
height: 18px;
border: 1px solid #0f3460;
border-radius: 4px;
background-color: #16213e;
}
QCheckBox::indicator:checked {
background-color: #e94560;
border: 1px solid #e94560;
image: url(data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHdpZHRoPSIxMiIgaGVpZ2h0PSIxMiIgdmlld0JveD0iMCAwIDEyIDEyIj48cGF0aCBkPSJNMSA2bDMgMyA2LTYiIGZpbGw9Im5vbmUiIHN0cm9rZT0iI2ZmZiIgc3Ryb2tlLXdpZHRoPSIyIiBzdHJva2UtbGluZWNhcD0icm91bmQiIHN0cm9rZS1saW5lam9pbj0icm91bmQiLz48L3N2Zz4=);
}
QSpinBox::up-button, QSpinBox::down-button {
background-color: #0f3460;
width: 24px;
border-radius: 3px;
}
QComboBox::drop-down {
border: none;
width: 30px;
}
QComboBox::down-arrow {
width: 12px;
height: 12px;
}
QTabWidget::pane {
border: 1px solid #0f3460;
background-color: #16213e;
border-radius: 8px;
margin: 4px;
}
QTabBar::tab {
background: #0f3460;
padding: 12px 24px;
margin-right: 4px;
border-radius: 6px 6px 0 0;
font-weight: bold;
color: #a0a0a0;
}
QTabBar::tab:hover {
background: #1a4a7a;
}
QTabBar::tab:selected {
background: #e94560;
color: white;
}
QTextEdit#log_output {
background-color: #0d1117;
color: #58a6ff;
border: 1px solid #0f3460;
border-radius: 6px;
padding: 8px;
font-family: 'Consolas', 'Monaco', monospace;
font-size: 10px;
line-height: 1.4;
}
QProgressBar {
border: 1px solid #0f3460;
border-radius: 6px;
text-align: center;
background-color: #16213e;
}
QProgressBar::chunk {
background-color: #e94560;
border-radius: 5px;
}
QSlider::groove:horizontal {
border: 1px solid #0f3460;
height: 6px;
background-color: #16213e;
border-radius: 3px;
}
QSlider::handle:horizontal {
background-color: #e94560;
border: 2px solid #0f3460;
width: 16px;
margin: -5px 0;
border-radius: 8px;
}
QSlider::handle:horizontal:hover {
background-color: #ff6b6b;
}
""")
# Title with gradient effect
title_layout = QHBoxLayout()
title_layout.setSpacing(10)
title_label = QLabel("⚡ Minecraft Username Sniper")
title_label.setFont(QFont("Arial", 20, QFont.Bold))
title_label.setStyleSheet("color: linear-gradient(90deg, #e94560, #ff6b6b);")
title_layout.addWidget(title_label)
version_label = QLabel("v9.0 - NUCLEAR PACKAGE")
version_label.setFont(QFont("Arial", 12, QFont.Bold))
version_label.setStyleSheet("color: #e94560; padding: 4px 12px; background-color: #16213e; border-radius: 12px;")
title_layout.addWidget(version_label)
title_layout.addStretch()
main_layout.addLayout(title_layout, stretch=0)
# Status bar with modern styling
self.status_label = QLabel("🟢 Ready")
self.status_label.setFont(QFont("Arial", 11, QFont.Bold))
self.status_label.setStyleSheet("""
padding: 10px 16px;
background-color: #16213e;
border-radius: 8px;
border: 1px solid #0f3460;
color: #4CAF50;
min-height: 32px;
""")
main_layout.addWidget(self.status_label, stretch=0)
# Tabs for different sniper sections
tabs = QTabWidget()
main_layout.addWidget(tabs)
# === NAMEMC Tab (Windowed Timing) ===
namemc_tab = self._create_sniper_tab("📅 NAMEMC (Windowed Mode)", timing_mode="windowed")
tabs.addTab(namemc_tab, "NAMEMC")
# === LABYNAMES Tab (Exact Timing) ===
labynames_tab = self._create_sniper_tab("⏰ LABYNAMES (Exact Mode)", timing_mode="exact")
tabs.addTab(labynames_tab, "LABYNAMES")
# === Log Output ===
log_group = QGroupBox("📋 Live Log Output")
log_group.setStyleSheet("""
QGroupBox {
font-weight: bold;
font-size: 12px;
color: #eaeaea;
border: 1px solid #0f3460;
border-radius: 8px;
margin-top: 16px;
padding-top: 12px;
background-color: #16213e;
}
QGroupBox::title {
subcontrol-origin: margin;
left: 12px;
padding: 0 8px;
color: #58a6ff;
font-size: 12px;
}
""")
log_layout = QVBoxLayout()
self.log_output = QTextEdit()
self.log_output.setObjectName("log_output")
self.log_output.setReadOnly(True)
self.log_output.setFont(QFont("Consolas", 10))
self.log_output.setLineWrapMode(QTextEdit.NoWrap)
log_layout.addWidget(self.log_output)
log_buttons = QHBoxLayout()
clear_btn = QPushButton("🗑️ Clear Log")
clear_btn.clicked.connect(lambda: self.log_output.clear())
log_buttons.addWidget(clear_btn)
save_btn = QPushButton("💾 Save Log")
save_btn.clicked.connect(self.save_log)
log_buttons.addWidget(save_btn)
log_buttons.addStretch()
log_layout.addLayout(log_buttons)
log_group.setLayout(log_layout)
main_layout.addWidget(log_group, stretch=1)
def _create_sniper_tab(self, title, timing_mode):
"""Create a sniper configuration tab"""
tab = QWidget()
layout = QVBoxLayout(tab)
layout.setSpacing(16)
# Title label with modern styling
title_label = QLabel(title)
title_label.setFont(QFont("Arial", 14, QFont.Bold))
title_label.setStyleSheet("""
color: #e94560;
padding: 12px;
background-color: #0f3460;
border-radius: 6px;
border-left: 4px solid #e94560;
""")
layout.addWidget(title_label)
# Input fields container
inputs_group = QGroupBox("📝 Target Configuration")
inputs_group.setStyleSheet("""
QGroupBox {
font-weight: bold;
color: #eaeaea;
border: 1px solid #0f3460;
border-radius: 8px;
padding-top: 12px;
background-color: #0f3460;
}
QGroupBox::title {
subcontrol-origin: margin;
left: 12px;
padding: 0 8px;
color: #58a6ff;
}
""")
inputs_layout = QVBoxLayout()
# Username input
username_layout = QHBoxLayout()
username_label = QLabel("🎯 Username to Snipe:")
username_label.setStyleSheet("font-weight: bold; color: #eaeaea; padding-right: 12px;")
username_layout.addWidget(username_label, alignment=Qt.AlignRight)
self.username_input = QLineEdit()
self.username_input.setPlaceholderText("Enter username (e.g., Eternal)")
username_layout.addWidget(self.username_input, stretch=1)
inputs_layout.addLayout(username_layout)
# Drop time input
time_layout = QHBoxLayout()
time_label = QLabel("🕐 Drop Time (UTC):")
time_label.setStyleSheet("font-weight: bold; color: #eaeaea; padding-right: 12px;")
time_layout.addWidget(time_label, alignment=Qt.AlignRight)
self.drop_time_input = QLineEdit()
self.drop_time_input.setPlaceholderText("YYYY-MM-DDTHH:MM:SS.000Z (e.g., 2025-09-11T00:18:47.000Z)")
time_layout.addWidget(self.drop_time_input, stretch=1)
inputs_layout.addLayout(time_layout)
inputs_group.setLayout(inputs_layout)
layout.addWidget(inputs_group)
# Tokens input (multi-line)
tokens_group = QGroupBox("🔑 OAuth Tokens (one per line)")
tokens_group.setStyleSheet("""
QGroupBox {
font-weight: bold;
color: #eaeaea;
border: 1px solid #0f3460;
border-radius: 8px;
padding-top: 12px;
background-color: #0f3460;
}
QGroupBox::title {
subcontrol-origin: margin;
left: 12px;
padding: 0 8px;
color: #58a6ff;
}
""")
tokens_layout = QVBoxLayout()
self.tokens_input = QTextEdit()
self.tokens_input.setPlaceholderText("Paste tokens here, one per line...\nToken1\nToken2\nToken3\n\n(We'll validate them before firing)")
self.tokens_input.setFont(QFont("Consolas", 9))
self.tokens_input.setMaximumHeight(150)
tokens_layout.addWidget(self.tokens_input)
tokens_group.setLayout(tokens_layout)
layout.addWidget(tokens_group)
# === FEATURE #9: AUTO-AUTH SECTION ===
autoauth_group = QGroupBox("🤖 Auto-Auth System (email:password)")
autoauth_group.setStyleSheet("""
QGroupBox {
font-weight: bold;
color: #eaeaea;
border: 1px solid #0f3460;
border-radius: 8px;
padding-top: 12px;
background-color: #0f3460;
}
QGroupBox::title {
subcontrol-origin: margin;
left: 12px;
padding: 0 8px;
color: #9c27b0;
}
""")
autoauth_layout = QVBoxLayout()
# Auto-auth checkbox
self.autoauth_checkbox = QCheckBox("✨ Auto-authenticate from credentials (no manual tokens)")
self.autoauth_checkbox.setChecked(False)
self.autoauth_checkbox.setStyleSheet("color: #9c27b0; font-weight: bold; font-size: 11px;")
self.autoauth_checkbox.setToolTip("Load email:password from file, authenticate automatically 5 mins before each drop")
autoauth_layout.addWidget(self.autoauth_checkbox)
# Account file input
account_file_row = QHBoxLayout()
account_file_label = QLabel("📁 Accounts File:")
account_file_label.setStyleSheet("font-weight: bold; color: #eaeaea;")
account_file_row.addWidget(account_file_label)
self.account_file_input = QLineEdit()
self.account_file_input.setPlaceholderText("accounts.txt (email:password or bearer tokens)")
self.account_file_input.setText("accounts.txt")
self.account_file_input.setStyleSheet("background-color: #16213e; border: 1px solid #0f3460; border-radius: 4px; padding: 6px; color: #eaeaea;")
account_file_row.addWidget(self.account_file_input, stretch=1)
browse_btn = QPushButton("Browse")
browse_btn.setStyleSheet("background-color: #0f3460; color: #eaeaea; border: none; border-radius: 4px; padding: 6px 12px; font-size: 9px;")
browse_btn.clicked.connect(self._browse_account_file)
account_file_row.addWidget(browse_btn)
autoauth_layout.addLayout(account_file_row)
# Auto-refresh before drops
self.auto_refresh_checkbox = QCheckBox("🔄 Auto-refresh tokens 5min before each drop")
self.auto_refresh_checkbox.setChecked(False)
self.auto_refresh_checkbox.setStyleSheet("color: #9c27b0; font-size: 10px;")
self.auto_refresh_checkbox.setToolTip("Re-authenticate silently before every snipe window")
autoauth_layout.addWidget(self.auto_refresh_checkbox)
# Save tokens after auth
self.save_tokens_checkbox = QCheckBox("💾 Save tokens to file after authentication")
self.save_tokens_checkbox.setChecked(True)
self.save_tokens_checkbox.setStyleSheet("color: #9c27b0; font-size: 10px;")
self.save_tokens_checkbox.setToolTip("Cache authenticated tokens for reuse (skip OAuth next time)")
autoauth_layout.addWidget(self.save_tokens_checkbox)
autoauth_group.setLayout(autoauth_layout)
layout.addWidget(autoauth_group)
# Advanced settings
advanced_group = QGroupBox("⚙️ Advanced Settings")
advanced_group.setStyleSheet("""
QGroupBox {
font-weight: bold;
color: #eaeaea;
border: 1px solid #0f3460;
border-radius: 8px;
padding-top: 12px;
background-color: #0f3460;
}
QGroupBox::title {
subcontrol-origin: margin;
left: 12px;
padding: 0 8px;
color: #58a6ff;
}
""")
advanced_layout = QFormLayout()
# Threads per token
threads_row = QHBoxLayout()
threads_label = QLabel("🧵 Threads per Token:")
threads_label.setStyleSheet("font-weight: bold; color: #eaeaea;")
threads_row.addWidget(threads_label)
self.threads_spin = QSpinBox()
self.threads_spin.setRange(1, 100)
self.threads_spin.setValue(10)
threads_row.addWidget(self.threads_spin, stretch=1)
advanced_layout.addRow(threads_row)
# Priority mode
priority_row = QHBoxLayout()
priority_label = QLabel("📊 Token Priority:")
priority_label.setStyleSheet("font-weight: bold; color: #eaeaea;")
priority_row.addWidget(priority_label)
self.priority_combo = QComboBox()
self.priority_combo.addItems(["Fresh Tokens First", "Original Order"])
priority_row.addWidget(self.priority_combo, stretch=1)
advanced_layout.addRow(priority_row)
# Token Warm-Up System (FEATURE #1)
self.warmup_checkbox = QCheckBox("🔥 Token Warm-Up (pre-warms 30s before drop)")
self.warmup_checkbox.setChecked(True)
self.warmup_checkbox.setStyleSheet("color: #FF5722; font-weight: bold;")
self.warmup_checkbox.setToolTip("Pre-warms tokens 30s before drop to keep cache fresh")
advanced_layout.addRow("", self.warmup_checkbox)
# Warm-up duration
warmup_row = QHBoxLayout()
warmup_label = QLabel("⏱️ Warm-up Duration:")
warmup_label.setStyleSheet("font-weight: bold; color: #eaeaea;")
warmup_row.addWidget(warmup_label)
self.warmup_duration_spin = QSpinBox()
self.warmup_duration_spin.setRange(10, 60)
self.warmup_duration_spin.setValue(30)
self.warmup_duration_spin.setSuffix("s")
warmup_row.addWidget(self.warmup_duration_spin, stretch=1)
advanced_layout.addRow(warmup_row)
# === FEATURE #5: ADAPTIVE THROTTLING CONTROLS ===
self.adaptive_checkbox = QCheckBox("📊 Adaptive Throttling (auto-adjusts based on competition)")
self.adaptive_checkbox.setChecked(True)
self.adaptive_checkbox.setStyleSheet("color: #2196F3; font-weight: bold;")
self.adaptive_checkbox.setToolTip("Dynamically adjust thread count and delays based on competition and error rates")
advanced_layout.addRow("", self.adaptive_checkbox)
# Aggressiveness slider
agg_row = QHBoxLayout()
agg_label = QLabel("⚡ Aggressiveness:")
agg_label.setStyleSheet("font-weight: bold; color: #eaeaea;")
agg_row.addWidget(agg_label)
self.aggressiveness_slider = QSlider(Qt.Horizontal)
self.aggressiveness_slider.setRange(1, 10)
self.aggressiveness_slider.setValue(7) # Default to moderately aggressive
self.aggressiveness_slider.setToolTip("1 = Conservative (slow but safe), 10 = Extreme (maximum speed, higher risk)")
agg_row.addWidget(self.aggressiveness_slider, stretch=1)
self.aggressiveness_label = QLabel("7")
self.aggressiveness_label.setMinimumWidth(35)
self.aggressiveness_label.setAlignment(Qt.AlignCenter)
self.aggressiveness_label.setStyleSheet("""
font-weight: bold;
color: #2196F3;
padding: 4px 8px;
background-color: #0f3460;
border-radius: 4px;
border: 1px solid #2196F3;
""")
agg_row.addWidget(self.aggressiveness_label)
advanced_layout.addRow(agg_row)
# Connect slider to update label
self.aggressiveness_slider.valueChanged.connect(lambda v: self.aggressiveness_label.setText(str(v)))
# === FEATURE #7: FINGERPRINT ROTATION TOGGLE ===
self.fingerprint_checkbox = QCheckBox("🎭 Fingerprint Rotation (evades rate limiting)")
self.fingerprint_checkbox.setChecked(True)
self.fingerprint_checkbox.setStyleSheet("color: #9C27B0; font-weight: bold;")
self.fingerprint_checkbox.setToolTip("Rotate User-Agent per request to avoid fingerprint-based rate limiting")
advanced_layout.addRow("", self.fingerprint_checkbox)
# === FEATURE #8: TOKEN ROTATION TOGGLE ===
self.token_rotation_checkbox = QCheckBox("🔄 Token Rotation on 429 (auto-skips poisoned tokens)")
self.token_rotation_checkbox.setChecked(True)
self.token_rotation_checkbox.setStyleSheet("color: #FF9800; font-weight: bold;")
self.token_rotation_checkbox.setToolTip("Automatically skip rate-limited tokens and rotate to healthy ones")
advanced_layout.addRow("", self.token_rotation_checkbox)
# === FEATURE #13: TEMPORAL JITTER TOGGLE ===
self.temporal_jitter_checkbox = QCheckBox("⚡ Temporal Jitter (±3ms randomization)")
self.temporal_jitter_checkbox.setChecked(True)
self.temporal_jitter_checkbox.setStyleSheet("color: #E91E63; font-weight: bold;")
self.temporal_jitter_checkbox.setToolTip("Add microsecond-level timing randomization to evade detection systems")
advanced_layout.addRow("", self.temporal_jitter_checkbox)
# === FEATURE #15: RESPONSE ANALYSIS CHECKBOX ===
self.response_analysis_checkbox = QCheckBox("📊 Response Analysis & Early Abort (saves resources)")
self.response_analysis_checkbox.setChecked(True)
self.response_analysis_checkbox.setStyleSheet("color: #00BCD4; font-weight: bold;")
self.response_analysis_checkbox.setToolTip("Analyze responses in real-time and abort early if outcome is determined")
advanced_layout.addRow("", self.response_analysis_checkbox)
advanced_group.setLayout(advanced_layout)
layout.addWidget(advanced_group)
# Progress bar with modern styling
progress_group = QGroupBox("📈 Execution Progress")
progress_group.setStyleSheet("""
QGroupBox {
font-weight: bold;
color: #eaeaea;
border: 1px solid #0f3460;
border-radius: 8px;
padding-top: 12px;
background-color: #0f3460;
}
QGroupBox::title {
subcontrol-origin: margin;
left: 12px;
padding: 0 8px;
color: #4CAF50;
}
""")
progress_layout = QVBoxLayout()
self.progress_bar = QProgressBar()
self.progress_bar.setValue(0)
self.progress_bar.setTextVisible(True)
progress_layout.addWidget(self.progress_bar)
progress_group.setLayout(progress_layout)
layout.addWidget(progress_group)
# Start/Stop buttons
buttons_group = QGroupBox("🎮 Execution Controls")
buttons_group.setStyleSheet("""
QGroupBox {
font-weight: bold;
color: #eaeaea;
border: 1px solid #0f3460;
border-radius: 8px;
padding-top: 12px;
background-color: #0f3460;
}
QGroupBox::title {
subcontrol-origin: margin;
left: 12px;
padding: 0 8px;
color: #4CAF50;
}
""")
buttons_layout = QHBoxLayout()
self.start_btn = QPushButton("🚀 START SNIPE")
self.start_btn.setFont(QFont("Arial", 11, QFont.Bold))
self.start_btn.setStyleSheet("""
QPushButton {
background-color: #4CAF50;
color: white;
padding: 14px 28px;
border: none;
border-radius: 8px;
font-weight: bold;
font-size: 11px;
}
QPushButton:hover {
background-color: #66bb6a;
}
QPushButton:pressed {
background-color: #3d8b40;
}
""")
self.start_btn.clicked.connect(lambda: self.start_sniper(timing_mode))
buttons_layout.addWidget(self.start_btn)
self.stop_btn = QPushButton("⏹️ EMERGENCY STOP")
self.stop_btn.setStyleSheet("""
QPushButton {
background-color: #f44336;
color: white;
padding: 14px 28px;
border: none;
border-radius: 8px;
font-weight: bold;
font-size: 11px;
}
QPushButton:hover {
background-color: #ef5350;
}
QPushButton:pressed {
background-color: #c62828;
}
""")
self.stop_btn.clicked.connect(self.stop_sniper)
buttons_layout.addWidget(self.stop_btn)
load_tokens_btn = QPushButton("📂 Load Tokens")
load_tokens_btn.setStyleSheet("""
QPushButton {
background-color: #2196F3;
color: white;
padding: 14px 20px;
border: none;
border-radius: 8px;
font-weight: bold;
}
QPushButton:hover {
background-color: #42a5f5;
}
""")
load_tokens_btn.clicked.connect(self.load_tokens_from_file)
buttons_layout.addWidget(load_tokens_btn)
buttons_group.setLayout(buttons_layout)
layout.addWidget(buttons_group)
# Store timing mode in tab
tab.timing_mode = timing_mode
return tab
def _log(self, message):
"""Add message to log output"""
timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]
# Color code important messages
if "SUCCESS" in message or "🏆" in message:
self.log_output.append(f"[{timestamp}] {message}")
elif "ERROR" in message or "❌" in message:
self.log_output.append(f"[{timestamp}] {message}")
elif "FIRE" in message or "🎯" in message:
self.log_output.append(f"[{timestamp}] {message}")
elif "AUTH" in message:
self.log_output.append(f"[{timestamp}] {message}")
else:
self.log_output.append(f"[{timestamp}] {message}")
self.log_output.verticalScrollBar().setValue(self.log_output.verticalScrollBar().maximum())
def _browse_account_file(self):
"""Open file dialog to select accounts file"""
filepath, _ = QFileDialog.getOpenFileName(self, "Select Accounts File", "", "Text Files (*.txt);;All Files (*)")
if filepath:
self.account_file_input.setText(filepath)
def _authenticate_accounts_if_needed(self, tokens_text: str, log_callback=None) -> list:
"""
Auto-authenticate from email:password if enabled
Returns: list of bearer tokens
"""
log = log_callback or self._log
# Check if auto-auth is enabled
if not self.autoauth_checkbox.isChecked():
return [t.strip() for t in tokens_text.split('\n') if t.strip()]
# Load accounts from file
account_file = self.account_file_input.text().strip()
if not account_file:
log("⚠️ Auto-auth enabled but no account file specified!")
QMessageBox.warning(self, "Warning", "Please specify an accounts file for auto-auth!")
return []
log(f"🤖 Auto-auth: Loading accounts from {account_file}")
accounts = load_accounts_from_file(account_file)
if not accounts:
log("❌ No accounts found in file!")
QMessageBox.critical(self, "Error", f"No valid accounts found in {account_file}")
return []
log(f"📊 Found {len(accounts)} account(s) to authenticate")
# Authenticate each account
tokens_dict = authenticate_accounts(accounts, log_callback=log)
if not tokens_dict:
log("❌ All authentications failed!")
QMessageBox.critical(self, "Error", "All account authentications failed!")
return []
# Save tokens if requested
if self.save_tokens_checkbox.isChecked():
import os
tokens_file = account_file.rsplit('.', 1)[0] + '_tokens.txt'
save_tokens_to_file(tokens_dict, tokens_file)
log(f"💾 Saved {len(tokens_dict)} authenticated token(s) to {tokens_file}")
# Return token list
return list(tokens_dict.values())
def start_sniper(self, timing_mode):
"""Start the sniper with current settings"""
# Get current tab's inputs
current_tab = self.centralWidget().findChild(QTabWidget()).currentWidget()
username = self.username_input.text().strip()
drop_time = self.drop_time_input.text().strip()
tokens_text = self.tokens_input.toPlainText()
threads_count = self.threads_spin.value()
priority_mode = "fresh_first" if self.priority_combo.currentText() == "Fresh Tokens First" else "original"
# Validate inputs
if not username:
QMessageBox.warning(self, "Error", "Please enter a username to snipe!")
return
if not drop_time:
QMessageBox.warning(self, "Error", "Please enter a drop time (UTC)!")
return
# === FEATURE #9: AUTO-AUTH ===
if self.autoauth_checkbox.isChecked():
# Authenticate from credentials
tokens = self._authenticate_accounts_if_needed(tokens_text, log_callback=self._log)
else:
# Use manual tokens
tokens = [t.strip() for t in tokens_text.split('\n') if t.strip()]
if not tokens:
QMessageBox.warning(self, "Error", "No valid tokens available!")
return
# Stop any existing workers
self.stop_sniper()
# Update status
self.status_label.setText("🔥 SNIPE ACTIVE - GOOD LUCK!")
self.status_label.setStyleSheet("padding: 5px; background-color: #4CAF50; color: white; border-radius: 3px;")
# Create and start worker
self._log(f"🚀 Starting {timing_mode} sniper...")
# Get warm-up settings
warmup_enabled = self.warmup_checkbox.isChecked()
warmup_duration = self.warmup_duration_spin.value()
# === FEATURE #5: Get adaptive throttling settings ===
adaptive_enabled = self.adaptive_checkbox.isChecked()
aggressiveness = self.aggressiveness_slider.value()
# === FEATURE #6: Get fingerprint rotation settings ===
fingerprint_rotation = self.fingerprint_checkbox.isChecked()
# === FEATURE #8: Get token rotation settings ===
token_rotation = self.token_rotation_checkbox.isChecked()
# === FEATURE #13: Get temporal jitter settings ===
temporal_jitter = self.temporal_jitter_checkbox.isChecked()
# === FEATURE #15: Get response analysis settings ===
response_analysis = self.response_analysis_checkbox.isChecked()
worker = SniperWorker(username, tokens, drop_time, threads_count, timing_mode, priority_mode, warmup_enabled, warmup_duration, adaptive_enabled, aggressiveness, fingerprint_rotation, token_rotation, temporal_jitter, response_analysis)
worker.log_signal.connect(self._log)
worker.progress_signal.connect(self.progress_bar.setValue)
worker.finished.connect(lambda: self.on_sniper_finished())
worker.start()
self.workers.append(worker)
self._log(f"✅ Sniper initialized! Username: {username}, Tokens: {len(tokens)}, Threads: {threads_count}")
self._log(f"⏱️ Waiting for drop time: {drop_time}")
if warmup_enabled:
self._log(f"🔥 Token warm-up: ENABLED ({warmup_duration}s before drop)")
else:
self._log(f"⚠️ Token warm-up: DISABLED")
if adaptive_enabled:
self._log(f"📊 Adaptive throttling: ENABLED (Aggressiveness: {aggressiveness}/10)")
else:
self._log(f"⚠️ Adaptive throttling: DISABLED")
if fingerprint_rotation:
self._log(f"🎭 Fingerprint rotation: ENABLED")
else:
self._log(f"⚠️ Fingerprint rotation: DISABLED")
if token_rotation:
self._log(f"🔄 Token rotation on rate limit: ENABLED")
else:
self._log(f"⚠️ Token rotation on rate limit: DISABLED")
if temporal_jitter:
self._log(f"⚡ Temporal jitter: ENABLED (±3ms window)")
else:
self._log(f"⚠️ Temporal jitter: DISABLED")
if response_analysis:
self._log(f"📊 Response analysis & early abort: ENABLED")
else:
self._log(f"⚠️ Response analysis: DISABLED")
def stop_sniper(self):
"""Stop all running snipers"""
self._log("⏹️ Stopping all snipers...")
for worker in self.workers:
worker.stop_event.set()
worker.wait(3000) # Wait up to 3 seconds
self.workers = []
self._log("✅ All snipers stopped")
self.status_label.setText("Stopped")
self.status_label.setStyleSheet("padding: 5px; background-color: #f44336; color: white; border-radius: 3px;")
def on_sniper_finished(self):
"""Called when sniper completes"""
if self.success_event.is_set() if hasattr(self, 'success_event') else False:
self.status_label.setText("🏆 SUCCESS! NAME CLAIMED!")
self.status_label.setStyleSheet("padding: 5px; background-color: #4CAF50; color: white; border-radius: 3px;")
else:
self.status_label.setText("Completed")
self.status_label.setStyleSheet("padding: 5px; background-color: #2196F3; color: white; border-radius: 3px;")
def load_tokens_from_file(self):
"""Load tokens from a text file"""
file_path, _ = QFileDialog.getOpenFileName(self, "Load Tokens", "", "Text Files (*.txt)")
if file_path:
try:
with open(file_path, 'r') as f:
tokens = f.read()
self.tokens_input.setText(tokens)
token_count = len([t for t in tokens.split('\n') if t.strip()])
self._log(f"✅ Loaded {token_count} tokens from {file_path}")
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to load tokens: {e}")
def save_log(self):
"""Save log to file"""
file_path, _ = QFileDialog.getSaveFileName(self, "Save Log", "sniper_log.txt", "Text Files (*.txt)")
if file_path:
try:
with open(file_path, 'w') as f:
f.write(self.log_output.toPlainText())
self._log(f"✅ Log saved to {file_path}")
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to save log: {e}")
def main():
app = QApplication(sys.argv)
# Set application font
font = QFont("Arial", 10)
app.setFont(font)
# Set application-wide stylesheet
app.setStyleSheet("""
QMainWindow {
background-color: #ffffff;
}
QGroupBox {
margin-top: 10px;
border: 1px solid #ddd;
border-radius: 5px;
}
QLabel {
padding: 3px;
}
""")
window = SniperGUI()
window.show()
sys.exit(app.exec_())
if __name__ == "__main__":
main()