sip: support TLS listeners and transport telemetry

Author: djkurlander

.env.example

@@ -72,6 +72,8 @@ COMPOSE_FILE=docker-compose.host.yml
 # Run only specific honeypots (default: all)
 # Protocols: SSH, TNET, FTP, RDP, SMB, SIP, HTTP, SMTP
 # Use PROTO:PORT syntax to run on multiple ports, e.g. HTTP:80,HTTP:443
+# Declared protocol options can follow the port, e.g. SIP:5061:TLS.
+# SIP on 5061 auto-enables TLS by convention; SIP:5061:TLS is equivalent.
 # ENABLED_PROTOCOLS=SSH,TNET,FTP,RDP,SMB,SIP,HTTP,SMTP
 
 # Number of uvicorn worker processes (systemd only; Docker uses 2 via CLI)

CLAUDE.md

@@ -230,7 +230,7 @@ Port 80 is open to all — it's a honeypot port. Port 443 can also be mapped to
 | `REDIS_HOST` | `localhost` | Redis server hostname (set to `redis` in Docker) |
 | `REDIS_DB` | `0` | Redis database index |
 | `DB_DIR` | `data` | Directory for SQLite databases and caches |
-| `ENABLED_PROTOCOLS` | all protocols | Comma-separated list of active protocols with optional port overrides, e.g. `SSH,SMTP:25,SMTP:587,HTTP:80,HTTP:443`. Empty string = ingest-only mode (no local honeypots). |
+| `ENABLED_PROTOCOLS` | all protocols | Comma-separated list of active protocols with optional port overrides and declared protocol options, e.g. `SSH,SMTP:25,SMTP:587,HTTP:80,HTTP:443,SIP:5061:TLS`. Empty string = ingest-only mode (no local honeypots). |
 | `DEFAULT_HOSTNAME` | unset | Canonical hostname all protocols advertise by default (an FQDN, e.g. `mail.corp.example`); each renders its own form (SMTP uses it as-is, SMB derives the short NetBIOS name). Per-protocol vars (`SMTP_HOSTNAME`, `SMB_SERVER_NAME`) override it; `PROTOCOL_VAR=auto` forces that protocol's built-in default instead. Also added to the self-redaction identity so it's scrubbed from captured fields. **Unset = each protocol's prior default** (SMTP reverse-DNS, SMB `WIN-SRV####`). **Docker (bridge networking only):** set this — bridge in-container discovery can't recover the advertised hostname — and also set `REDACT_SELF_IPS` (it can't see the public IP either). Under **host networking** (`docker-compose.host.yml`, the default in `.env.example`) the container sees the host's real IP/PTR, so discovery works like a bare-metal install and both are optional. |
 
 ### Web Server (`main.py`)
@@ -321,7 +321,9 @@ Port 80 is open to all — it's a honeypot port. Port 443 can also be mapped to
 
 | Variable | Default | Purpose |
 |----------|---------|---------|
-| `SIP_PORT` | `5060` | Listening port (UDP + TCP) |
+| `SIP_PORT` | `5060` | Listening port (UDP + TCP for cleartext SIP; TCP/TLS only when SSL mode is enabled). `ENABLED_PROTOCOLS=SIP:5061` auto-enables TLS by convention; `SIP:5061:TLS` is also accepted. |
+| `SIP_TLS_CERT_PATH` | `data/sip_tls.crt` | TLS certificate path for SIP/TLS; auto-generated if missing. |
+| `SIP_TLS_KEY_PATH` | `data/sip_tls.key` | TLS key path for SIP/TLS; auto-generated if missing. |
 | `SIP_OK_DIALPLAN` | `+,bare,00,011,9` | Which dialed forms the fake PBX answers (200) vs rejects (404). `all` (any resolvable E.164), `none`, or a comma list of dial-out prefixes prepended to the bare E.164 digits: `bare` (no prefix), `+`, or digit prefixes like `00`/`011`/`9`. A dial is accepted iff its canonicalized digits equal `prefix + computed-E.164`; numbers that resolve to no E.164 are always rejected. Knocks are recorded regardless. |
 | `SIP_REALM` | `asterisk` | SIP realm in authentication challenge |
 | `SIP_AUTH_CHALLENGE_MODE` | `mixed` | `always`, `never`, or `mixed` |
@@ -410,7 +412,8 @@ knocks_tnet(... username, password)
 knocks_ftp(... username, password)
 knocks_smtp(... username, password, smtp_stage, smtp_mail_from, smtp_rcpt_to, subject, body_id)  -- body_id → smtp_body_intel.id (full body stored deduped there, not inline)
 knocks_sip(... sip_method, sip_dial_string, sip_dial_number, sip_call_id, sip_cseq,
-           sip_extension, sip_dial_country, sip_dial_country_name, sip_dial_lat, sip_dial_lng)
+           sip_port, sip_transport, sip_extension, sip_dial_country, sip_dial_country_name,
+           sip_dial_lat, sip_dial_lng)
 knocks_smb(... username, smb_action, smb_share, smb_file, smb_version, smb_domain, smb_host)
 knocks_rdp(... username, rdp_source, domain)
 knocks_http(... http_method, http_path, http_user_agent, http_body)

