907 lines
45 KiB
Python
907 lines
45 KiB
Python
#!/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()
|