108 lines
5.1 KiB
Python
108 lines
5.1 KiB
Python
"""Run locally: fake SSH only; never runs received commands or guest.main."""
|
|
import base64
|
|
import json
|
|
import pathlib
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from evidence import EvidenceError
|
|
from honeypot import Engine
|
|
from vm_backend import VMBackend, LIMIT
|
|
|
|
|
|
def main():
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = pathlib.Path(directory)
|
|
fake = root / 'transport.py'
|
|
fake.write_text('''import sys,json,base64
|
|
r=json.loads(sys.stdin.readline())
|
|
if r.get('command') == 'invalid-event':
|
|
print(json.dumps({'event':'response','base64':base64.b64encode(b'bad').decode()}))
|
|
else:
|
|
print(json.dumps({'event':'guest_stdout', 'peer':'forged', 'request_id':'forged',
|
|
'telemetry_trust':'trusted', 'evidence':{'path':'/forged'},
|
|
'base64':r.get('base64',base64.b64encode(b'fixture').decode())}))
|
|
''')
|
|
config = root / 'config.json'
|
|
config.write_text(json.dumps(dict(host='192.0.2.1', vmid=223, identity='/unused', known_hosts='/unused',
|
|
allowed_peers=str(root / 'peers.json'), stop_file=str(root / 'STOP'))))
|
|
events = []
|
|
engine = Engine(lambda event, **data: events.append(dict(event=event, **data)), delay=0)
|
|
backend = VMBackend(config, lambda event, **data: events.append(dict(event=event, **data)))
|
|
# Only test fixtures run locally. Production constructs SSH argv with a fixed remote command.
|
|
assert backend.ssh[-1] == 'python3 /opt/decoy/guest.py' and backend.ssh[0] == 'ssh'
|
|
assert 'StrictHostKeyChecking=yes' in backend.ssh
|
|
backend.ssh = [sys.executable, str(fake)]
|
|
ctx = dict(peer='198.51.100.4', request_id='request-one', session='session-one')
|
|
payload = b'\x00\xff' * (1024 * 1024) # proves upload beyond old 131k request limit
|
|
backend.upload(payload, ctx)
|
|
row = events[-1]
|
|
assert row['content'] == payload and row['peer'] == ctx['peer']
|
|
assert row['request_id'] == ctx['request_id'] and row['telemetry_trust'] == 'guest-reported'
|
|
assert 'evidence' not in row and row['guest_shared']
|
|
assert backend.read('/netscaler/ns_gui/marker', ctx) == b'fixture'
|
|
try:
|
|
backend.upload(b'x' * (LIMIT + 1), ctx)
|
|
raise AssertionError('upload cap')
|
|
except ValueError:
|
|
pass
|
|
try:
|
|
list(backend.records(dict(action='execute', command='invalid-event'), ctx))
|
|
raise AssertionError('guest event whitelist')
|
|
except ValueError:
|
|
pass
|
|
for i in range(130):
|
|
backend.register_peer('198.51.100.' + str(i))
|
|
peers = json.loads((root / 'peers.json').read_text())
|
|
assert len(peers) == 128 and peers[-1] == '198.51.100.129'
|
|
assert stat.S_IMODE((root / 'peers.json').stat().st_mode) == 0o600
|
|
backend.register_peer('::1')
|
|
assert json.loads((root / 'peers.json').read_text()) == peers
|
|
# Mode dispatch with an inert transport: source/request context survives the delayed queue.
|
|
engine.backend = backend
|
|
state = engine.session(ctx['peer'])
|
|
marker = root / 'must-not-exist'
|
|
state.pending.append((0, 'echo fixture > ' + str(marker), ctx))
|
|
engine.tick()
|
|
deadline = time.monotonic() + 3
|
|
while backend.threads and time.monotonic() < deadline:
|
|
time.sleep(.01)
|
|
assert not backend.threads and not marker.exists()
|
|
assert any(e['event'] == 'execution_start' and e['request_id'] == 'request-one' for e in events)
|
|
assert not any(e['event'] == 'simulated_command' for e in events)
|
|
# A full evidence store latches STOP, calls server failure hook, and disables new transport.
|
|
failures = []
|
|
def fail_emit(*args, **kwargs):
|
|
raise EvidenceError('test quota')
|
|
backend.emit = fail_emit
|
|
backend.on_failure = failures.append
|
|
try:
|
|
backend.read('/netscaler/ns_gui/marker', ctx)
|
|
raise AssertionError('evidence failure swallowed')
|
|
except EvidenceError:
|
|
pass
|
|
assert backend.closed and (root / 'STOP').exists() and failures
|
|
assert not backend.active
|
|
# Losing execution telemetry must latch STOP even when the evidence sink still works.
|
|
config_data = json.loads(config.read_text())
|
|
config_data['stop_file'] = str(root / 'STOP-transport')
|
|
config.write_text(json.dumps(config_data))
|
|
broken = VMBackend(config, lambda event, **data: events.append(dict(event=event, **data)))
|
|
broken.ssh = [sys.executable, str(fake)]
|
|
broken.on_failure = failures.append
|
|
broken.execute('invalid-event', ctx)
|
|
deadline = time.monotonic() + 3
|
|
while broken.threads and time.monotonic() < deadline:
|
|
time.sleep(.01)
|
|
assert broken.closed and (root / 'STOP-transport').exists()
|
|
assert any(e['event'] == 'execution_error' for e in events)
|
|
# Guest entrypoint is inert on non-guest systems (no import side effects).
|
|
import guest
|
|
assert guest.ACTIVE is None
|
|
print('PASS: bounded fake SSH, large upload, guest sanitization, mode dispatch, correlation, allowlist, fail-closed')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|