punchd/examples/app/app.nim

268 lines
9.5 KiB
Nim

import asyncdispatch, asyncnet, os, strformat, strutils, tables
from nativeSockets import Domain, SockType, Protocol
from net import IpAddress, Port, isIpAddress, `$`
from posix import CMSG_SPACE
import asyncutils
import ../../message
import random
const rendezvousServers: seq[tuple[hostname: string, port: Port]] = @[
("strangeplace.net", Port(5320)),
("ulrich.earth", Port(5320))
]
type
PunchdResponse = Future[tuple[msgContent: string, sock: AsyncSocket]]
PunchdProgressCb = proc (future: PunchdResponse, msgContent: string) {.async.}
OutgoingPunchdMessage = object
future: PunchdResponse
progressCb: PunchdProgressCb
PunchdConnection = object
sock: AsyncSocket
outMessages: TableRef[string, OutgoingPunchdMessage]
inConnections: FutureStream[AsyncSocket]
ServerConnection = object
sock: AsyncSocket
outMessages: TableRef[string, Future[string]]
peerNotifications: FutureStream[string]
probedIp: IpAddress
srcPort: Port
probedSrcPorts: seq[Port]
# Punchd messages
Progress* = object
command: string
args: string
# Server messages
OkGetPeerinfo* = object
ip: string
localPort: Port
probedPorts: seq[Port]
OkGetEndpoint* = object
ip: IpAddress
port: Port
NotifyPeer* = object
sender: string
recipient: string
technique: string
srcIp: IpAddress
srcPort: Port
probedSrcPorts: seq[Port]
dstIp: IpAddress
dstPort: Port
probedDstPorts: seq[Port]
extraArgs: string
# Exceptions
PunchdError = object of ValueError # FIXME: not used yet
ServerError = object of ValueError
proc usage() =
echo &"usage: {paramStr(0)} PEER_ID [OTHER_PEER_ID]"
proc handleServerMessages(conn: ServerConnection) {.async.} =
while true:
let line = await conn.sock.recvLine(maxLength = 400)
let args = line.parseArgs(3, 1)
case args[0]:
of "ok":
let future = conn.outMessages[args[1]]
conn.outMessages.del(args[1])
future.complete(args[2])
of "error":
let future = conn.outMessages[args[1]]
conn.outMessages.del(args[1])
future.fail(newException(ServerError, args[2]))
of "notify-peer":
asyncCheck conn.peerNotifications.write(line.substr(args[0].len + 1))
else:
raise newException(ValueError, "invalid server message")
proc handlePunchdMessages(conn: PunchdConnection) {.async.} =
while true:
let fd = conn.sock.getFd.AsyncFD
let cmsgSize = CMSG_SPACE(sizeof(AsyncFD).csize_t)
let resp = await fd.asyncRecvMsg(400, cmsgSize)
let line = resp.data.strip(leading = false, trailing = true, chars = {'\n'})
echo "received punchd message: ", line
let args = line.parseArgs(3, 1)
case args[0]:
of "ok":
let outMsg = conn.outMessages[args[1]]
conn.outMessages.del(args[1])
if resp.cmsgs.len < 1:
echo "no cmsg"
raise newException(ValueError, "invalid punchd message")
let sock = newAsyncSocket(resp.cmsgs[0].getFd)
register(sock.getFd.AsyncFD)
outMsg.future.complete((args[2], sock))
of "error":
let outMsg = conn.outMessages[args[1]]
conn.outMessages.del(args[1])
outMsg.future.fail(newException(PunchdError, args[2]))
of "progress":
let outMsg = conn.outMessages[args[1]]
asyncCheck outMsg.progressCb(outMsg.future, args[2])
else:
raise newException(ValueError, "invalid punchd message")
proc sendRequest(connection: PunchdConnection, command: string, content: string,
progressCb: PunchdProgressCb = nil): PunchdResponse =
result = newFuture[PunchdResponse.T]("sendRequest")
let id = $rand(uint32)
asyncCheck connection.sock.send(&"{command}|{id}|{content}\n")
let outMsg = OutgoingPunchdMessage(future: result, progressCb: progressCb)
echo "id = ", id
connection.outMessages[id] = outMsg
proc sendRequest(connection: ServerConnection, command: string,
content: string): Future[string] =
result = newFuture[string]("sendRequest")
let id = $rand(uint32)
var request: string
if content.len != 0:
request = &"{command}|{id}|{content}\n"
else:
request = &"{command}|{id}\n"
asyncCheck connection.sock.send(request)
connection.outMessages[id] = result
proc acceptConnection(punchdConn: PunchdConnection, command: string,
msgContent: string) {.async.} =
echo "accepting connection ", msgContent
let resp = await punchdConn.sendRequest(command, msgContent)
asyncCheck punchdConn.inConnections.write(resp.sock)
proc handlePeerNotifications(serverConn: ServerConnection,
punchdConn: PunchdConnection,
peerId: string) {.async.} =
while true:
let (hasData, data) = await serverConn.peerNotifications.read
if not hasData:
break
try:
let msg = parseMessage[NotifyPeer](data)
# FIXME: check if we want to receive messages from the sender
echo "received message from ", msg.sender
let probedSrcPorts = msg.probedSrcPorts.join(",")
let probedDstPorts = msg.probedDstPorts.join(",")
let req = &"{msg.technique}|{msg.srcIp}|{msg.srcPort}|{probedSrcPorts}|{msg.dstIp}|{msg.dstPort}|{probedDstPorts}|{msg.extraArgs}"
asyncCheck acceptConnection(punchdConn, "respond", req)
except ValueError as e:
echo e.msg
discard
proc punchHole(punchdConn: PunchdConnection, serverConn: ServerConnection,
peerId: string, otherPeerId: string, technique: string):
Future[AsyncSocket] {.async.} =
let sResp = await serverConn.sendRequest("get-peerinfo", otherPeerId)
let peerInfo = parseMessage[OkGetPeerinfo](sResp)
proc progressCb(future: PunchdResponse, msgContent: string) {.async.} =
try:
let parsedResp = parseMessage[Progress](msgContent)
let req = &"{peerId}|{otherPeerId}|{parsedResp.command}|{parsedResp.args}"
discard await serverConn.sendRequest("notify-peer", req)
except ServerError as e:
future.fail(e)
except ValueError as e:
future.fail(e)
let myProbedPorts = serverConn.probedSrcPorts.join(",")
let probedPeerPorts = peerInfo.probedPorts.join(",")
let req = &"{technique}|{serverConn.probedIp}|{serverConn.srcPort}|{myProbedPorts}|{peerInfo.ip}|{peerInfo.localPort}|{probedPeerPorts}"
let pResp = await punchdConn.sendRequest("initiate", req, progressCb)
result = pResp.sock
proc getEndpoint(srcPort: Port, serverHostname: string, serverPort: Port):
Future[OkGetEndpoint] {.async.} =
let sock = newAsyncSocket()
var failCount = 0
while true:
try:
sock.bindAddr(srcPort)
break
except OSError as e:
if failCount == 3:
raise e
failCount.inc
await sleepAsync(100)
await sock.connect(serverHostname, serverPort)
let id = rand(uint32)
await sock.send(&"get-endpoint|{id}\n")
let line = await sock.recvLine(maxLength = 400)
let args = line.parseArgs(3)
assert(args[0] == "ok")
assert(args[1] == $id)
result = parseMessage[OkGetEndpoint](args[2])
let emptyLine = await sock.recvLine(maxLength = 400)
assert(emptyLine.len == 0)
sock.close()
proc initServerConnection(srcPort: Port): Future[ServerConnection] {.async.} =
result.srcPort = srcPort
for r in rendezvousServers:
let endpoint = await getEndpoint(srcPort, r.hostname, r.port)
# FIXME: what if we get get different IPs from different servers
result.probedIp = endpoint.ip
result.probedSrcPorts.add(endpoint.port)
result.sock = await asyncnet.dial(rendezvousServers[0].hostname,
rendezvousServers[0].port)
result.outMessages = newTable[string, Future[string]]()
result.peerNotifications = newFutureStream[string]("initServerConnection")
proc runApp(peerId: string, otherPeerId: string = "") {.async.} =
randomize() # initialize random number generator
var punchdConn = PunchdConnection()
punchdConn.sock = newAsyncSocket(AF_UNIX, SOCK_STREAM, IPPROTO_IP)
punchdConn.outMessages = newTable[string, OutgoingPunchdMessage]()
punchdConn.inConnections = newFutureStream[AsyncSocket]("runApp")
await punchdConn.sock.connectUnix("/tmp/punchd.socket")
asyncCheck handlePunchdMessages(punchdConn)
let srcPort = rand(Port(1024) .. Port.high)
if otherPeerId.len == 0:
# register and wait for connections
echo &"init server connection, source port: {srcPort}"
let serverConn = await initServerConnection(srcPort)
asyncCheck handleServerMessages(serverConn)
asyncCheck handlePeerNotifications(serverConn, punchdConn, peerId)
let probedPorts = serverConn.probedSrcPorts.join(",")
let req = &"{peerId}|{serverConn.probedIp}|{serverConn.srcPort}|{probedPorts}"
echo "registering: ", req
discard await serverConn.sendRequest("register", req)
while true:
let (hasSock, sock) = await punchdConn.inConnections.read
if not hasSock:
break
echo "accepted!"
let msg = await sock.recv(4)
echo "received message: ", msg
await sock.send("pong")
sock.close()
else:
# initiate a new connection
var serverConn = await initServerConnection(srcPort)
asyncCheck handleServerMessages(serverConn)
let sock = await punchHole(punchdConn, serverConn, peerId, otherPeerId,
"udp")
echo "connected!"
await sock.send("ping")
let msg = await sock.recv(4)
echo "received message: ", msg
sock.close()
proc main() =
if paramCount() < 1 or paramCount() > 2:
usage()
quit(1)
if paramCount() == 2:
waitFor runApp(paramStr(1), paramStr(2))
else:
waitFor runApp(paramStr(1))
when isMainModule:
main()