"""Bounded SSH transport to a disposable VM; attacker input never reaches a local shell.""" import base64 import collections import ipaddress import json import os import pathlib import secrets import subprocess import tempfile import threading import time from evidence import EvidenceError LIMIT = 16 * 1024 * 1024 EVENTS = {'guest_http_file', 'guest_upload', 'guest_file_deleted', 'guest_file_version', 'guest_file_capture_error', 'guest_process_start', 'guest_stdout', 'guest_syscalls_stderr', 'guest_execution_limit', 'guest_process_end', 'guest_limit'} FIELDS = {'guest_time_ns', 'path', 'size', 'mode', 'truncated', 'error', 'bytes', 'pid', 'command', 'seconds', 'returncode', 'resource'} class VMBackend: def __init__(self, config, emit): self.config = json.loads(pathlib.Path(config).read_text()) self.emit, self.on_failure = emit, None self.slots = threading.BoundedSemaphore(4) self.io_slots = threading.BoundedSemaphore(4) self.lock = threading.Lock() self.active, self.threads, self.peers = set(), set(), collections.OrderedDict() self.closed = False host = str(ipaddress.ip_address(self.config['host'])) port = int(self.config.get('port', 22)) if not 1 <= port <= 65535: raise ValueError('invalid SSH port') self.peer_path = pathlib.Path(self.config.get('allowed_peers', '/var/lib/decoy/allowed-peers.json')) self.stop_path = pathlib.Path(self.config.get('stop_file', '/var/lib/decoy/STOP')) self.ssh = ['ssh', '-T', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=5', '-o', 'ServerAliveInterval=10', '-o', 'ServerAliveCountMax=2', '-o', 'StrictHostKeyChecking=yes', '-o', 'IdentitiesOnly=yes', '-o', 'UserKnownHostsFile=' + self.config['known_hosts'], '-i', self.config['identity'], '-p', str(port), 'root@' + host, 'python3 /opt/decoy/guest.py'] def close(self, trip=False): if trip: # Broker independently watches this latch; never remove it automatically. self.stop_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) fd = os.open(self.stop_path, os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW, 0o600) os.close(fd) with self.lock: self.closed = True for proc in tuple(self.active): try: proc.kill() except ProcessLookupError: pass def fail(self, error): try: self.close(trip=True) finally: if self.on_failure: self.on_failure(error) def register_peer(self, peer): ip = ipaddress.ip_address(peer) if ip.version != 4: return # broker is IPv4 only try: with self.lock: if self.closed or self.stop_path.exists(): raise EvidenceError('VM capture stopped') now = time.monotonic() self.peers = collections.OrderedDict((k, v) for k, v in self.peers.items() if now - v < 3600) self.peers.pop(str(ip), None) self.peers[str(ip)] = now while len(self.peers) > 128: self.peers.popitem(last=False) self.peer_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) fd, name = tempfile.mkstemp(prefix='.peers-', dir=self.peer_path.parent) try: with os.fdopen(fd, 'w') as f: json.dump(list(self.peers), f) f.flush() os.fsync(f.fileno()) os.replace(name, self.peer_path) finally: if os.path.exists(name): os.unlink(name) except OSError as error: failure = EvidenceError(str(error)) self.fail(failure) raise failure from error def records(self, request, context): encoded = (json.dumps(request, separators=(',', ':')) + '\n').encode() if len(encoded) > 4 * ((LIMIT + 2) // 3) + 131072: raise ValueError('guest request limit') with self.lock: if self.closed or self.stop_path.exists(): raise EvidenceError('VM capture stopped') proc = subprocess.Popen(self.ssh, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) self.active.add(proc) timer = threading.Timer(620 if request['action'] == 'execute' else 20, proc.kill) timer.daemon = True timer.start() try: proc.stdin.write(encoded) proc.stdin.close() total = 0 for count in range(65536): raw = proc.stdout.readline(24 * 1024 * 1024) if not raw: break total += len(raw) if not raw.endswith(b'\n') or total > 128 * 1024 * 1024: raise ValueError('guest telemetry limit') event = json.loads(raw) if not isinstance(event, dict) or event.get('event') not in EVENTS: raise ValueError('invalid guest event') name = event['event'] clean = {k: v for k, v in event.items() if k in FIELDS and (v is None or type(v) in (str, int, bool))} if any(isinstance(v, str) and len(v) > 65536 for v in clean.values()): raise ValueError('guest field limit') if 'base64' in event: content = base64.b64decode(event['base64'], validate=True) if len(content) > LIMIT: raise ValueError('guest artifact limit') clean['content'] = content # The guest cannot supply or override event correlation or storage paths. self.emit(name, **clean, **context, telemetry_trust='guest-reported', guest_shared=True) yield clean else: raise ValueError('guest event count limit') if proc.wait() != 0: raise OSError('guest transport or command failed') except EvidenceError as error: self.fail(error) raise finally: timer.cancel() if proc.poll() is None: proc.kill() proc.wait() proc.stdout.close() try: proc.stdin.close() except OSError: pass with self.lock: self.active.discard(proc) def execute(self, command, context): if not self.slots.acquire(False): self.emit('execution_rejected', **context, reason='VM concurrency limit') return context = dict(context, execution_id=secrets.token_hex(16)) def run(): try: self.emit('execution_start', **context, command=command, vm=self.config['vmid'], guest_shared=True) for _ in self.records(dict(action='execute', command=command), context): pass self.emit('execution_end', **context) except EvidenceError as error: self.fail(error) except Exception as error: try: self.emit('execution_error', **context, error=str(error)) except EvidenceError: pass self.fail(EvidenceError('execution telemetry lost: ' + str(error))) finally: self.slots.release() with self.lock: self.threads.discard(threading.current_thread()) thread = threading.Thread(target=run, daemon=True) with self.lock: self.threads.add(thread) thread.start() def io(self, request, context): if not self.io_slots.acquire(False): self.emit('guest_io_rejected', **context, action=request['action'], reason='concurrency limit') return None try: result = None for event in self.records(request, context): if 'content' in event: result = event['content'] return result except EvidenceError: raise except (OSError, ValueError, TypeError) as error: self.emit('guest_io_error', **context, action=request['action'], error=str(error)) return None finally: self.io_slots.release() def upload(self, payload, context): if len(payload) > LIMIT: raise ValueError('upload limit') return self.io(dict(action='upload', base64=base64.b64encode(payload).decode()), context) def read(self, path, context): return self.io(dict(action='read', path=path), context)