#!/usr/bin/env python3 """Harmless VM integration check. Run ONLY on the dedicated collector after containment is ready.""" import argparse import hashlib import http.client import pathlib import secrets import tempfile import threading import time from evidence import Evidence from honeypot import Engine, Server from vm_backend import VMBackend def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--vm-config', required=True) args = parser.parse_args() marker = pathlib.Path('/etc/decoy-collector') if not marker.is_file() or marker.is_symlink(): raise SystemExit('Refusing live test outside dedicated collector') token = secrets.token_hex(8) with tempfile.TemporaryDirectory(prefix='decoy-live-') as folder: evidence = Evidence(folder, quota=512 * 1024 * 1024) rows = [] def sink(event, **data): evidence.emit(event, **data) rows.append(dict(event=event, **data)) engine = Engine(sink, delay=0, save=evidence.save) backend = VMBackend(args.vm_config, engine.emit) engine.backend = backend # Test-only HTTP traffic must not overwrite the production reverse-peer authorization file. backend.register_peer = lambda peer: None server = Server(('127.0.0.1', 0), engine) backend.on_failure = server.stop_capture thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() def payload(row): return (pathlib.Path(folder) / row['evidence']['artifact']).read_bytes() def run(command): request_id = secrets.token_hex(16) ctx = dict(request_id=request_id, peer='192.0.2.42', peer_port=12345) with engine.lock: state = engine.session(ctx['peer']) ctx['session'] = state.sid state.pending.append((0, command, ctx)) engine.tick() deadline = time.monotonic() + 20 while time.monotonic() < deadline: result = [r for r in rows if r.get('request_id') == request_id] if any(r['event'] in ('execution_error', 'guest_execution_limit') for r in result): raise AssertionError('guest execution failed: ' + request_id) if any(r['event'] == 'execution_end' for r in result): assert all(r.get('peer') == '192.0.2.42' and r.get('session') == state.sid for r in result) assert all(r.get('execution_id') for r in result) return b''.join(payload(r) for r in result if r['event'] == 'guest_stdout'), result time.sleep(.05) raise AssertionError('live command did not complete within20s') def request(method, path, body=None, headers=None): conn = http.client.HTTPConnection('127.0.0.1', server.server_port, timeout=25) try: conn.request(method, path, body, headers or {}) response = conn.getresponse() return response.status, response.read() finally: conn.close() try: path = '/netscaler/ns_gui/live-' + token + '.txt' output, result = run('id > ' + path + '; sleep 1; cat ' + path) assert b'uid=0(root)' in output captured = [r for r in result if r['event'] == 'guest_file_version' and r.get('path') == path] assert captured and b'uid=0(root)' in payload(captured[-1]) ctx = dict(request_id='read-' + token, peer='192.0.2.42', session='live-test') assert b'uid=0(root)' in backend.read(path, ctx) status, body = request('GET', '/live-' + token + '.txt') assert status == 200 and b'uid=0(root)' in body # Harmless User-Agent log-poisoning shape from the public write-up; loopback decoy only. ua_path = '/netscaler/ns_gui/ua-' + token + '.txt' ua = 'pitboss PPE unexpectedly died NSPPE;:`id>' + ua_path + '`;# X' assert request('GET', '/', headers={'User-Agent': ua})[0] == 200 candidate = next(r for r in reversed(rows) if r['event'] == 'injection_candidate') assert candidate['source'] == 'header:User-Agent' and candidate['value'] == ua assert candidate['peer'] == '127.0.0.1' and candidate['peer_port'] > 0 assert request('GET', '/ua-' + token + '.txt')[0] == 404 engine.tick() deadline = time.monotonic() + 20 while time.monotonic() < deadline: matching = [r for r in rows if r.get('request_id') == candidate['request_id']] if any(r['event'] == 'execution_error' for r in matching): raise AssertionError('User-Agent guest execution failed') if any(r['event'] == 'execution_end' for r in matching): break time.sleep(.05) else: raise AssertionError('User-Agent execution timeout') start = next(r for r in matching if r['event'] == 'execution_start') assert start['command'] == 'id>' + ua_path assert all(start[k] == candidate[k] for k in ('request_id', 'session', 'peer', 'peer_port')) status, body = request('GET', '/ua-' + token + '.txt') assert status == 200 and b'uid=0(root)' in body binary = b'\x00\xffinert-live-upload-' + token.encode() digest = hashlib.sha256(binary).hexdigest() multipart = (b'--boundary\r\nContent-Disposition: form-data; name="file"; filename="../../escape.bin"\r\n' b'Content-Type: application/octet-stream\r\n\r\n' + binary + b'\r\n--boundary--\r\n') assert request('POST', '/upload', multipart, {'Content-Type': 'multipart/form-data; boundary=boundary'})[0] == 404 upload = next(r for r in rows if r['event'] == 'upload') assert upload['filename'] == '../../escape.bin' and payload(upload) == binary materialized = next(r for r in rows if r['event'] == 'guest_upload') assert materialized['path'] == '/var/tmp/uploads/' + digest + '.bin' assert materialized['request_id'] == upload['request_id'] and materialized['peer'] == '127.0.0.1' output, _ = run('sha256sum /var/tmp/uploads/' + digest + '.bin') assert output.startswith(digest.encode()) run('ln -s /etc/passwd /netscaler/ns_gui/link-' + token) assert request('GET', '/link-' + token)[0] == 404 endpoints = [('192.0.2.10', 22), ('198.51.100.10', 22), ('169.254.169.254', 80), ('10.0.2.2', 22), ('203.0.113.1', 25)] # Bypass guest NAT via relay UID: prove QEMU restrict blocks direct egress even if guest root disables interception. commands = [] for host, port in endpoints: endpoint = host + ':' + str(port) commands.append("if runuser -u decoy-relay -- timeout 2 bash -c 'echo probe >/dev/tcp/" + host + '/' + str(port) + "' 2>/dev/null; then echo UNEXPECTED_ALLOWED " + endpoint + '; else echo BLOCKED ' + endpoint + '; fi') output, _ = run('; '.join(commands)) assert b'UNEXPECTED_ALLOWED' not in output assert all(('BLOCKED ' + h + ':' + str(p)).encode() in output for h, p in endpoints) run('rm -f ' + path + ' ' + ua_path + ' /netscaler/ns_gui/link-' + token + ' /var/tmp/uploads/' + digest + '.bin') print('PASS: real root command, offguest file/stdout evidence, HTTP file, User-Agent trigger, binary multipart, source correlation, symlink refusal, five denied egress probes') finally: backend.close() server.shutdown() server.server_close() thread.join(timeout=3) evidence.close() if __name__ == '__main__': main()