citrixpot/honeypot.py

#!/usr/bin/env python3
"""Experimental CVE-2026-88771 decoy. Simulation by default; opt-in disposable VM execution."""
import argparse
import base64
import binascii
import collections
import gzip
import email.parser
import email.policy
import email.utils
import hashlib
import html
import http.server
import ipaddress
import io
import json
import os
import pathlib
import posixpath
import re
import secrets
import shlex
import ssl
import sys
import threading
import time
import urllib.parse
import zlib
from trusted_proxy import source as proxy_source, forwarded_client
from evidence import Evidence, EvidenceError, RecordingReader
from nitro import respond as nitro_response
from env_file import load, env_int

MAX_BODY = 16 * 1024 * 1024
MAX_FILES = 64
MAX_FILE = 8192
MAX_SESSIONS = 128
MAX_PENDING = 16
TTL = 3600
WEBROOT = '/netscaler/ns_gui'
UI = pathlib.Path(__file__).parent / 'ui'
LOGIN = (UI / 'login.html').read_bytes()
# The login page is served at every entry point a Citrix Gateway exposes.
LOGIN_PATHS = {'/', '/vpn/', '/vpn/index.html', '/vpn/tmindex.html',
               '/logon/LogonPoint/tmindex.html', '/logon/LogonPoint/index.html',
               '/logon/LogonPoint/custom.html', '/menu/neo/'}
JS = 'application/javascript; charset=utf-8'
STATIC = {
    '/logon/LogonPoint/login.js': ((UI / 'login.js').read_bytes(), JS),
    '/logon/LogonPoint/receiver/images/common/ReceiverFullScreenBackground.jpg': ((UI / 'background.jpg').read_bytes(), 'image/jpeg'),
    '/vpn/media/citrixgateway_logo_white.png': ((UI / 'logo.png').read_bytes(), 'image/png'),
    '/logon/LogonPoint/receiver/images/common/icon_vpn.ico': ((UI / 'favicon.ico').read_bytes(), 'image/vnd.microsoft.icon'),
    '/vpn/images/AccessGateway.ico': ((UI / 'favicon.ico').read_bytes(), 'image/vnd.microsoft.icon'),
    '/favicon.ico': ((UI / 'favicon.ico').read_bytes(), 'image/vnd.microsoft.icon'),
    '/vpn/js/rdx/core/css/rdx.css': ((UI / 'rdx.css').read_bytes(), 'text/css; charset=utf-8'),
    '/vpn/js/rdx.js': ((UI / 'rdx.js').read_bytes(), JS),
    '/vpn/js/rdx/core/rdx.js': ((UI / 'rdx.js').read_bytes(), JS),
    '/vpn/js/gateway_login_view.js': (b"rdx.gateway_login_view=function(){return new rdx.base_view(document.querySelector('main'));};\n", JS),
    '/vpn/js/gateway_login_form_view.js': (b"rdx.gateway_login_form_view=function(){return new rdx.base_view(document.getElementById('login-form'));};\n", JS),
    '/logon/LogonPoint/init.js': (b"window.CTXS=window.CTXS||{};CTXS.init=function(){return rdx.utils.get_view('login-form');};\n", JS),
    '/logon/LogonPoint/receiver/js/ctxs.core.min.js': ((UI / 'rdx.js').read_bytes(), JS),
}

# Synthetic translations; metadata matches the default 14.1-29.63 persona.
# Source: Fox-IT citrix-netscaler-triage version timestamp catalog.
ASSET_MTIME = env_int('HONEYPOT_ASSET_MTIME', 1723114781)
UI_MTIME = int(max(p.stat().st_mtime for p in UI.iterdir() if p.is_file()))
for language in ('en', 'de'):
    translation = {'LogOn': 'Log On' if language == 'en' else 'Anmelden',
                   'UserName': 'User name' if language == 'en' else 'Benutzername',
                   'Password': 'Password' if language == 'en' else 'Kennwort',
                   'CitrixCopyright': 'Copyright (c) Citrix Systems, Inc.'}
    for suffix in ('lang', 'lang-ext'):
        path = f'/vpn/js/rdx/core/{suffix}/rdx_{language}.json'
        encoded = json.dumps(translation, ensure_ascii=False).encode()
        STATIC[path] = (encoded, 'application/json; charset=utf-8')
        compressed = io.BytesIO()
        with gzip.GzipFile(filename=f'rdx_{language}.json', mode='wb', fileobj=compressed,
                           mtime=ASSET_MTIME) as stream:
            stream.write(encoded)
        STATIC[path + '.gz'] = (compressed.getvalue(), 'application/json; charset=utf-8')


class Session:
    def __init__(self):
        self.sid = secrets.token_hex(16)
        self.seen = time.monotonic()
        self.cwd = '/var/tmp'
        self.files = {'/etc/passwd': 'root:*:0:0:Charlie &:/root:/bin/sh\n',
                      '/nsconfig/ns.conf': '# synthetic honeypot configuration\n'}
        self.pending = collections.deque(maxlen=MAX_PENDING)
        self.auth_started = time.monotonic()
        self.auth_count = 0

    def path(self, name):
        # All paths stay dictionary keys, including traversal and absolute paths.
        return posixpath.normpath(posixpath.join(self.cwd, name))


