#!/usr/bin/env python3 """ Minecraft Username Sniper - Headless CLI Mode (Phase 5 Feature #1) Run the sniper from the command line without PyQt5 GUI. Designed for servers, cron jobs, and remote SSH sessions. Usage: python3 sniper_headless.py -n Steve -d "2026-05-20T21:08:26Z" -t tokens.txt python3 sniper_headless.py -n Steve -d "2026-05-20T21:08:26Z" -a accounts.txt python3 sniper_headless.py -n Steve -d "2026-05-20T21:08:26Z" -t tokens.txt --webhook-url https://discord.com/api/webhooks/... """ import argparse import hashlib import logging import os import random import signal import sys import threading import time from datetime import datetime, timezone import httpx # Import shared modules try: from sniper_webhook import send_snipe_alert, send_auth_alert, send_info_alert except ImportError: send_snipe_alert = send_auth_alert = send_info_alert = None try: from sniper_history import record_snipe except ImportError: record_snipe = None # === Constants from main module === MICROSOFT_CLIENT_ID = "00000000441cc96b" MAX_CONCURRENT_TOKENS = 5 BATCH_DELAY = 0.1 FAST_FAIL_TIMEOUT = 3.0 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 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", ] def get_random_user_agent(): return random.choice(USER_AGENTS) # --- Microsoft Auth (headless version) --- class MicrosoftAuthHeadless: """Microsoft OAuth device flow for headless use.""" def __init__(self, email, password, logger=None): self.email = email self.password = password self.log = logger or print self.session = httpx.Client( headers={"User-Agent": get_random_user_agent()}, timeout=30.0, ) def authenticate(self): """Run full OAuth flow. Returns bearer token or raises.""" self.log(f"[AUTH] Starting authentication for {self.email}") # Step 1: Microsoft OAuth ms_token = self._get_microsoft_token() if not ms_token: raise Exception("Microsoft OAuth failed") self.log("[AUTH] ✓ Microsoft OAuth") # Step 2: XBL xbl_token = self._get_xbl_token(ms_token) if not xbl_token: raise Exception("XBL token failed") self.log("[AUTH] ✓ XBL token") # Step 3: XSTS xsts_token = self._get_xsts_token(xbl_token) if not xsts_token: raise Exception("XSTS token failed") self.log("[AUTH] ✓ XSTS token") # Step 4: Minecraft mc_token = self._get_minecraft_token(xsts_token) if not mc_token: raise Exception("Minecraft token failed") self.log("[AUTH] ✓✓✓ Minecraft bearer token obtained!") return mc_token def _get_microsoft_token(self): 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", } resp = self.session.post(url, data=data) if resp.status_code != 200: self.log(f"[AUTH] ✗ Microsoft OAuth failed: {resp.status_code}") return "" result = resp.json() device_code = result.get("device_code") user_code = result.get("user_code") verification_url = result.get("verification_url") interval = result.get("interval", 5) if not device_code: self.log("[AUTH] ✗ No device code received") return "" 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...\n") self.log(f"{'='*60}\n") 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, } for _ in range(60): time.sleep(interval) resp = self.session.post(poll_url, data=poll_data) if resp.status_code == 200: return resp.json().get("access_token", "") elif resp.status_code == 400: error = resp.json().get("error", "") if error != "authorization_pending": self.log(f"[AUTH] ✗ {error}") return "" self.log("[AUTH] ✗ Authentication timeout") return "" def _get_xbl_token(self, ms_token): url = "https://user.auth.xboxlive.com/user/authenticate" resp = self.session.post(url, headers={ "Authorization": f"Bearer {ms_token}", "Content-Type": "application/json", "X-Xbl-Contract": "1", }, json={ "Properties": {"AuthMethod": "RPS", "SiteName": "user.auth.xboxlive.com", "RpsTicket": f"d={ms_token}"}, "RelyingParty": "http://auth.xboxlive.com", "TokenType": "JWT", }) return resp.json() if resp.status_code == 200 else {} def _get_xsts_token(self, xbl_data): xbl_token = xbl_data.get("Token", "") user_hwid = xbl_data.get("DisplayClaims", {}).get("xui", [{}])[0].get("userHWId", "") resp = self.session.post("https://xsts.auth.xboxlive.com/xsts/authorize", headers={ "Content-Type": "application/json", "X-Xbl-Contract": "1", }, json={ "Properties": {"SandboxId": "RETAIL", "UserTokens": [xbl_token], "UserTileId": user_hwid}, "RelyingParty": "rp://api.minecraftservices.com/", "TokenType": "JWT", }) return resp.json() if resp.status_code == 200 else {} def _get_minecraft_token(self, xsts_data): xsts_token = xsts_data.get("Token", "") resp = self.session.post("https://api.minecraftservices.com/authentication/login_with_xbox", headers={ "Content-Type": "application/json", }, json={"identity": {"issuer": "xp", "token": xsts_token}}) return resp.json().get("access_token", "") if resp.status_code == 200 else "" # --- Token loading --- def load_tokens_from_file(path): """Load bearer tokens from file (one per line).""" tokens = [] with open(path) as f: for line in f: line = line.strip() if line and not line.startswith("#"): tokens.append(line) return tokens def authenticate_accounts_from_file(path, logger=None): """Load email:password pairs and authenticate each.""" accounts = [] with open(path) as f: for line in f: line = line.strip() if line and not line.startswith("#"): if ":" in line and "@" in line: email, password = line.split(":", 1) accounts.append((email.strip(), password.strip())) elif line.startswith("eyJ"): # Raw token pass # Would need separate handling tokens = [] for email, password in accounts: try: auth = MicrosoftAuthHeadless(email, password, logger) token = auth.authenticate() if token: tokens.append(token) if send_auth_alert: send_auth_alert(None, email, True) # webhook set later except Exception as e: logger(f"[AUTH] ✗ Failed for {email}: {e}") return tokens # --- Connection pool --- connection_pool = httpx.HTTPTransport(retries=1) # --- Headless Sniper Worker --- class HeadlessSniper: """Non-GUI sniper that uses plain logging.""" def __init__(self, name, tokens, drop_time_utc, threads=10, timing_mode="exact", logger=None, webhook_url=None): self.name = name self.tokens = tokens self.drop_time_utc = drop_time_utc self.threads_count = threads self.timing_mode = timing_mode self.log = logger or print self.webhook_url = webhook_url self.stop_event = threading.Event() self.success_event = threading.Event() self.requests_fired = 0 self.requests_success = 0 self.requests_failed = 0 self.success_response_time_ms = None self.success_token_hash = None self.success_error = None def run(self): """Main entry point.""" self.log(f"[*] Headless sniper starting for: {self.name}") self.log(f"[*] Drop time: {self.drop_time_utc} UTC") self.log(f"[*] Mode: {self.timing_mode}, Tokens: {len(self.tokens)}, Threads: {self.threads_count}") target = datetime.fromisoformat(self.drop_time_utc.replace("Z", "+00:00")) target_ts = target.timestamp() self.log(f"[*] Waiting for drop time...") self.progress(target_ts) if self.stop_event.is_set(): self.log("[!] Stopped by user") return False self.log(f"[!] 🎯🎯🎯 DROP TIME! FIRING NOW! 🎯🎯🎯") # Nuclear burst first self.log(f"[!] ⚛️ NUCLEAR BURST!") self._fire_burst(nuclear=True) if self.success_event.is_set(): self._on_success() return True self.log(f"[*] Switching to controlled bursts (every 2-5s)...") burst_count = 0 while not self.stop_event.is_set(): if self.success_event.is_set(): break burst_count += 1 wait = random.uniform(2.0, 5.0) self.log(f"[*] ⏱️ Next burst in {wait:.1f}s... (Ctrl+C to stop)") for _ in range(int(wait * 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(f"[!] 🔥 Burst #{burst_count}") self._fire_burst() if self.success_event.is_set(): self._on_success() return True self.log("[!] Sniper stopped without success") return False def progress(self, target_ts): """Wait until drop time with countdown.""" while not self.stop_event.is_set(): now = time.time() delta = target_ts - now if delta <= 0: break mins, secs = divmod(max(0, delta), 60) self.log(f"\r[*] Countdown: {int(mins):02d}:{int(secs):02d} (Ctrl+C to stop)", end="", flush=True) sleep_time = min(delta, 1.0) time.sleep(sleep_time) self.log("") # newline def _fire_burst(self, nuclear=False): """Fire all tokens in a burst.""" tokens = self.tokens.copy()[::-1] # fresh first num_batches = -(-len(tokens) // MAX_CONCURRENT_TOKENS) if tokens else 0 self.log(f"[*] Firing {len(tokens)*self.threads_count} requests in {num_batches} batch(es)") threads_list = [] for batch_idx in range(num_batches): if self.success_event.is_set() or self.stop_event.is_set(): break start = batch_idx * MAX_CONCURRENT_TOKENS end = min(start + MAX_CONCURRENT_TOKENS, len(tokens)) batch = tokens[start:end] for token in batch: if self.success_event.is_set() or self.stop_event.is_set(): break for _ in range(self.threads_count): if self.success_event.is_set(): break t = threading.Thread(target=self._send_request, args=(token,), daemon=True) t.start() threads_list.append(t) if batch_idx < num_batches - 1: time.sleep(BATCH_DELAY) for t in threads_list: t.join(timeout=5.0) def _send_request(self, token): """Send a single name change request.""" url = f"https://api.minecraftservices.com/minecraft/profile/name/{self.name}" max_retries = 3 for attempt in range(max_retries): if self.success_event.is_set() or self.stop_event.is_set(): return client = httpx.Client( transport=httpx.HTTPTransport(pool=connection_pool), http2=True, timeout=httpx.Timeout(FAST_FAIL_TIMEOUT, connect=FAST_FAIL_TIMEOUT), headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", "User-Agent": get_random_user_agent(), }, ) try: start = time.time() r = client.put(url, json={}) elapsed_ms = (time.time() - start) * 1000 self.requests_fired += 1 if r.status_code in (200, 201, 204): self.requests_success += 1 self.success_response_time_ms = elapsed_ms self.success_token_hash = hashlib.md5(token.encode()).hexdigest()[:8] self.success_event.set() self.log(f"\n[🏆] SUCCESS! {r.status_code} ({elapsed_ms:.1f}ms)") return self.requests_failed += 1 try: body = r.json() error_msg = body.get("errorMessage", str(body)) except: error_msg = r.text[:100] if r.status_code == 429: retry_after = int(r.headers.get("Retry-After", 30)) self.log(f"[⚠️] Rate limited (429), retry after {retry_after}s") return elif r.status_code == 409: self.log(f"[⚠️] Name taken (409)") return else: if attempt < max_retries - 1: time.sleep(0.1 * (attempt + 1)) except Exception as e: self.requests_failed += 1 if attempt == max_retries - 1: self.log(f"[!] Request error: {e}") def _on_success(self): """Handle success: alerts, logging, database.""" self.log(f"\n{'='*60}") self.log(f"[🏆] NAME CLAIMED: {self.name}") self.log(f"[🏆] Response time: {self.success_response_time_ms:.1f}ms") self.log(f"[🏆] Total requests: {self.requests_fired}") self.log(f"{'='*60}") # Discord webhook if self.webhook_url and send_snipe_alert: send_snipe_alert( self.webhook_url, self.name, True, account_hash=self.success_token_hash, response_time_ms=self.success_response_time_ms, drop_time=self.drop_time_utc, ) self.log("[WEBHOOK] Discord alert sent") # Database if record_snipe: record_snipe( self.name, True, account_hash=self.success_token_hash, response_time_ms=self.success_response_time_ms, drop_time_utc=self.drop_time_utc, ) self.log("[DB] Result recorded to history") def stop(self): self.stop_event.set() # --- Parallel sniper for multiple names --- class HeadlessParallelSniper: """Snipe multiple names simultaneously.""" def __init__(self, names_tokens, threads=10, logger=None, webhook_url=None): """ Args: names_tokens: List of {"name": str, "drop_time": str, "tokens": list} threads: Threads per name logger: Log function webhook_url: Discord webhook URL """ self.jobs = names_tokens self.threads = threads self.log = logger or print self.webhook_url = webhook_url self.stop_event = threading.Event() self.results = {} def run(self): """Run all snipes in parallel.""" self.log(f"[*] Parallel sniper: {len(self.jobs)} target(s)") snipers = [] threads = [] for job in self.jobs: sniper = HeadlessSniper( name=job["name"], tokens=job.get("tokens", []), drop_time_utc=job["drop_time"], threads=self.threads, logger=self.log, webhook_url=self.webhook_url, ) sniper.stop_event = self.stop_event # shared stop snipers.append(sniper) t = threading.Thread(target=self._run_sniper, args=(sniper, job["name"])) t.start() threads.append(t) # Wait for all for t in threads: t.join() # Summary self.log(f"\n{'='*60}") self.log("[*] PARALLEL SNIPE COMPLETE") self.log(f"{'='*60}") for sniper in snipers: status = "✅ SUCCESS" if sniper.success_event.is_set() else "❌ FAILED" self.log(f" {status} {sniper.name}") return all(s.success_event.is_set() for s in snipers) def _run_sniper(self, sniper, name): try: sniper.run() except Exception as e: self.log(f"[ERROR] Sniper for {name} crashed: {e}") def stop(self): self.stop_event.set() # --- CLI --- def parse_args(): parser = argparse.ArgumentParser( description="⚡ Minecraft Username Sniper - Headless Mode", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Basic snipe with token file python3 sniper_headless.py -n Steve -d "2026-05-20T21:08:26Z" -t tokens.txt # Auto-auth with credentials python3 sniper_headless.py -n Steve -d "2026-05-20T21:08:26Z" -a accounts.txt # With Discord webhook alerts python3 sniper_headless.py -n Steve -d "2026-05-20T21:08:26Z" -t tokens.txt --webhook-url https://discord.com/api/webhooks/... # Parallel snipe (multiple names, same drop time) python3 sniper_headless.py -n Steve -n Alex -d "2026-05-20T21:08:26Z" -t tokens.txt --parallel # Windowed mode python3 sniper_headless.py -n Steve -d "2026-05-20T21:08:26Z" -t tokens.txt --timing-mode windowed """, ) parser.add_argument("-n", "--name", action="append", required=True, help="Target username(s). Can specify multiple for parallel mode.") parser.add_argument("-d", "--drop-time", required=True, help="Drop time in UTC ISO format (e.g. 2026-05-20T21:08:26Z)") parser.add_argument("-t", "--tokens-file", help="File with bearer tokens (one per line)") parser.add_argument("-a", "--accounts-file", help="File with email:password credentials for auto-auth") parser.add_argument("-c", "--threads", type=int, default=10, help="Threads per name (default: 10)") parser.add_argument("--timing-mode", choices=["exact", "windowed"], default="exact", help="Timing mode (default: exact)") parser.add_argument("--webhook-url", help="Discord webhook URL for alerts") parser.add_argument("--log-file", help="Log file path (default: sniper.log)") parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output") parser.add_argument("--parallel", action="store_true", help="Snipe multiple names simultaneously") parser.add_argument("--max-attempts", type=int, default=0, help="Max burst attempts (0 = unlimited)") return parser.parse_args() def main(): args = parse_args() # Setup logging log_kwargs = {"level": logging.DEBUG if args.verbose else logging.INFO} if args.log_file: log_kwargs["filename"] = args.log_file log_kwargs["filemode"] = "a" logger = logging.getLogger("sniper") logger.setLevel(logging.DEBUG) # Console handler ch = logging.StreamHandler() ch.setLevel(logging.DEBUG if args.verbose else logging.INFO) ch.setFormatter(logging.Formatter("[%(asctime)s] %(message)s", datefmt="%H:%M:%S")) logger.addHandler(ch) if args.log_file: fh = logging.FileHandler(args.log_file, mode="a") fh.setLevel(logging.DEBUG) fh.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) logger.addHandler(fh) def log(msg, **kwargs): logger.info(msg) # Signal handling snipers_to_stop = [] def signal_handler(sig, frame): log("[!] Received stop signal, stopping...") for s in snipers_to_stop: if hasattr(s, 'stop'): s.stop() sys.exit(130) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) # Load tokens tokens = [] if args.tokens_file: tokens = load_tokens_from_file(args.tokens_file) log(f"[*] Loaded {len(tokens)} tokens from {args.tokens_file}") elif args.accounts_file: log(f"[*] Authenticating from {args.accounts_file}...") tokens = authenticate_accounts_from_file(args.accounts_file, log) log(f"[*] Got {len(tokens)} tokens from {len(tokens)} accounts") else: log("[ERROR] No tokens provided. Use -t (tokens file) or -a (accounts file)") sys.exit(1) if not tokens: log("[ERROR] No valid tokens loaded") sys.exit(1) # Run if len(args.name) > 1 and args.parallel: log(f"[*] Parallel mode: {len(args.name)} names") jobs = [{"name": n, "drop_time": args.drop_time, "tokens": tokens} for n in args.name] sniper = HeadlessParallelSniper(jobs, threads=args.threads, logger=log, webhook_url=args.webhook_url) snipers_to_stop.append(sniper) success = sniper.run() else: for target_name in args.name: log(f"\n{'='*60}") log(f"[*] Target: {target_name}") log(f"{'='*60}") sniper = HeadlessSniper( name=target_name, tokens=tokens, drop_time_utc=args.drop_time, threads=args.threads, timing_mode=args.timing_mode, logger=log, webhook_url=args.webhook_url, ) snipers_to_stop.append(sniper) success = sniper.run() if not success: # Log failure to DB if record_snipe: record_snipe( target_name, False, error_message="Stopped without success", drop_time_utc=args.drop_time, ) if args.webhook_url and send_snipe_alert: send_snipe_alert( args.webhook_url, target_name, False, error="Stopped without success", drop_time=args.drop_time, ) if success: log("\n[✓] Exit: SUCCESS") sys.exit(0) else: log("\n[✗] Exit: FAILED") sys.exit(1) if __name__ == "__main__": main()