honeypots/sip_honeypot.py

@@ -1,11 +1,13 @@
 #!/usr/bin/env python3
+import argparse
 import base64
 import json
 import os
 import random
 import re
 import socket
 import sqlite3
+import ssl
 import string
 import sys
 import threading
@@ -36,11 +38,14 @@
     PerIpTokenBucket,
     create_dualstack_tcp_listener,
     create_dualstack_udp_listener,
+    ensure_self_signed_server_cert,
     is_blocked,
     normalize_ip,
 )
 
 SIP_PORT = int(os.environ.get('SIP_PORT', '5060'))
+SIP_TLS_CERT_PATH = os.environ.get('SIP_TLS_CERT_PATH', 'data/sip_tls.crt')
+SIP_TLS_KEY_PATH = os.environ.get('SIP_TLS_KEY_PATH', 'data/sip_tls.key')
 SIP_REALM = os.environ.get('SIP_REALM', 'asterisk')
 SIP_SERVER_HEADER = os.environ.get('SIP_SERVER_HEADER', 'Asterisk PBX 18.0.0')
 SIP_SDP_SESSION_NAME = os.environ.get('SIP_SDP_SESSION_NAME', 'Asterisk')
@@ -960,7 +965,7 @@ def emit_knock(client_ip, extra=None, dedup_key=None):
     print(json.dumps(knock), flush=True)
 
 
-def process_sip_request(req, client_ip, allow_b2bua=False):
+def process_sip_request(req, client_ip, client_port=None, transport=None, allow_b2bua=False):
     headers = req.get('headers', {})
     method = req.get('method', 'UNKNOWN')
     uri = req.get('uri', '')