def commands(text):
    """Small shell-like grammar; unsupported constructs are recorded, never executed."""
    lexer = shlex.shlex(text, posix=True, punctuation_chars=';&|<>')
    lexer.whitespace_split = True
    lexer.commenters = '#'
    group = []
    for token in lexer:
        if token in (';', '&&', '||', '|', '&'):
            if group:
                yield group
            group = []
        else:
            group.append(token)
    if group:
        yield group


def simulate(session, text, emit, download=None):
    # ponytail: bounded subset, not a POSIX shell; add grammar only from captured evidence.
    try:
        groups = list(commands(text))
        if len(groups) > 32:
            emit('limit', resource='simulated_commands', total=len(groups), retained=32)
        groups = groups[:32]
    except ValueError:
        emit('unsupported', reason='shell_parse', command=text)
        return
    for tokens in groups:
        original = tokens[:]
        dest, append = None, False
        for operator in ('>>', '>'):
            if operator in tokens:
                index = tokens.index(operator)
                if index + 1 < len(tokens):
                    dest, append = session.path(tokens[index + 1]), operator == '>>'
                tokens = tokens[:index]
                break
        if not tokens:
            continue
        name, *args = tokens
        name = posixpath.basename(name)
        output = ''
        supported = True
        if name == 'id':
            output = 'uid=0(root) gid=0(wheel) groups=0(wheel)\n'
        elif name == 'whoami':
            output = 'root\n'
        elif name == 'uname':
            output = 'FreeBSD ns 11.4-RELEASE-p17 FreeBSD 11.4-RELEASE-p17 amd64\n'
        elif name == 'pwd':
            output = session.cwd + '\n'
        elif name == 'cd' and args:
            session.cwd = session.path(args[0])
            emit('virtual_directory_change', path=session.cwd)
        elif name == 'cat':
            for arg in args:
                path = session.path(arg)
                emit('virtual_file_read', path=path, exists=path in session.files,
                     content=session.files.get(path, '').encode())
            output = ''.join(session.files.get(session.path(arg), 'cat: No such file or directory\n') for arg in args)
        elif name == 'ls':
            output = '\n'.join(sorted(session.files)) + '\n'
        elif name == 'echo':
            output = ' '.join(args) + '\n'
        elif name == 'printf' and args:
            if args[0] in ('%s', '%s\\n'):
                output = ('\n' if args[0].endswith('\\n') else '').join(args[1:])
                if args[0].endswith('\\n'):
                    output += '\n'
            elif len(args) == 1 and '%' not in args[0]:
                output = args[0].replace('\\n', '\n')
            else:
                supported = False
        elif name == 'touch' and len(args) == 1:
            dest = session.path(args[0])
            output = session.files.get(dest, '')
        elif name == 'rm':
            for arg in args:
                if not arg.startswith('-'):
                    path = session.path(arg)
                    emit('virtual_file_delete', path=path, exists=path in session.files,
                         content=session.files.get(path, '').encode())
                    session.files.pop(path, None)
        elif name in ('curl', 'wget', 'nc', 'ncat', 'telnet', 'ssh', 'ftp', 'fetch'):
            emit('outbound_attempt', argv=original, disposition='recorded_only')
            if download and name in ('curl', 'wget', 'fetch'):
                download(' '.join(original))  # GET and store the URL's bytes; never executed
            supported = False
        elif name in (':', 'true', 'mkdir'):
            pass
        else:
            supported = False
        emit('simulated_command' if supported else 'unsupported', argv=original,
             cwd=session.cwd, output=output[:MAX_FILE], output_truncated=len(output) > MAX_FILE)
        if dest and supported:
            if dest in session.files or len(session.files) < MAX_FILES:
                content = (session.files.get(dest, '') if append else '') + output
                session.files[dest] = content[:MAX_FILE]
                emit('virtual_file_write', path=dest, bytes=len(session.files[dest]),
                     content=session.files[dest].encode(),
                     truncated=len(content) > MAX_FILE)
            else:
                emit('limit', resource='virtual_files')


def extract(value):
    """Approximate the documented log-to-core-name transform, without running it."""
    if not re.search(r'pitboss.*PPE.*(?:missed too many heartbeats|unexpectedly died)', value, re.I):
        return None
    if 'NSPPE' not in value:
        return None
    tail = 'NSPPE' + value.rsplit('NSPPE', 1)[1]
    # The published backtick wrapper carries the full command, spaces included, anywhere in the line.
    substitutions = re.findall(r'`([^`]+)`', tail)
    if substitutions:
        return ';'.join(substitutions)
    fields = tail.replace('(', '').replace(')', '').split()
    core = '-'.join(fields[:2])
    if ';' not in core:
        return None
    # Simple semicolon injection without backticks.
    return core.split(';', 1)[-1]


# Conservative, path-based indicators for known Citrix ADC/Gateway exploitation and probes.
# Every request is already logged in full; these labels make hunting fast without guessing.
SIGNATURES = (
    ('CVE-2019-19781', re.compile(r'/vpn/\.\./vpns|/vpns/(?:cfg|portal|scripts)|\.\./vpns', re.I)),
    ('path-traversal', re.compile(r'\.\.(?:/|%2f)|%2e%2e|%252e', re.I)),
    ('config-read', re.compile(r'/nsconfig/|/flash/nsconfig|/ns\.conf', re.I)),
    ('management-probe', re.compile(r'/(?:gwtest|rapi|cgi-bin|menu|nitro)/', re.I)),
)


