118 lines
7.0 KiB
Python
118 lines
7.0 KiB
Python
"""Run: python3 test_profile.py. Passive fingerprints and inert header/body intake."""
|
|
import gzip
|
|
import base64
|
|
import hashlib
|
|
import http.client
|
|
import struct
|
|
import threading
|
|
import xml.etree.ElementTree as ET
|
|
import urllib.parse
|
|
import zlib
|
|
|
|
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
|
|
status, headers, xml = request('/nf/auth/doAuthentication.do', 'POST',
|
|
b'login=profile-check&passwd=synthetic',
|
|
{'Content-Type': 'application/x-www-form-urlencoded'})
|
|
root = ET.fromstring(xml)
|
|
labels = root.findall('.//c:Label', ns)
|
|
assert any(label.find('c:Type', ns).text == 'error' for label in labels)
|
|
assert root.find('c:Result', ns).text == 'more-info' # Never grant authentication.
|
|
assert request('/nf/auth/not-a-real-endpoint.do')[0] == 404
|
|
assert request('/nf/auth/doAuthentication.do')[0] == 405
|
|
assert b'<script>' not in html and b'<style>' not in html # Compatible with our CSP.
|
|
for path in ('/vpn/js/rdx/core/css/rdx.css', '/vpn/js/rdx.js', '/logon/LogonPoint/login.js'):
|
|
status, headers, asset = request(path)
|
|
assert status == 200 and asset and 'Set-Cookie' not in headers
|
|
assert 'no-store' not in headers['Cache-Control']
|
|
assert request(path, headers={'If-None-Match': headers['ETag']})[0] == 304
|
|
assert request(path, headers={'If-None-Match': 'W/' + headers['ETag']})[0] == 304
|
|
assert request(path, headers={'If-Modified-Since': headers['Last-Modified']})[0] == 304
|
|
assert request(path, headers={'If-Modified-Since': 'invalid'})[0] == 200
|
|
assert request(path, headers={'If-None-Match': '"different"',
|
|
'If-Modified-Since': headers['Last-Modified']})[0] == 200
|
|
head = request(path, 'HEAD')
|
|
assert head[0] == 200 and head[2] == b'' and int(head[1]['Content-Length']) == len(asset)
|
|
before = len(events)
|
|
status, rejected_headers, _ = request(path, 'POST', b'capture-only-check-method')
|
|
assert status == 405 and rejected_headers['Allow'] == 'GET, HEAD'
|
|
assert any(e['event'] == 'request_body' and e['evidence']['bytes'] == 25 for e in events[before:])
|
|
before = len(events)
|
|
assert request('/nscsp_violation/report_uri', 'POST', b'{}',
|
|
{'Content-Type': 'application/csp-report'})[0] == 204
|
|
assert any(e['event'] == 'request_body' and e['evidence']['bytes'] == 2 for e in events[before:])
|
|
assert request('/nscsp_violation/report_uri')[0] == 405
|
|
status, headers, body = request('/cgi/logout')
|
|
assert status == 302 and headers['Location'] == '/vpn/index.html' and body == b''
|
|
assert 'Max-Age=0' in headers['Set-Cookie']
|
|
# 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)
|
|
xml = b'<Response xmlns="urn:oasis:names:tc:SAML:2.0:protocol">inert-marker</Response>'
|
|
form = urllib.parse.urlencode({'SAMLResponse': base64.b64encode(xml).decode()}).encode()
|
|
assert request('/cgi/samlauth', 'POST', form, {'Content-Type': 'application/x-www-form-urlencoded'})[0] == 404
|
|
saml = next(e for e in reversed(events) if e['event'] == 'saml_payload')
|
|
assert saml['source'] == 'form' and saml['evidence']['sha256'] == hashlib.sha256(xml).hexdigest()
|
|
raw_body = next(e for e in events if e['event'] == 'request_body' and e['request_id'] == saml['request_id'])
|
|
assert raw_body['evidence']['sha256'] == hashlib.sha256(form).hexdigest()
|
|
def redirect(value):
|
|
compressor = zlib.compressobj(wbits=-zlib.MAX_WBITS)
|
|
compressed = compressor.compress(value) + compressor.flush()
|
|
return '/cgi/samlauth?' + urllib.parse.urlencode({'SAMLRequest': base64.b64encode(compressed).decode()})
|
|
assert request(redirect(xml))[0] == 404
|
|
assert any(e['event'] == 'saml_payload' and e['encoding'] == 'base64+deflate'
|
|
and e['evidence']['sha256'] == hashlib.sha256(xml).hexdigest() for e in events)
|
|
before = len(events)
|
|
assert request(redirect(b'A' * 65537))[0] == 404
|
|
assert any(e['event'] == 'limit' and e.get('resource') == 'saml_inflation' for e in events[before:])
|
|
assert not any(e['event'] == 'saml_payload' and e['encoding'] == 'base64+deflate' for e in events[before:])
|
|
assert request('/cgi/samlauth?SAMLResponse=not-base64')[0] == 404
|
|
assert any(e['event'] == 'saml_decode_error' for e in events)
|
|
assert request('/logon/LogonPoint/Authentication/GetUserName')[0] == 404
|
|
assert any({'source': 'uri', 'kind': 'gateway-username-probe'} in e.get('indicators', []) 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 metadata, auth XML, SAML bytes/inflation bounds, indicators, private boundary')
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(3)
|