@@ -974,6 +979,8 @@ def process_sip_request(req, client_ip, allow_b2bua=False):
     dial_string = re.sub(r'^sips?:', '', uri).split('@')[0] if uri else ''
     common = {
         'sip_method': method,
+        'sip_port': SIP_PORT,
+        'sip_transport': transport or '',
         'sip_dial_string': dial_string,
         'sip_call_id': _header_first(headers, 'call-id') or '',
         'sip_cseq': _header_first(headers, 'cseq') or '',
@@ -1159,7 +1166,7 @@ def udp_send(resp, _addr=addr): sock.sendto(resp, _addr)
             if HAVE_B2BUA and sip_b2bua.handle_in_dialog(req, client_ip, udp_send):
                 trace(session_id, client_ip, 'udp_b2bua_in_dialog', method=req.get('method'))
                 continue
-            result = process_sip_request(req, client_ip, allow_b2bua=True)
+            result = process_sip_request(req, client_ip, client_port=addr[1], transport='udp', allow_b2bua=True)
             if result[0] == 'INVITE_B2BUA':
                 threading.Thread(
                     target=_start_b2bua_or_fake,
@@ -1221,7 +1228,7 @@ def recv_one_sip_message(sock, timeout):
     return None, 'incomplete_timeout', len(buf)
 
 
-def handle_tcp_client(client_sock, client_ip):
+def handle_tcp_client(client_sock, client_ip, transport='tcp'):
     session_id = uuid.uuid4().hex[:8]
     started_at = time.time()
     message_count = 0
@@ -1251,7 +1258,7 @@ def handle_tcp_client(client_sock, client_ip):
                 trace(session_id, client_ip, 'tcp_parse_invalid', index=message_count)
                 break
             trace(session_id, client_ip, 'tcp_parsed', index=message_count, method=req.get('method'), uri=req.get('uri'))
-            result = process_sip_request(req, client_ip)
+            result = process_sip_request(req, client_ip, client_port=client_sock.getpeername()[1], transport=transport)
             if result[0] == 'INVITE_FAKE':
                 try:
                     _send_invite_sequence(result[1], client_sock.sendall)
@@ -1293,7 +1300,7 @@ def handle_tcp_client(client_sock, client_ip):
         )
 
 
-def tcp_loop(sock):
+def tcp_loop(sock, ssl_context=None):
     while True:
         client, addr = sock.accept()
         client_ip = normalize_ip(addr[0])
@@ -1304,22 +1311,62 @@ def tcp_loop(sock):
             except Exception:
                 pass
             continue
-        threading.Thread(target=handle_tcp_client, args=(client, client_ip), daemon=True).start()
+        if ssl_context:
+            try:
+                client = ssl_context.wrap_socket(client, server_side=True)
+            except (ssl.SSLError, OSError):
+                trace(f"t{uuid.uuid4().hex[:8]}", client_ip, 'tls_handshake_failed')
+                try:
+                    client.close()
+                except Exception:
+                    pass
+                continue
+        transport = 'tls' if ssl_context else 'tcp'
+        threading.Thread(target=handle_tcp_client, args=(client, client_ip, transport), daemon=True).start()
 
 
-def start_honeypot():
+def start_honeypot(use_ssl=False, cert_path=None, key_path=None):
     _seed_dial_cache_from_db()
     if HAVE_B2BUA and sip_b2bua.capturing():
         sip_dial_profile.ensure_table()
-    udp_sock = create_dualstack_udp_listener(SIP_PORT)
-
     tcp_sock = create_dualstack_tcp_listener(SIP_PORT, backlog=200)
 
-    print(f'🚀 SIP Honeypot Active on Port {SIP_PORT} (UDP+TCP IPv4+IPv6). Collecting knocks...', flush=True)
+    ssl_context = None
+    if use_ssl:
+        cert_path = cert_path or SIP_TLS_CERT_PATH
+        key_path = key_path or SIP_TLS_KEY_PATH
+        ensure_self_signed_server_cert(
+            cert_path=cert_path,
+            key_path=key_path,
+            subject='/CN=localhost/O=Asterisk/C=US',
+            days=825,
+        )
+        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        ssl_context.load_cert_chain(certfile=cert_path, keyfile=key_path)
+
+    if use_ssl:
+        print(f'🚀 SIP/TLS Honeypot Active on Port {SIP_PORT} (TCP/TLS IPv4+IPv6). Collecting knocks...', flush=True)
+    else:
+        udp_sock = create_dualstack_udp_listener(SIP_PORT)
+        print(f'🚀 SIP Honeypot Active on Port {SIP_PORT} (UDP+TCP IPv4+IPv6). Collecting knocks...', flush=True)
+        threading.Thread(target=udp_loop, args=(udp_sock,), daemon=True).start()
+    tcp_loop(tcp_sock, ssl_context=ssl_context)
+
+
+def main():
+    global SIP_PORT
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--port', type=int, default=SIP_PORT)
+    parser.add_argument('--ssl', dest='ssl', action='store_true', default=None)
+    parser.add_argument('--no-ssl', dest='ssl', action='store_false')
+    parser.add_argument('--ssl-cert', default=SIP_TLS_CERT_PATH)
+    parser.add_argument('--ssl-key', default=SIP_TLS_KEY_PATH)
+    args = parser.parse_args()
+    SIP_PORT = args.port
 
-    threading.Thread(target=udp_loop, args=(udp_sock,), daemon=True).start()
-    tcp_loop(tcp_sock)
+    use_ssl = args.ssl if args.ssl is not None else (SIP_PORT == 5061)
+    start_honeypot(use_ssl=use_ssl, cert_path=args.ssl_cert, key_path=args.ssl_key)
 
 
 if __name__ == '__main__':
-    start_honeypot()
+    main()

protocols/sip.py

@@ -33,12 +33,17 @@ def db_update(data, cur, ctx):
     badge_color="#ff7a00",
     ui_order=60,
     honeypot_script="honeypots/sip_honeypot.py",
+    option_args={
+        "TLS": ["--ssl"],
+    },
     description="SIP allows humans and bots to make phone calls.",
     default_enabled_entries=["SIP"],
     supports_user_panel=False,
     supports_pass_panel=False,
     knock_table="knocks_sip",
     columns=[
+        Column("sip_port",              "INTEGER"),
+        Column("sip_transport",         "TEXT"),
         Column("sip_method",            "TEXT"),
         Column("sip_dial_string",       "TEXT"),
         Column("sip_dial_number",       "TEXT"),