def exploit_signatures(path):
    raw = urllib.parse.unquote(path)
    return [name for name, pattern in SIGNATURES if pattern.search(path) or pattern.search(raw)]


class Engine:
    def __init__(self, emit, delay=1, save=None, fetcher=None):
        self.sink, self.save = emit, save
        self.delay = delay
        self.sessions = collections.OrderedDict()
        self.lock = threading.RLock()
        self.backend = None
        self.fetcher = fetcher

    def emit(self, event, **data):
        try:
            if 'content' in data:
                content = data.pop('content')
                data['evidence'] = (self.save(content) if self.save else
                                    dict(sha256=hashlib.sha256(content).hexdigest(), bytes=len(content)))
            self.sink(event, **data)
        except OSError as error:
            raise EvidenceError(str(error)) from error

    def session(self, peer):
        # ponytail: per-source state merges clients behind NAT; use validated sessions if needed.
        now = time.monotonic()
        for key in list(self.sessions):
            if now - self.sessions[key].seen > TTL:
                self.emit('session_expired', peer=key, session=self.sessions[key].sid,
                          pending_dropped=len(self.sessions[key].pending))
                del self.sessions[key]
        if peer not in self.sessions:
            if len(self.sessions) >= MAX_SESSIONS:
                key, old = self.sessions.popitem(last=False)
                self.emit('session_evicted', peer=key, session=old.sid, pending_dropped=len(old.pending))
            self.sessions[peer] = Session()
        state = self.sessions[peer]
        state.seen = now
        self.sessions.move_to_end(peer)
        return state

    def auth_attempt(self, state, context, path):
        # ponytail: bounded fixed 5-minute window; raw events support later cross-window analysis.
        with self.lock:
            now = time.monotonic()
            if now - state.auth_started >= 300:
                state.auth_started, state.auth_count = now, 0
            state.auth_count += 1
            data = dict(context, path=path, attempts_in_window=state.auth_count,
                        window_seconds=300, threshold=10,
                        classification='suspected_brute_force' if state.auth_count >= 10 else 'auth_submission',
                        outcome='decoy_no_authentication')
            self.emit('auth_attempt', **data)
            if state.auth_count % 10 == 0:
                self.emit('brute_force_suspected', **data)

    def ingest(self, state, candidates, context=None):
        context = dict(context or {}, session=state.sid)
        for source, value in candidates:
            command = extract(value)
            if command is not None:
                self.emit('injection_candidate', **context, source=source, value=value,
                          command=command, confidence='heuristic')
                if len(state.pending) < MAX_PENDING:
                    state.pending.append((time.monotonic() + self.delay, command, context))
                else:
                    self.emit('limit', **context, resource='pending_commands')

    def tick(self):
        pending_vm = []
        with self.lock:
            for state in self.sessions.values():
                while state.pending and state.pending[0][0] <= time.monotonic():
                    _, command, context = state.pending.popleft()
                    if self.backend:
                        pending_vm.append((command, context))
                        continue
                    self.emit('simulation_start', **context, command=command)
                    download = (lambda text: self.fetcher.submit(text, context)) if self.fetcher else None
                    simulate(state, command, lambda event, **data: self.emit(event, **context, **data), download)
                    self.emit('simulation_end', **context)
        for command, context in pending_vm:
            self.backend.execute(command, context)


