92 lines
4.1 KiB
Python
92 lines
4.1 KiB
Python
"""Run: python3 test_viewer.py. Loopback-only read-only viewer checks; no third-party dependencies."""
|
|
import hashlib
|
|
import http.client
|
|
import http.server
|
|
import json
|
|
import pathlib
|
|
import tempfile
|
|
import threading
|
|
|
|
import viewer
|
|
|
|
|
|
def main():
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
root = pathlib.Path(folder)
|
|
(root / 'artifacts').mkdir()
|
|
payload = b'\x00\xffcommand-output\x00'
|
|
sha = hashlib.sha256(payload).hexdigest()
|
|
(root / 'artifacts' / (sha + '.bin')).write_bytes(payload)
|
|
with (root / 'events.jsonl').open('w') as stream:
|
|
for event, peer, has_evidence in (('connection', '8.8.8.8', False),
|
|
('request', '1.1.1.1', False),
|
|
('simulated_command', '8.8.8.8', True)):
|
|
row = {'event': event, 'timestamp': '2026-09-28T00:00:00.000000+00:00', 'peer': peer,
|
|
'command': 'id > /netscaler/ns_gui/x'}
|
|
if has_evidence:
|
|
row['evidence'] = {'sha256': sha, 'bytes': len(payload)}
|
|
stream.write(json.dumps(row) + '\n')
|
|
stream.write('{not json\n')
|
|
server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), viewer.Handler)
|
|
server.journal = root / 'events.jsonl'
|
|
server.artifacts = root / 'artifacts'
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
|
|
def get(path):
|
|
connection = http.client.HTTPConnection('127.0.0.1', server.server_port, timeout=3)
|
|
connection.request('GET', path)
|
|
response = connection.getresponse()
|
|
result = response.status, response.read()
|
|
connection.close()
|
|
return result
|
|
|
|
def data(path):
|
|
return json.loads(get(path)[1])
|
|
|
|
try:
|
|
status, body = get('/')
|
|
assert status == 200 and b'CitrixPot' in body and b'full-text search' in body
|
|
all_events = data('/api/events?size=10')
|
|
assert all_events['scanned'] == 4 and len(all_events['events']) == 3
|
|
# newest first
|
|
assert all_events['events'][0]['event'] == 'simulated_command'
|
|
# pagination
|
|
page0 = data('/api/events?size=2&page=0')
|
|
page1 = data('/api/events?size=2&page=1')
|
|
assert len(page0['events']) == 2 and page0['has_more'] is True
|
|
assert len(page1['events']) == 1 and page1['has_more'] is False
|
|
assert page0['events'][0]['event'] == 'simulated_command'
|
|
# filters
|
|
assert len(data('/api/events?event=request')['events']) == 1
|
|
assert len(data('/api/events?q=simulated')['events']) == 1
|
|
assert len(data('/api/events?q=command')['events']) == 3
|
|
assert len(data('/api/events?peer=1.1.1.1')['events']) == 1
|
|
assert len(data('/api/events?payload=1')['events']) == 1
|
|
# multi-term full-text (all terms must match)
|
|
assert len(data('/api/events?q=id%20netscaler')['events']) == 3
|
|
assert len(data('/api/events?q=id%20nonexistent')['events']) == 0
|
|
# artifacts
|
|
status, body = get('/api/artifact/' + sha)
|
|
assert status == 200 and body == payload
|
|
status, body = get('/api/artifact/' + sha + '?inline=1')
|
|
assert status == 200 and b'command-output' in body
|
|
assert get('/api/artifact/not-a-hash')[0] == 404
|
|
assert get('/api/artifact/' + 'a' * 64)[0] == 404
|
|
assert get('/nope')[0] == 404
|
|
# hide operator IPs
|
|
server.hide_ips = {'8.8.8.8'}
|
|
hidden_on = data('/api/events?hide=1')
|
|
assert len(hidden_on['events']) == 1 and hidden_on['hidden'] == 1
|
|
hidden_off = data('/api/events?hide=0')
|
|
assert len(hidden_off['events']) == 3 and hidden_off['hidden'] == 0
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(3)
|
|
print('PASS: local read-only viewer: paged, filtered, multi-term search and inert artifacts')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|