nat-traversal

Building peer-to-peer applications

NAT Traversal Techniques

Use this skill when:

STUN (Session Traversal Utilities for NAT)

Discover Public IP and Port

import stun

# Discover public IP/port
nat_type, external_ip, external_port = stun.get_ip_info()

print(f"NAT Type: {nat_type}")
print(f"External IP: {external_ip}")
print(f"External Port: {external_port}")

# Use external IP/port for peer connection

STUN Server

# Using public STUN server
STUN_SERVER = "stun.l.google.com:19302"

import socket
import struct

def stun_request(server, port=19302):
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    # STUN Binding Request
    message = b'\x00\x01\x00\x00\x21\x12\xa4\x42' + b'\x00' * 12

    sock.sendto(message, (server, port))
    data, addr = sock.recvfrom(1024)

    # Parse STUN response for external IP/port
    return parse_stun_response(data)

TURN (Traversal Using Relays around NAT)

TURN Relay

# When direct connection fails, use TURN relay
import asyncio
from aioice import Candidate, Connection

async def connect_with_turn():
    connection = Connection(ice_controlling=True)

    # Add TURN server
    await connection.gather_candidates(
        stun_server=("stun.example.com", 3478),
        turn_server=("turn.example.com", 3478),
        turn_username="user",
        turn_password="pass"
    )

    # Exchange candidates with peer
    # Then connect through relay if needed

ICE (Interactive Connectivity Establishment)

ICE Implementation

import asyncio
from aiortc import RTCPeerConnection, RTCConfiguration, RTCIceServer

async def create_peer_connection():
    config = RTCConfiguration(
        iceServers=[
            RTCIceServer(urls="stun:stun.l.google.com:19302"),
            RTCIceServer(
                urls="turn:turn.example.com:3478",
                username="user",
                credential="pass"
            )
        ]
    )

    pc = RTCPeerConnection(configuration=config)

    # ICE candidate gathering
    @pc.on("icecandidate")
    def on_icecandidate(candidate):
        if candidate:
            # Send candidate to peer via signaling
            send_to_peer({"candidate": candidate})

    # Create offer
    offer = await pc.createOffer()
    await pc.setLocalDescription(offer)

    return pc

UDP Hole Punching

Simultaneous UDP

import socket
import threading

def udp_hole_punch(peer_addr, local_port=0):
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind(('0.0.0.0', local_port))

    # Send empty packet to create hole
    sock.sendto(b'', peer_addr)

    # Both peers send simultaneously
    # Creates mapping in both NATs

    return sock

# Usage (coordinated via signaling server)
# Peer A: sock = udp_hole_punch(('peer_b_ip', peer_b_port))
# Peer B: sock = udp_hole_punch(('peer_a_ip', peer_a_port))

# Then communicate directly

Signaling Server

WebSocket Signaling

import asyncio
import websockets
import json

peers = {}

async def signaling_handler(websocket, path):
    peer_id = None

    try:
        async for message in websocket:
            data = json.loads(message)

            if data['type'] == 'register':
                peer_id = data['peer_id']
                peers[peer_id] = websocket

            elif data['type'] == 'offer':
                # Forward offer to target peer
                target = data['target']
                if target in peers:
                    await peers[target].send(json.dumps({
                        'type': 'offer',
                        'from': peer_id,
                        'sdp': data['sdp']
                    }))

            elif data['type'] == 'answer':
                # Forward answer
                target = data['target']
                if target in peers:
                    await peers[target].send(json.dumps({
                        'type': 'answer',
                        'from': peer_id,
                        'sdp': data['sdp']
                    }))

    finally:
        if peer_id and peer_id in peers:
            del peers[peer_id]

# Run signaling server
async def main():
    async with websockets.serve(signaling_handler, "0.0.0.0", 8080):
        await asyncio.Future()

asyncio.run(main())

NAT Type Detection

Determine NAT Behavior

def detect_nat_type():
    # Full cone NAT: Port mapping same for all destinations
    # Restricted cone: IP filtering
    # Port restricted: IP+Port filtering
    # Symmetric: Different mapping per destination

    import stun
    nat_type, ext_ip, ext_port = stun.get_ip_info()

    return nat_type

# Traversal difficulty:
# Full cone: Easy (simple hole punching)
# Restricted cone: Medium (need simultaneous UDP)
# Port restricted: Hard (precise timing needed)
# Symmetric: Very hard (need TURN relay)

Related Skills