class Server(http.server.ThreadingHTTPServer):
    daemon_threads = True
    allow_reuse_address = True

    def __init__(self, address, engine, tls=None, proxy_from=None, serve_guest_files=True,
                 trusted_forwarded=()):
        self.engine, self.tls = engine, tls
        self.proxy_from = proxy_from
        self.serve_guest_files = serve_guest_files
        # Only a hop (the PROXY-declared source) inside one of these networks may set client headers.
        self.trusted_forwarded = [ipaddress.ip_network(value, strict=False) for value in trusted_forwarded]
        self.capture_failed = False
        self.slots = threading.BoundedSemaphore(8)
        self.local = threading.local()
        super().__init__(address, Handler)

    def _trusts_forwarded(self, peer):
        try:
            address = ipaddress.ip_address(peer)
        except ValueError:
            return False
        return any(address in network for network in self.trusted_forwarded)

    def process_request(self, request, address):
        if not self.slots.acquire(blocking=False):
            self.engine.emit('connection_rejected', peer=address[0], peer_port=address[1], reason='connection_limit')
            self.shutdown_request(request)
            return
        try:
            super().process_request(request, address)
        except BaseException:
            self.slots.release()
            raise

    def process_request_thread(self, request, address):
        try:
            request.settimeout(5)
            if self.proxy_from:
                original = address
                address, header = proxy_source(request, original, self.proxy_from)
                self.engine.emit('trusted_ingress', peer=address[0], peer_port=address[1],
                                 transport_peer=original[0], transport_port=original[1], content=header)
            if self.tls:
                request = self.tls.wrap_socket(request, server_side=True)
            # The hop that connected to us is the PROXY-declared source (or socket peer without PROXY).
            self.local.trusted = self._trusts_forwarded(address[0])
            super().process_request_thread(request, address)
        except (OSError, ssl.SSLError, ValueError, UnicodeError):
            if isinstance(sys.exc_info()[1], EvidenceError):
                self.handle_error(request, address)
            else:
                try:
                    self.engine.emit('transport_error', peer=address[0], peer_port=address[1],
                                     error=type(sys.exc_info()[1]).__name__)
                except EvidenceError as error:
                    self.stop_capture(error)
            self.shutdown_request(request)
        finally:
            self.slots.release()

    def handle_error(self, request, address):
        error = sys.exc_info()[1]
        if isinstance(error, EvidenceError):
            self.stop_capture(error)
        else:
            try:
                self.engine.emit('handler_error', peer=address[0], peer_port=address[1], error=type(error).__name__)
            except EvidenceError as failure:
                self.stop_capture(failure)

    def stop_capture(self, error):
        if self.capture_failed:
            return
        self.capture_failed = True
        if self.engine.save:
            store = getattr(self.engine.save, '__self__', None)
            if isinstance(store, Evidence):
                try:
                    marker = store.directory / 'STOP'
                    marker.write_text('Capture failed; preserve evidence and review storage before clearing.\n')
                    marker.chmod(0o600)
                except OSError:
                    pass
        if self.engine.backend:
            try:
                self.engine.backend.close(trip=True)
            except OSError:
                pass
        print(f'CAPTURE STOPPED: {error}', file=sys.stderr, flush=True)
        threading.Thread(target=self.shutdown, daemon=True).start()


