CGNat hole punching

README.md

NAT Hole Punching

I've never fully understood how NAT traversal works and I went on a dive today.

Run whatismyip.py on a public server. This plays the role of a STUN server.

Run easynat.py on a laptop behind a home router.

You could run hardnat.py also on a laptop behind a home router, but don't do it on the same home router because "hairpinning" is almost always broken. You could also run it on a public server, it should work there too. But to really push it put it in a hostile environment, like on a phone data. Cellphone data is always carrier-grade NATed, sometimes multiple times over. If you can get a p2p connection between a cellphone and a laptop you're doing pretty well.

This implements the birthday paradox port scan described by Tailscale. It spawns many sockets behind one NAT then tries to connect to many ports from the other without knowing what external ports the NAT assigned in hopes of stumbling across a port that was chosen. I found I had to bump it from their recommended 22048 to 256,3072 to really make it reliable.

It's more reliable to launch easynat before hardnat.

There's an asymmetry: easynat is meant for running on a "cone NAT", one where the inner port = the outer port. This isn't always available. It's apparently spec'd somewhere that UDP NATs should be "coned" but it's. I wonder if the asymmetry is necessary. Maybe both sides should open many ports, and each port should scan many ports.

I also wonder if the middle server can help more; perhaps it can give start/stop commands to synchronize the sides better.

Refs:

• https://tailscale.com/blog/nat-traversal-improvements-pt-1
• https://en.wikipedia.org/wiki/User_Datagram_Protocol
• https://tailscale.com/blog/how-nat-traversal-works#the-benefits-of-birthdays
• tonarino/innernet#115

easynat.py

#!/usr/bin/env python

import sys, os, select, socket
from random import shuffle
import time

target = 'localhost' # termux is dumb;
rendezvous_server = 'test.kousu.net'

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.bind(('', 0))
listenaddr = s.getsockname()[0] + ":" + str(s.getsockname()[1])
print("Listening on", s.getsockname())

known_hosts = {}
# find out our external IP and our target's
while target not in known_hosts:
  s.sendto(socket.gethostname().encode("ascii") + b":whatismyip", (rendezvous_server, 10000))
  payload, addr = s.recvfrom(1024)
  known_hosts = {host: (ip, int(port)) for (host, ip, port) in [line.split(":") for line in payload.decode("ascii").split("\n")]}
  time.sleep(3)
print("rdvz server sees", known_hosts)

# send an initial outgoing port knock
# hopefully, this makes the NAT record that
#
#  also, ignore port because we're going to port scan
target, _ = known_hosts[target]

# there's a race condition
# who should port scan firstek


break_outer = False
while not break_outer:
  PORTS = list(range(2**15, 2**16))[:int(2048*1.2)]
  shuffle(PORTS)
  for port in PORTS:
     print("knocking",(target,port),"from",s.getsockname())
     s.sendto(f'knockknock:{s.getsockname()}:{(target, port)}'.encode('ascii'), (target, port))

  while True:
    rs, _, _ = select.select([s], [], [], 3.0)
    if not rs: break
    payload, addr = s.recvfrom(1025)
    print("RECVB", payload, addr)
    if payload == b"SYN":
      s.sendto(b"ACK", addr)
      successful_connection = s
      successful_connection_target = addr
      break_outer = True
      break
  time.sleep(1)

successful_connection.sendto(b'hi there from allium', successful_connection_target)
print(successful_connection.recvfrom(1024))

hardnat.py

#!/usr/bin/env python

# https://tailscale.com/blog/how-nat-traversal-works#the-benefits-of-birthdays

import socket
from select import select
from random import shuffle
import sys
import time

target = 'laptop'
rendezvous_server = 'test.kousu.net'

# open up many ports
# we just need ONE working port
socks = []
for i in range(256):
  s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  s.bind(('', 0))  # bind to 0.0.0.0 at a random port
  print("listening on port", s.getsockname()[1])
  socks.append(s)

# find out our external IP and our target's
known_hosts = {}
while target not in known_hosts:
    socks[0].sendto(socket.gethostname().encode("ascii") + b":whatismyip", (rendezvous_server, 10000))
    payload, addr = socks[0].recvfrom(1024)
    known_hosts = {host: (ip, int(port)) for (host, ip, port) in [line.split(":") for line in payload.decode("ascii").split("\n")]}
    time.sleep(3)
print("rdvz server sees", known_hosts)

# send a port knock
# since the other side is an easy NAT
# 
target = known_hosts[target]

def find():
    while True:
        for s in socks:
          print('knocking', target, "from", s.getsockname())
          listenaddr = s.getsockname()[0] + ":" + str(s.getsockname()[1])
          s.sendto(f"knockknock:{s.getsockname()}:{target}".encode("ascii"), target)

        rs, _, _ = select(socks, [], [], 5.0)
        if rs:
            print(len(rs))
            for r in rs:
              payload, addr = r.recvfrom(1024)
              # attempt to confirm the connection
              r.sendto(b'SYN', addr)
              _rs, _, _ = select([r], [], [], 5)
              if _rs:
                _ack, _ackaddr = r.recvfrom(2048)
                if _ack == b"ACK" and _ackaddr == addr:
                  print("successful connetion:", payload, addr, r.getsockname())
                  global successful_connection, successful_connection_target
                  successful_connection = r
                  successful_connection_target = target
                  return

find()
successful_connection.sendto(b'hi there from behind the hard NAT', successful_connection_target)
print(successful_connection.recvfrom(1024))

whatismyip.py

#!/usr/bin/env python

import sys, os, select, socket

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.bind(('', 10000))
listenaddr = s.getsockname()[0] + ":" + str(s.getsockname()[1])
print("Listening on", s.getsockname())

known_hosts = {}

while True:
  rs, _, _ = select.select([s], [], [])
  if rs:
    payload, addr = s.recvfrom(1024)
    print(payload, addr)
    clientaddr = addr[0] + ":" + str(addr[1])
    host, cmd = payload.decode('ascii').split(':')
    known_hosts[host] = addr
    resp = None
    if cmd == 'whatismyip':
      resp = "\n".join(f"{host}:{addr[0]}:{addr[1]}" for host, addr in known_hosts.items()).encode("ascii")
    if resp:
      s.sendto(resp, addr)