A peer-to-peer prototype application using QUIC
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

147 lines
5.7 KiB

import asyncdispatch, asyncnet, net, port_prediction, strformat
from nativesockets import
SockAddr,
SockAddr_storage,
SockLen,
getSockOptInt,
setSockOptInt
from sequtils import any, map
type
Attempt* = object
## A hole punching attempt.
socks*: seq[AsyncSocket]
dstIp*: IpAddress
dstPorts*: seq[Port]
future*: Future[(AsyncSocket, Port)]
Puncher* = ref object
socks*: seq[AsyncSocket]
natProps*: NatProperties
attempts*: seq[Attempt]
PunchHoleError* = object of ValueError
var IPPROTO_IP {.importc: "IPPROTO_IP", header: "<netinet/in.h>".}: cint
var IP_TTL {.importc: "IP_TTL", header: "<netinet/in.h>".}: cint
const Timeout = 3000
const InitiatorMaxOutPacketCount = 1000
const ResponderMaxOutPacketCount = 70
const MaxOutPacketCount = max(InitiatorMaxOutPacketCount,
ResponderMaxOutPacketCount)
proc srcPort(sock: AsyncSocket): Port =
result = sock.getLocalAddr[1]
proc `==`(a, b: Attempt): bool =
## ``==`` for hole punching attempts.
##
## Two hole punching attempts are considered equal if their ``dstIp`` is
## equal and their ``dstPorts`` overlap.
a.dstIp == b.dstIp and a.dstPorts.any(proc (p: Port): bool = p in b.dstPorts)
proc initPuncher*(sock: AsyncSocket, probedSrcPorts: seq[Port]): Puncher =
# TODO: determine IP_TTL
let (_, primarySrcPort) = sock.getLocalAddr()
let natProps = getNatProperties(primarySrcPort, probedSrcPorts)
result = Puncher(socks: @[sock], natProps: natProps)
if result.natProps.natType == SymmetricRandom:
# our NAT is of the evil symmetric type with random port allocation. We are
# trying to help the other peer by allocating a lot of auxillary sockets
# for punching more holes
result.socks.setLen(MaxOutPacketCount)
for i in 1 .. MaxOutPacketCount - 1:
result.socks[i] = newAsyncSocket(sockType = SOCK_DGRAM,
protocol = IPPROTO_UDP, buffered = false)
result.socks[i].bindAddr(Port(0))
proc primarySrcPort*(puncher: Puncher): Port =
puncher.socks[0].srcPort
proc punch(puncher: Puncher, peerIp: IpAddress, peerPort: Port,
peerProbedPorts: seq[Port], isInitiating: bool, msg: string):
Future[Attempt] {.async.} =
let punchFuture = newFuture[(AsyncSocket, Port)]("punch")
let peerNatProps = getNatProperties(peerPort, peerProbedPorts)
let maxOutPacketCount = if isInitiating:
InitiatorMaxOutPacketCount
else:
ResponderMaxOutPacketCount
var sockCount = 1
if puncher.natProps.natType == SymmetricRandom:
if peerNatProps.natType == SymmetricRandom:
# If both peers are behind a SymmetricRandom NAT we give up.
raise newException(PunchHoleError,
"both peers behind symmetric NAT with random port allocation")
sockCount = maxOutPacketCount
let predictedDstPorts = predictPortRange(peerNatProps, maxOutPacketCount.uint16)
result = Attempt(dstIp: peerIp, dstPorts: predictedDstPorts,
future: punchFuture)
if puncher.attempts.contains(result):
raise newException(PunchHoleError,
"hole punching to given destination already active")
puncher.attempts.add(result)
let srcPorts = puncher.socks[0 .. sockCount - 1].map(srcPort)
echo &"sending msg {msg} to {peerIp}, srcPorts: {srcPorts}, dstPorts: {result.dstPorts}"
var peerAddr: Sockaddr_storage
var peerSockLen: SockLen
try:
var defaultTTL: int
for i in 0 .. sockCount - 1:
let sock = puncher.socks[i]
if isInitiating:
defaultTTL = sock.getFd.getSockOptInt(IPPROTO_IP, IP_TTL)
sock.getFd.setSockOptInt(IPPROTO_IP, IP_TTL, 2)
for dstPort in result.dstPorts:
toSockAddr(result.dstIp, dstPort, peerAddr, peerSockLen)
# TODO: replace asyncdispatch.sendTo with asyncnet.sendTo (Nim 1.4 required)
await sendTo(sock.getFd.AsyncFD, msg.cstring, msg.len,
cast[ptr SockAddr](addr peerAddr), peerSockLen)
if isInitiating:
sock.getFd.setSockOptInt(IPPROTO_IP, IP_TTL, defaultTTL)
except OSError as e:
raise newException(PunchHoleError, e.msg)
proc initiate*(puncher: Puncher, peerIp: IpAddress, peerPort: Port,
peerProbedPorts: seq[Port]): Future[Attempt] =
punch(puncher, peerIp, peerPort, peerProbedPorts, true, "SYN")
proc respond*(puncher: Puncher, peerIp: IpAddress, peerPort: Port,
peerProbedPorts: seq[Port]): Future[Attempt] =
punch(puncher, peerIp, peerPort, peerProbedPorts, false, "ACK")
proc finalize*(attempt: Attempt): Future[(AsyncSocket, Port)] {.async.} =
await attempt.future or sleepAsync(Timeout)
if attempt.future.finished:
result = attempt.future.read()
else:
raise newException(PunchHoleError, "timeout")
proc handleMsg*(puncher: Puncher, msg: string, sock: AsyncSocket,
peerIp: IpAddress, peerPort: Port) =
## Handles an incoming UDP message which may complete the Futures returned by
## ``initiate`` and ``respond``.
if msg == "SYN":
# We received a SYN packet. We ignore it because we expected it to be
# filtered by our NAT.
return
let query = Attempt(dstIp: peerIp, dstPorts: @[peerPort])
let i = puncher.attempts.find(query)
if i != -1:
if msg == "ACK":
echo &"handling ACK message from {peerIp}:{peerPort}"
else:
echo &"handling QUIC message from {peerIp}:{peerPort}"
puncher.attempts[i].future.complete((sock, peerPort))
puncher.attempts.del(i)
else:
echo &"received unexpected packet from {peerIp}:{peerPort}"
proc handleMsg*(puncher: Puncher, msg: string, sock: AsyncSocket,
peerAddr: SockAddr | Sockaddr_storage, peerSockLen: SockLen) =
var peerIp: IpAddress
var peerPort: Port
fromSockAddr(peerAddr, peerSockLen, peerIp, peerPort)
handleMsg(puncher, msg, sock, peerIp, peerPort)