class Handler(http.server.BaseHTTPRequestHandler):
    server_version = 'Apache'
    sys_version = ''
    protocol_version = 'HTTP/1.1'

    def version_string(self):
        return self.server_version

    def log_message(self, format, *args):
        self.event('http_diagnostic', message=format % args)

    def event(self, event, **data):
        self.server.engine.emit(event, **self.context, **data)

    def handle_one_request(self):
        self.context = dict(request_id=secrets.token_hex(16), peer=self.client_address[0],
                            peer_port=self.client_address[1],
                            local_ip=self.connection.getsockname()[0],
                            local_port=self.connection.getsockname()[1])
        self.status = None
        self.body_complete = False
        self.started = time.monotonic_ns()
        original = self.rfile
        recorder = RecordingReader(original, MAX_BODY + 131072)
        self.rfile = recorder
        try:
            self.event('connection', tls=isinstance(self.connection, ssl.SSLSocket),
                       tls_cipher=self.connection.cipher() if isinstance(self.connection, ssl.SSLSocket) else None)
            super().handle_one_request()
        except EvidenceError:
            raise
        except (TimeoutError, ConnectionError, OSError) as error:
            self.event('request_interrupted', reason=type(error).__name__)
        finally:
            self.rfile = original
            self.event('request_end', status=self.status,
                       duration_ms=(time.monotonic_ns() - self.started) / 1e6,
                       body_complete=self.body_complete, wire_truncated=recorder.truncated,
                       content=bytes(recorder.data),
                       capture_scope='consumed HTTP bytes after TLS; unread/rejected bytes not captured')

    def send_response(self, code, message=None):
        self.status = code
        super().send_response(code, message)

    def send_error(self, code, message=None, explain=None):
        self.event('http_rejected', status=code, reason=message or self.responses.get(code, ('',))[0])
        super().send_error(code, message, explain)

    def __getattr__(self, name):
        if name.startswith('do_'):
            return self.handle_decoy
        raise AttributeError(name)

    def read_bytes(self, size):
        chunks = bytearray()
        while len(chunks) < size:
            if time.monotonic_ns() - self.started > 30 * 10**9:
                raise ValueError('request duration limit')
            data = self.rfile.read1(min(65536, size - len(chunks)))
            if not data:
                raise ValueError('incomplete request body')
            chunks.extend(data)
        return bytes(chunks)

    def read_body(self):
        lengths = self.headers.get_all('Content-Length', [])
        transfers = self.headers.get_all('Transfer-Encoding', [])
        if len(lengths) > 1 or len(transfers) > 1 or (lengths and transfers):
            raise ValueError('ambiguous body framing')
        if transfers:
            if transfers[0].lower().strip() != 'chunked':
                raise ValueError('unsupported transfer encoding')
            body = bytearray()
            for _ in range(4096):
                line = self.rfile.readline(1025)
                if len(line) > 1024 or not line.endswith(b'\r\n'):
                    raise ValueError('invalid chunk header')
                size_text = line.split(b';', 1)[0].strip()
                if not re.fullmatch(rb'[0-9a-fA-F]+', size_text):
                    raise ValueError('invalid chunk size')
                size = int(size_text, 16)
                if len(body) + size > MAX_BODY:
                    raise OverflowError('body limit')
                if size == 0:
                    trailers = bytearray()
                    while len(trailers) <= 32768:
                        line = self.rfile.readline(1025)
                        trailers.extend(line)
                        if line == b'\r\n':
                            self.event('request_trailers', content=bytes(trailers))
                            return bytes(body)
                        if not line or len(line) > 1024:
                            break
                    raise ValueError('invalid trailers')
                body.extend(self.read_bytes(size))
                if self.read_bytes(2) != b'\r\n':
                    raise ValueError('invalid chunk terminator')
            raise ValueError('chunk count limit')
        length = int(lengths[0]) if lengths else 0
        if length < 0:
            raise ValueError('negative content length')
        if length > MAX_BODY:
            raise OverflowError('body limit')
        return self.read_bytes(length)

    def capture_body(self, body):
        self.event('request_body', content=body, content_type=self.headers.get('Content-Type', ''))
        kind = self.headers.get_content_type()
        if kind == 'multipart/form-data':
            raw_type = self.headers.get('Content-Type', '')
            # MIME parsing only: never use a supplied filename as a host path.
            message = email.parser.BytesParser(policy=email.policy.default).parsebytes(
                ('Content-Type: ' + raw_type + '\r\nMIME-Version: 1.0\r\n\r\n').encode('latin1') + body)
            if not message.is_multipart():
                self.event('multipart_parse_error', reason='not multipart; raw body retained')
                return []
            fields = []
            for index, part in enumerate(message.walk()):
                if index >= 128:
                    self.event('limit', resource='multipart_parts', raw_body_retained=True)
                    break
                if part.is_multipart():
                    continue
                payload = part.get_payload(decode=True) or b''
                filename = part.get_filename()
                field = part.get_param('name', header='content-disposition')
                self.event('upload' if filename is not None else 'multipart_field',
                           filename=filename, field=field, content_type=part.get_content_type(),
                           defects=[type(d).__name__ for d in part.defects], content=payload)
                if filename is not None and self.server.engine.backend:
                    self.server.engine.backend.upload(payload, self.context)
                if filename is None and len(payload) <= 65536:
                    fields.append((field or '', payload.decode('utf-8', 'replace')))
            return fields
        if kind == 'application/x-www-form-urlencoded':
            if len(body) > 65536:
                self.event('limit', resource='form_decoding', raw_body_retained=True)
                return []
            try:
                fields = urllib.parse.parse_qsl(body.decode('utf-8', 'replace'), max_num_fields=128)
                self.event('form_fields', fields=fields)
                return fields
            except ValueError:
                self.event('form_parse_error', raw_body_retained=True)
                return []
        if kind == 'application/json':
            try:
                document = json.loads(body)
                files = document.get('systemfile', []) if isinstance(document, dict) else []
                files = [files] if isinstance(files, dict) else files
                if isinstance(files, list):
                    for item in files[:128]:
                        if isinstance(item, dict) and isinstance(item.get('filecontent'), str):
                            content = base64.b64decode(item['filecontent'], validate=True)
                            self.event('upload', filename=item.get('filename'), field='systemfile.filecontent',
                                       content_type='application/octet-stream', content=content)
            except (ValueError, RecursionError, binascii.Error):
                self.event('json_upload_parse_error', raw_body_retained=True)
        if body:
            self.event('upload_or_raw_payload', filename=None, content=body)
            if self.server.engine.backend:
                self.server.engine.backend.upload(body, self.context)
        return []

    def capture_saml(self, pairs, source):
        # Decode protocol envelopes only. Never parse XML, expand entities or authenticate.
        for field, value in pairs:
            if field.lower() not in ('samlrequest', 'samlresponse'):
                continue
            try:
                decoded = base64.b64decode(value, altchars=b'-_', validate=True)
            except (ValueError, binascii.Error):
                self.event('saml_decode_error', field=field, source=source, raw_request_retained=True)
                continue
            self.event('saml_payload', field=field, source=source, encoding='base64', content=decoded)
            # HTTP-Redirect binding may carry raw DEFLATE. Bound output before allocation.
            if source == 'query' and not decoded.lstrip().startswith(b'<'):
                try:
                    inflater = zlib.decompressobj(-zlib.MAX_WBITS)
                    inflated = inflater.decompress(decoded, 65537)
                    if len(inflated) > 65536 or inflater.unconsumed_tail:
                        self.event('limit', resource='saml_inflation', decoded_payload_retained=True)
                    elif inflater.eof and not inflater.unused_data:
                        self.event('saml_payload', field=field, source=source,
                                   encoding='base64+deflate', content=inflated)
                    else:
                        self.event('saml_decode_error', field=field, source=source,
                                   reason='incomplete deflate envelope', decoded_payload_retained=True)
                except zlib.error:
                    self.event('saml_decode_error', field=field, source=source,
                               reason='invalid deflate envelope', decoded_payload_retained=True)

    def handle_decoy(self):
        engine = self.server.engine
        # Trusted Cloudflare/edge headers only when the connecting hop is a configured proxy.
        if getattr(self.server.local, 'trusted', False) and self.headers is not None:
            client, source = forwarded_client(self.headers)
            if client and client != self.client_address[0]:
                self.context.update(peer=client, client_ip_source=source,
                                    proxy_peer=self.client_address[0])
                self.client_address = (client, self.client_address[1])
        with engine.lock:
            state = engine.session(self.client_address[0])
            self.context['session'] = state.sid
        if engine.backend:
            engine.backend.register_peer(self.client_address[0])
        self.event('request', method=self.command, uri=self.path, version=self.request_version,
                   headers=list(self.headers.raw_items()))
        signatures = exploit_signatures(self.path)
        if signatures:
            self.event('exploit_attempt', signatures=signatures, uri=self.path)
        if sum(len(k) + len(v) for k, v in self.headers.items()) > 32768:
            self.send_error(431)
            return
        try:
            body = self.read_body()
        except (ValueError, OverflowError) as error:
            self.send_error(413 if isinstance(error, OverflowError) else 400, str(error))
            return
        self.body_complete = True
        fields = self.capture_body(body)
        # Hunt across all intake surfaces; labels are indicators, never CVE confirmation.
        indicators = []
        for source, value in [('header:' + k, v) for k, v in self.headers.raw_items()] + [
                ('body', body[:65536].decode('utf-8', 'replace'))]:
            decoded = urllib.parse.unquote_plus(value)
            if re.search(r'pitboss|NSPPE', decoded, re.I):
                indicators.append({'source': source, 'kind': 'netscaler-log-marker'})
            if re.search(r'base64|b64decode|atob\s*\(', decoded, re.I):
                indicators.append({'source': source, 'kind': 'encoded-content'})
            if source.startswith('header:') and len(value) > 8192:
                indicators.append({'source': source, 'kind': 'large-header'})
        try:
            url = urllib.parse.urlsplit(self.path)
        except ValueError:
            self.send_error(400, 'invalid URL')
            return
        self.capture_saml(fields, 'form')
        if len(url.query) <= 65536:
            try:
                query_fields = urllib.parse.parse_qsl(url.query, max_num_fields=128)
                self.capture_saml(query_fields, 'query')
                if any(k.lower() in ('samlrequest', 'samlresponse') for k, _ in fields + query_fields):
                    indicators.append({'source': 'request', 'kind': 'saml-envelope'})
            except ValueError:
                self.event('limit', resource='query_decoding', raw_request_retained=True)
        else:
            self.event('limit', resource='query_decoding', raw_request_retained=True)
        if url.path == '/logon/LogonPoint/Authentication/GetUserName':
            indicators.append({'source': 'uri', 'kind': 'gateway-username-probe'})
        if indicators:
            self.event('request_indicators', indicators=indicators,
                       body_inspected_bytes=min(len(body), 65536), body_inspection_truncated=len(body) > 65536)
        auth_fields = {key.lower() for key, _ in fields}
        if self.command == 'POST' and (url.path.startswith('/nf/auth/') or
                url.path in ('/vpn/login', '/vpn/login.html', '/cgi/login') or
                auth_fields.intersection({'passwd', 'password', 'pwd', 'passcode'})):
            engine.auth_attempt(state, self.context, url.path)
        vm_read_path = None
        extra_headers = []
        with engine.lock:
            candidates = [('uri', urllib.parse.unquote(self.path))]
            candidates += [('header:' + key, value) for key, value in self.headers.items()]
            candidates += [('form:' + key, value) for key, value in fields]
            engine.ingest(state, candidates, self.context)
            code, data, mime = 200, LOGIN, 'text/html; charset=utf-8'
            if url.path in STATIC:
                data, mime = STATIC[url.path]
                if url.path.endswith('.gz'):
                    extra_headers.append(('Content-Encoding', 'gzip'))
                asset_mtime = ASSET_MTIME if '/lang/' in url.path or '/lang-ext/' in url.path else UI_MTIME
                extra_headers.append(('Last-Modified', self.date_time_string(asset_mtime)))
            elif url.path in ('/nf/auth/getAuthenticationRequirements.do', '/nf/auth/doAuthentication.do'):
                data = ('<AuthenticateResponse xmlns="http://citrix.com/authentication/response/1">'
                        '<Status>success</Status><Result>more-info</Result>'
                        '<StateContext>' + html.escape(state.sid) + '</StateContext>'
                        '<AuthenticationRequirements><PostBack>/nf/auth/doAuthentication.do</PostBack>'
                        '<Requirements>' + (
                        '<Requirement><Credential><Type>none</Type></Credential>'
                        '<Label><Text>Incorrect user name or password.</Text><Type>error</Type></Label>'
                        '<Input/></Requirement>' if url.path == '/nf/auth/doAuthentication.do'
                        and self.command == 'POST' and auth_fields.intersection({'login', 'passwd'}) else '') +
                        '<Requirement><Credential><ID>login</ID><Type>username</Type></Credential>'
                        '<Label><Text>User name:</Text><Type>plain</Type></Label>'
                        '<Input><Text><InitialValue/><Constraint>.+</Constraint></Text></Input></Requirement>'
                        '<Requirement><Credential><ID>passwd</ID><Type>password</Type></Credential>'
                        '<Label><Text>Password:</Text><Type>plain</Type></Label>'
                        '<Input><Text><Secret>true</Secret><InitialValue/><Constraint>.+</Constraint></Text></Input></Requirement>'
                        '<Requirement><Credential><ID>nsg-x1-logon-button</ID><Type>none</Type></Credential>'
                        '<Label><Type>none</Type></Label><Input><Button>Log On</Button></Input></Requirement>'
                        '</Requirements></AuthenticationRequirements>'
                        '</AuthenticateResponse>').encode()
                mime = 'application/vnd.citrix.authenticateresponse-1+xml; charset=utf-8'
            elif url.path == '/nscsp_violation/report_uri':
                code, data, mime = 204, b'', 'text/plain; charset=utf-8'
            elif url.path in ('/cgi/logout', '/vpn/logout.html'):
                code, data = 302, b''
                extra_headers = [('Location', '/vpn/index.html'),
                                 ('Set-Cookie', 'NSC_AAAC=; Path=/; Max-Age=0; Secure; HttpOnly')]
            elif url.path.startswith('/nitro/'):
                code, result, extra_headers = nitro_response(state, self.command, url, body, self.headers)
                data, mime = json.dumps(result).encode(), 'application/json'
            elif url.path not in LOGIN_PATHS:
                key = posixpath.normpath(WEBROOT + '/' + urllib.parse.unquote(url.path).lstrip('/'))
                if not self.server.serve_guest_files:
                    code, data = 404, b'Not Found'
                elif engine.backend and key.startswith(WEBROOT + '/'):
                    vm_read_path = key
                    code, data = 404, b'Not Found'
                elif key.startswith(WEBROOT + '/') and key in state.files:
                    data = state.files[key].encode()
                    mime = 'text/plain; charset=utf-8'
                    self.event('virtual_file_http_read', path=key, content=data)
                else:
                    code, data = 404, b'Not Found'
        # Capture and ingest precede protocol responses, including rejected methods.
        if url.path in STATIC or url.path in LOGIN_PATHS:
            allowed = ('GET', 'HEAD')
        elif url.path == '/nf/auth/doAuthentication.do' or url.path == '/nscsp_violation/report_uri':
            allowed = ('POST',)
        elif url.path == '/nf/auth/getAuthenticationRequirements.do':
            allowed = ('GET', 'HEAD', 'POST')
        else:
            allowed = None
        if allowed and self.command not in allowed:
            code, data, mime = 405, b'Method Not Allowed', 'text/plain; charset=utf-8'
            extra_headers = [('Allow', ', '.join(allowed))]
        if url.path in STATIC and code == 200:
            etag = '"' + hashlib.sha256(data).hexdigest() + '"'
            extra_headers.append(('ETag', etag))
            matches = self.headers.get('If-None-Match')
            not_modified = matches is not None and any(
                tag.strip().removeprefix('W/') in ('*', etag) for tag in matches.split(','))
            if matches is None and self.headers.get('If-Modified-Since'):
                try:
                    not_modified = email.utils.parsedate_to_datetime(
                        self.headers['If-Modified-Since']).timestamp() >= asset_mtime
                except (ValueError, TypeError, OverflowError):
                    pass
            if not_modified:
                code, data = 304, b''
        if vm_read_path:
            result = engine.backend.read(vm_read_path, self.context)
            if result is not None:
                code, data, mime = 200, result, 'text/plain; charset=utf-8'
        # Headers shaped like the real NetScaler Gateway AAA response; HTTP/1.1 keep-alive is default.
        headers = [('Content-Type', mime),
                   ('Cache-Control', 'public, max-age=0, must-revalidate' if url.path in STATIC
                    and code in (200, 304) else 'no-cache, no-store, must-revalidate'),
                   ('Pragma', 'no-cache'), ('Expires', '0'),
                   ('X-Frame-Options', 'SAMEORIGIN'), ('X-Content-Type-Options', 'nosniff'),
                   ('X-XSS-Protection', '1; mode=block'),
                   ('Content-Security-Policy',
                    "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; "
                    "img-src 'self' data:; font-src 'self'; frame-ancestors 'self'; "
                    "connect-src 'self' citrixng:// com.citrix.nsgclient:// nsgcepa://; "
                    "report-uri /nscsp_violation/report_uri")]
        if code not in (204, 304):
            headers.append(('Content-Length', str(len(data))))
        if url.path not in STATIC and url.path != '/nscsp_violation/report_uri' and not any(
                key == 'Set-Cookie' for key, _ in extra_headers):
            headers.append(('Set-Cookie', f'NSC_AAAC={state.sid}; Path=/; Secure; HttpOnly'))
        headers.extend(extra_headers)
        self.event('response', status=code, headers=headers, content=data,
                   body_suppressed=self.command == 'HEAD')
        self.send_response(code)
        for key, value in headers:
            self.send_header(key, value)
        self.end_headers()
        if self.command != 'HEAD':
            self.wfile.write(data)
        self.event('response_sent', status=code, body_bytes=0 if self.command == 'HEAD' else len(data))


