272 lines
10 KiB
Nim
272 lines
10 KiB
Nim
{.passL: "-l crypto -l quicly -l picotls-core -l picotls-openssl".}
|
|
|
|
import asyncdispatch
|
|
import asyncnet
|
|
import net
|
|
import os
|
|
import quicly/quicly
|
|
import quicly/cid
|
|
import quicly/constants
|
|
import quicly/defaults
|
|
import quicly/recvstate
|
|
import quicly/sendstate
|
|
import quicly/streambuf
|
|
import picotls/picotls
|
|
import picotls/openssl as ptls_openssl
|
|
import strformat
|
|
|
|
from nativesockets import SockAddr, Sockaddr_storage, SockLen, getHostByName
|
|
from openssl import DLLSSLName, EVP_PKEY
|
|
from posix import IOVec
|
|
from strutils import parseUInt
|
|
|
|
const serverCertChainPath = "./certs/server-certchain.pem"
|
|
const serverKeyPath = "./certs/server-cert.key"
|
|
const clientCertChainPath = "./certs/server-certchain.pem"
|
|
const clientKeyPath = "./certs/server-cert.key"
|
|
|
|
type
|
|
QuicP2PContext = ref object
|
|
sock: AsyncSocket
|
|
streamOpen: quicly_stream_open_t
|
|
nextCid: quicly_cid_plaintext_t
|
|
signCertificate: ptls_openssl_sign_certificate_t
|
|
tlsCtx: ptls_context_t
|
|
quiclyCtx: quicly_context_t
|
|
connections: seq[ptr quicly_conn_t]
|
|
|
|
proc PEM_read_PrivateKey(fp: File, x: ptr EVP_PKEY,
|
|
cb: proc(buf: cstring, size: cint, rwflag: cint, u: pointer): cint {.cdecl.},
|
|
u: pointer): EVP_PKEY
|
|
{.importc, dynlib: DLLSSLName, cdecl.}
|
|
|
|
proc EVP_PKEY_free(key: EVP_PKEY) {.importc, dynlib: DLLSSLName, cdecl.}
|
|
|
|
proc getRelativeTimeout(ctx: QuicP2PContext): int32 =
|
|
## Obtain the absolute int64 timeout from quicly and convert it to the
|
|
## relative int32 timeout expected by poll.
|
|
result = 0
|
|
var nextTimeout = int64.high
|
|
var now = ctx.quiclyCtx.now.cb(ctx.quiclyCtx.now)
|
|
for c in ctx.connections:
|
|
let connTimeout = quicly_get_first_timeout(c)
|
|
if connTimeout < nextTimeout:
|
|
nextTimeout = connTimeout
|
|
if now < nextTimeout:
|
|
let delta = nextTimeout - now
|
|
result = min(delta, int32.high).int32
|
|
|
|
proc onStopSending(stream: ptr quicly_stream_t, err: cint) {.cdecl.} =
|
|
echo "onStopSending"
|
|
discard quicly_close(stream.conn, 0x30000, "")
|
|
|
|
proc onReceiveReset(stream: ptr quicly_stream_t, err: cint) {.cdecl.} =
|
|
echo "onReceiveReset"
|
|
discard quicly_close(stream.conn, 0x30000, "")
|
|
|
|
proc onServerReceive(stream: ptr quicly_stream_t, offset: csize_t, src: pointer,
|
|
len: csize_t) {.cdecl.} =
|
|
echo "onServerReceive"
|
|
if quicly_streambuf_ingress_receive(stream, offset, src, len) != 0:
|
|
return
|
|
let input = quicly_streambuf_ingress_get(stream)
|
|
if quicly_sendstate_is_open(addr stream.sendstate) != 0 and input.len > 0:
|
|
discard quicly_streambuf_egress_write(stream, input.base, input.len)
|
|
if quicly_recvstate_transfer_complete(addr stream.recvstate) != 0:
|
|
discard quicly_streambuf_egress_shutdown(stream)
|
|
quicly_streambuf_ingress_shift(stream, input.len)
|
|
|
|
proc onClientReceive(stream: ptr quicly_stream_t, offset: csize_t,
|
|
src: pointer, len: csize_t) {.cdecl.} =
|
|
echo "onClientReceive"
|
|
if quicly_streambuf_ingress_receive(stream, offset, src, len) != 0:
|
|
return
|
|
let input = quicly_streambuf_ingress_get(stream)
|
|
let msg = newString(input.len)
|
|
copyMem(msg.cstring, input.base, input.len)
|
|
echo "received message from server: ", msg
|
|
if quicly_recvstate_transfer_complete(addr stream.recvstate) != 0:
|
|
discard quicly_close(stream.conn, 0, "")
|
|
quicly_streambuf_ingress_shift(stream, input.len)
|
|
|
|
var streamCallbacksServer = quicly_stream_callbacks_t(
|
|
on_destroy: quicly_streambuf_destroy,
|
|
on_send_shift: quicly_streambuf_egress_shift,
|
|
on_send_emit: quicly_streambuf_egress_emit,
|
|
on_send_stop: onStopSending,
|
|
on_receive: onServerReceive,
|
|
on_receive_reset: onReceiveReset)
|
|
|
|
var streamCallbacksClient = quicly_stream_callbacks_t(
|
|
on_destroy: quicly_streambuf_destroy,
|
|
on_send_shift: quicly_streambuf_egress_shift,
|
|
on_send_emit: quicly_streambuf_egress_emit,
|
|
on_send_stop: onStopSending,
|
|
on_receive: onClientReceive,
|
|
on_receive_reset: onReceiveReset)
|
|
|
|
proc usage() =
|
|
echo &"usage in server mode: {paramStr(0)} LISTEN_PORT"
|
|
echo &"usage in client mode: {paramStr(0)} SERVER_HOSTNAME SERVER_PORT"
|
|
|
|
proc onServerStreamOpen(self: ptr quicly_stream_open_t,
|
|
stream: ptr quicly_stream_t): cint {.cdecl.} =
|
|
echo "onServerStreamOpen"
|
|
result = quicly_streambuf_create(stream, sizeof(quicly_streambuf_t).csize_t)
|
|
stream.callbacks = addr streamCallbacksServer
|
|
|
|
proc onClientStreamOpen(self: ptr quicly_stream_open_t,
|
|
stream: ptr quicly_stream_t): cint {.cdecl.} =
|
|
echo "onClientStreamOpen"
|
|
result = quicly_streambuf_create(stream, sizeof(quicly_streambuf_t).csize_t)
|
|
stream.callbacks = addr streamCallbacksClient
|
|
let msg = "hello server"
|
|
discard quicly_streambuf_egress_write(stream, msg.cstring, msg.len().csize_t)
|
|
|
|
proc handleMsg(ctx: QuicP2PContext, msg: string, peerAddr: ptr SockAddr,
|
|
isServer: bool) =
|
|
var offset: csize_t = 0
|
|
while offset < msg.len().csize_t:
|
|
var decoded: quicly_decoded_packet_t
|
|
let decodeResult = quicly_decode_packet(addr ctx.quiclyCtx, addr decoded,
|
|
cast[ptr uint8](msg.cstring),
|
|
msg.len().csize_t, addr offset)
|
|
if decode_result == csize_t.high:
|
|
return
|
|
var conn: ptr quicly_conn_t = nil
|
|
for c in ctx.connections:
|
|
if quicly_is_destination(c, nil, peerAddr, addr decoded) != 0:
|
|
conn = c
|
|
break
|
|
if conn != nil:
|
|
echo "quicly_receive"
|
|
discard quicly_receive(conn, nil, peerAddr, addr decoded)
|
|
elif isServer:
|
|
echo "quicly_accept"
|
|
discard quicly_accept(addr conn, addr ctx.quiclyCtx, nil, peerAddr,
|
|
addr decoded, nil, addr ctx.nextCid, nil)
|
|
ctx.connections.add(conn)
|
|
|
|
proc initContext(sock: AsyncSocket, certChainPath: string, keyPath: string,
|
|
streamOpenCb: typeof(quicly_stream_open_t.cb)):
|
|
QuicP2PContext =
|
|
|
|
var tlsCtx = ptls_context_t(randomBytes: ptls_openssl_random_bytes,
|
|
getTime: addr ptls_get_time,
|
|
keyExchanges: ptls_openssl_key_exchanges,
|
|
cipherSuites: ptls_openssl_cipher_suites)
|
|
quicly_amend_ptls_context(addr tlsCtx)
|
|
result = QuicP2PContext(sock: sock,
|
|
streamOpen: quicly_stream_open_t(cb: streamOpenCb),
|
|
tlsCtx: tlsCtx, quiclyCtx: quicly_spec_context)
|
|
result.quiclyCtx.tls = addr result.tlsCtx
|
|
result.quiclyCtx.stream_open = addr result.streamOpen
|
|
if ptls_load_certificates(addr result.tlsCtx, certChainPath.cstring) != 0:
|
|
raise newException(ValueError, &"cannot load certificate chain {certChainPath}")
|
|
let pKeyFile = open(keyPath)
|
|
let privateKey = PEM_read_PrivateKey(pkeyFile, nil, nil, nil)
|
|
pkeyFile.close()
|
|
if privateKey == nil:
|
|
raise newException(ValueError, &"cannot load private key {keyPath}")
|
|
discard ptls_openssl_init_sign_certificate(addr result.signCertificate, privateKey)
|
|
EVP_PKEY_free(privateKey)
|
|
result.tlsCtx.signCertificate = addr result.signCertificate.super
|
|
|
|
proc sendPackets(ctx: QuicP2PContext) =
|
|
if ctx.connections.len == 0:
|
|
return
|
|
let conns = ctx.connections.deepCopy()
|
|
for i in 0 .. conns.len - 1:
|
|
var srcAddr, dstAddr: quicly_address_t
|
|
var dgrams: array[10, IOVec]
|
|
var dgramCount = dgrams.len().csize_t
|
|
var dgramsBuf = newString(dgramCount * ctx.quiclyCtx.transport_params.max_udp_payload_size)
|
|
let sendResult = quicly_send(conns[i], addr dstAddr, addr srcAddr,
|
|
addr dgrams[0], addr dgramCount,
|
|
addr dgramsBuf[0], dgramsBuf.len().csize_t)
|
|
case sendResult:
|
|
of 0:
|
|
if dgramCount > 0:
|
|
echo &"sending {dgramCount} datagrams"
|
|
for j in 0 .. dgramCount - 1:
|
|
var sockLen = quicly_get_socklen(addr dstAddr.sa)
|
|
# TODO: replace asyncdispatch.sendTo with asyncnet.sendTo (Nim 1.4 required)
|
|
asyncCheck sendTo(ctx.sock.getFd().AsyncFD, dgrams[j].iov_base,
|
|
dgrams[j].iov_len.int, addr dstAddr.sa, sockLen)
|
|
of QUICLY_ERROR_FREE_CONNECTION:
|
|
let c = ctx.connections[i]
|
|
ctx.connections.del(i)
|
|
quicly_free(c)
|
|
else:
|
|
raise newException(ValueError, &"quicly_send returned {sendResult}")
|
|
|
|
proc receive(ctx: QuicP2PContext, isServer: bool) {.async.} =
|
|
while true:
|
|
# TODO: replace asyncdispatch.recvFromInto with asyncnet.recvFrom (Nim 1.4 required)
|
|
var msg = newString(BufferSize)
|
|
var peerAddr: Sockaddr_storage
|
|
var peerAddrLen = SockLen(sizeof(peerAddr))
|
|
let msgLen = await recvFromInto(ctx.sock.getFd().AsyncFD, msg.cstring,
|
|
msg.len, cast[ptr SockAddr](addr peerAddr),
|
|
addr peerAddrLen)
|
|
echo &"received {msgLen} bytes"
|
|
msg.setLen(msgLen)
|
|
if msg.len > 0:
|
|
handleMsg(ctx, msg, cast[ptr SockAddr](addr peerAddr), isServer)
|
|
|
|
proc main() =
|
|
var ctx: QuicP2PContext
|
|
let sock = newAsyncSocket(sockType = SOCK_DGRAM, protocol = IPPROTO_UDP,
|
|
buffered = false)
|
|
case paramCount():
|
|
of 1:
|
|
let portNumber = paramStr(1).parseUInt()
|
|
if portNumber > uint16.high:
|
|
usage()
|
|
quit(1)
|
|
sock.bindAddr(Port(portNumber))
|
|
ctx = initContext(sock, serverCertChainPath, serverKeyPath,
|
|
onServerStreamOpen)
|
|
asyncCheck receive(ctx, true)
|
|
|
|
of 2:
|
|
let hostname = paramStr(1)
|
|
let portNumber = paramStr(2).parseUInt()
|
|
if portNumber > uint16.high:
|
|
usage()
|
|
quit(1)
|
|
ctx = initContext(sock, clientCertChainPath, clientKeyPath,
|
|
onClientStreamOpen)
|
|
var conn: ptr quicly_conn_t
|
|
let hostent = getHostByName(hostname)
|
|
if hostent.addrList.len == 0:
|
|
echo "cannot resolve hostname ", hostname
|
|
quit(2)
|
|
var destAddr: Sockaddr_storage
|
|
var sockLen: SockLen
|
|
toSockAddr(parseIpAddress(hostent.addrList[0]), Port(portNumber), destAddr,
|
|
sockLen)
|
|
let addressToken = ptls_iovec_init(nil, 0)
|
|
let connectResult = quicly_connect(addr conn, addr ctx.quiclyCtx,
|
|
hostname.cstring, addr destAddr, nil,
|
|
addr ctx.nextCid, addressToken, nil, nil)
|
|
if connectResult != 0:
|
|
echo "quicly_connect failed: ", connectResult
|
|
quit(3)
|
|
ctx.connections.add(conn)
|
|
var stream: ptr quicly_stream_t
|
|
discard quicly_open_stream(conn, addr stream, 0)
|
|
asyncCheck receive(ctx, false)
|
|
|
|
else:
|
|
usage()
|
|
quit(1)
|
|
|
|
while true:
|
|
let nextTimeout = ctx.getRelativeTimeout()
|
|
poll(nextTimeout)
|
|
ctx.sendPackets()
|
|
|
|
when isMainModule:
|
|
main()
|