Improve Gateway fingerprints and header/body coverage

test_profile.py

@@ -0,0 +1,57 @@
+"""Run: python3 test_profile.py. Passive fingerprints and inert header/body intake."""
+import gzip
+import http.client
+import struct
+import threading
+import xml.etree.ElementTree as ET
+
+from honeypot import ASSET_MTIME, Engine, Server
+from test_private_boundary import PATHS
+
+events = []
+server = Server(('127.0.0.1', 0), Engine(lambda event, **data: events.append(dict(event=event, **data))))
+server.serve_guest_files = False
+thread = threading.Thread(target=server.serve_forever, daemon=True)
+thread.start()
+
+def request(path, method='GET', body=None, headers=None):
+    c = http.client.HTTPConnection('127.0.0.1', server.server_port, timeout=3)
+    c.request(method, path, body, headers or {})
+    r = c.getresponse()
+    result = r.status, dict(r.getheaders()), r.read()
+    c.close()
+    return result
+
+try:
+    status, headers, html = request('/')
+    assert status == 200 and b'<title class="_ctxstxt_NetscalerGateway">Citrix Gateway</title>' in html
+    assert headers['Set-Cookie'].startswith('NSC_')  # WhatWeb passive matcher
+    for suffix in ('lang', 'lang-ext'):
+        path = '/vpn/js/rdx/core/' + suffix + '/rdx_en.json'
+        status, headers, compressed = request(path + '.gz')
+        assert status == 200 and headers['Content-Encoding'] == 'gzip'
+        assert compressed.startswith(b'\x1f\x8b\x08\x08') and b'rdx_en.json\x00' in compressed[:100]
+        assert struct.unpack('<I', compressed[4:8])[0] == ASSET_MTIME == 1723114781
+        assert gzip.decompress(compressed) == request(path)[2]
+        assert request(path + '.gz', 'HEAD')[2] == b''
+    status, headers, xml = request('/nf/auth/getAuthenticationRequirements.do')
+    root = ET.fromstring(xml)
+    ns = {'c': 'http://citrix.com/authentication/response/1'}
+    assert root.find('c:AuthenticationRequirements/c:PostBack', ns).text == '/nf/auth/doAuthentication.do'
+    assert len(root.findall('.//c:Requirement', ns)) == 3
+    # Inert markers, no exploit syntax or commands. Original intake still correlated.
+    status, _, _ = request('/unknown', 'POST', b'base64=inert-marker',
+                           {'Content-Type': 'application/x-www-form-urlencoded', 'X_Test': 'NSPPE inert-marker'})
+    assert status == 404
+    indicator = next(e for e in events if e['event'] == 'request_indicators')
+    assert {'source': 'header:X_Test', 'kind': 'netscaler-log-marker'} in indicator['indicators']
+    assert {'source': 'body', 'kind': 'encoded-content'} in indicator['indicators']
+    assert any(e['event'] == 'request_body' and e['request_id'] == indicator['request_id'] for e in events)
+    for path in PATHS + ('/.env', '/nsconfig/ns.conf', '/api/events'):
+        status, _, body = request(path)
+        assert status == 404 and body == b'Not Found', path
+    print('PASS: Gateway fingerprint, gzip firmware metadata, auth XML, correlated indicators, private boundary')
+finally:
+    server.shutdown()
+    server.server_close()
+    thread.join(3)