def parse_trusted(spec):
    """Split a comma-separated CIDR list; @path reads one CIDR per line from a file."""
    items = []
    for raw in (spec or '').split(','):
        raw = raw.strip()
        if not raw:
            continue
        if raw.startswith('@'):
            items.extend(line.strip() for line in pathlib.Path(raw[1:]).read_text().splitlines())
        else:
            items.append(raw)
    result = [item for item in items if item and not item.startswith('#')]
    for item in result:
        ipaddress.ip_network(item, strict=False)
    return result


def main():
    load()
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--simulation-only', action='store_true', help='refuse real execution and URL fetching')
    parser.add_argument('--hide-guest-files', action='store_true', help='serve only fixed NetScaler UI/auth routes over HTTP')
    parser.add_argument('--vm-config', default=os.environ.get('DECOY_VM_CONFIG') or None,
                        help='opt-in dedicated disposable guest SSH configuration')
    parser.add_argument('--allow-real-execution', action='store_true',
                        help='required with --vm-config; real attacker commands then run in the isolated guest')
    parser.add_argument('--bind', default=os.environ.get('HONEYPOT_BIND', '127.0.0.1'))
    parser.add_argument('--port', type=int, default=env_int('HONEYPOT_PORT', 8443))
    parser.add_argument('--cert')
    parser.add_argument('--key')
    parser.add_argument('--proxy-protocol-from', default=os.environ.get('HONEYPOT_PROXY_PROTOCOL_FROM') or None,
                        help='require PROXY v1 only from this exact trusted transport IP')
    parser.add_argument('--trusted-forwarded-from', default=os.environ.get('HONEYPOT_TRUSTED_FORWARDED_FROM') or None,
                        help='comma-separated CIDRs whose CF-Connecting-IP/X-Forwarded-For headers are trusted')
    parser.add_argument('--fetch-payloads', action='store_true',
                        help='download payload URLs from attacker commands (HTTP GET only; never executed)')
    parser.add_argument('--delay', type=float, default=1,
                        help='simulated log processing delay in seconds (real appliance may take 24h)')
    parser.add_argument('--log-dir', default=os.environ.get('HONEYPOT_LOG_DIR', str(pathlib.Path(__file__).parent / 'logs')))
    parser.add_argument('--evidence-mib', type=int, default=env_int('HONEYPOT_EVIDENCE_MIB', 1024),
                        help='total evidence quota; stop rather than overwrite at this limit')
    args = parser.parse_args()
    if args.simulation_only and (args.vm_config or args.allow_real_execution or args.fetch_payloads):
        parser.error('--simulation-only cannot use execution or payload fetching')
    if bool(args.cert) != bool(args.key) or not 0 <= args.delay <= 86400 or args.evidence_mib < 1:
        parser.error('provide both cert/key; delay must be 0..86400; evidence-mib must be positive')
    if args.vm_config and not args.allow_real_execution:
        parser.error('--vm-config requires --allow-real-execution; keep internet-facing listeners in simulation mode')
    if args.allow_real_execution and not args.vm_config:
        parser.error('--allow-real-execution requires --vm-config')
    os.umask(0o077)
    directory = pathlib.Path(args.log_dir)
    if (directory / 'STOP').exists():
        print('Capture STOP latch exists; review storage before clearing.', file=sys.stderr)
        raise SystemExit(78)
    try:
        evidence = Evidence(directory, args.evidence_mib * 1024 * 1024)
    except OSError as error:
        print(f'Evidence storage unavailable: {error}', file=sys.stderr)
        raise SystemExit(78)
    context = None
    if args.cert:
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.load_cert_chain(args.cert, args.key)
    engine = Engine(evidence.emit, args.delay, evidence.save)
    try:
        server = Server((args.bind, args.port), engine, context, args.proxy_protocol_from,
                        serve_guest_files=not args.hide_guest_files,
                        trusted_forwarded=parse_trusted(args.trusted_forwarded_from))
    except (OSError, ValueError) as error:
        parser.error(f'invalid --trusted-forwarded-from: {error}')
    if args.fetch_payloads:
        from payload_fetch import Fetcher
        engine.fetcher = Fetcher(engine.emit, on_failure=server.stop_capture)
    if args.vm_config:
        from vm_backend import VMBackend
        engine.backend = VMBackend(args.vm_config, engine.emit)
        engine.backend.on_failure = server.stop_capture
    stopping = threading.Event()
    def worker():
        while not stopping.wait(0.25):
            try:
                engine.tick()
            except EvidenceError as error:
                server.stop_capture(error)
                break
    threading.Thread(target=worker, daemon=True).start()
    try:
        engine.emit('start', mode='disposable_vm' if engine.backend else 'simulation', bind=args.bind, port=server.server_port,
                    delay=args.delay, evidence_quota_bytes=evidence.quota, max_body_bytes=MAX_BODY,
                    serve_guest_files=server.serve_guest_files, real_execution=bool(engine.backend),
                    payload_fetch=bool(engine.fetcher))
    except EvidenceError as error:
        server.stop_capture(error)
        server.server_close()
        evidence.close()
        raise SystemExit(78)
    if engine.backend:
        print('WARNING: real execution enabled; attacker commands run only in the isolated disposable guest',
              file=sys.stderr, flush=True)
    if engine.fetcher:
        print('Payload fetch enabled: attacker URLs are downloaded (HTTP GET only, never executed)',
              file=sys.stderr, flush=True)
    print(f"Decoy: {'https' if context else 'http'}://{args.bind}:{server.server_port}; logs: {directory}", flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        stopping.set()
        if engine.backend:
            engine.backend.close()
        server.server_close()
        evidence.close()
    if server.capture_failed:
        raise SystemExit(78)


if __name__ == '__main__':
    main()