diff --git a/lsquic.nimble b/lsquic.nimble index fef1735..2af469c 100644 --- a/lsquic.nimble +++ b/lsquic.nimble @@ -9,7 +9,7 @@ installFiles = @["lsquic.nim"] requires "nim >= 2.0.0" requires "zlib" requires "stew >= 0.4.0" -requires "chronos >= 4.0.4" +requires "https://github.com/status-im/nim-chronos#012cd2c92e9e62b11976ad15b5fd1404e279cf9a" requires "unittest2" requires "chronicles >= 0.11.0" requires "https://github.com/vacp2p/nim-boringssl >= 0.0.11" diff --git a/lsquic/context/context.nim b/lsquic/context/context.nim index a565c5b..6191c32 100644 --- a/lsquic/context/context.nim +++ b/lsquic/context/context.nim @@ -40,6 +40,56 @@ type running*: bool ownedCids: HashSet[CidKey] + when defined(windows): + wsaSendMsg: osdefs.LPFN_WSASENDMSG + wsaRecvMsg: osdefs.LPFN_WSARECVMSG + +when defined(windows): + proc resolveExtension( + fd: SocketHandle, guid: var GUID, extension: var pointer + ): bool {.raises: [].} = + var bytesReturned: DWORD + wsaIoctl( + fd, + osdefs.SIO_GET_EXTENSION_FUNCTION_POINTER, + addr guid, + DWORD(sizeof(guid)), + addr extension, + DWORD(sizeof(extension)), + addr bytesReturned, + nil, + nil, + ) == 0 + + proc initPacketIo*(ctx: QuicContext, fd: SocketHandle): bool {.raises: [].} = + var + sendExtension, recvExtension: pointer + sendGuid = osdefs.WSAID_WSASENDMSG + recvGuid = osdefs.WSAID_WSARECVMSG + if not resolveExtension(fd, sendGuid, sendExtension) or + not resolveExtension(fd, recvGuid, recvExtension): + return false + ctx.wsaSendMsg = cast[osdefs.LPFN_WSASENDMSG](sendExtension) + ctx.wsaRecvMsg = cast[osdefs.LPFN_WSARECVMSG](recvExtension) + true + + proc sendMsg*( + ctx: QuicContext, fd: SocketHandle, msg: ptr osdefs.WSAMSG, bytesSent: ptr DWORD + ): cint {.inline, raises: [].} = + if ctx.wsaSendMsg.isNil: + return -1 + ctx.wsaSendMsg(fd, msg, DWORD(0), bytesSent, nil, nil) + + proc recvMsg*( + ctx: QuicContext, + fd: SocketHandle, + msg: ptr osdefs.WSAMSG, + bytesReceived: ptr DWORD, + ): cint {.inline, raises: [].} = + if ctx.wsaRecvMsg.isNil: + return -1 + ctx.wsaRecvMsg(fd, msg, bytesReceived, nil, nil) + func hash*(cid: CidKey): Hash = var h = hash(cid.len) for i in 0 ..< cid.len.int: diff --git a/lsquic/context/io.nim b/lsquic/context/io.nim index 3d4ab1d..7569638 100644 --- a/lsquic/context/io.nim +++ b/lsquic/context/io.nim @@ -16,10 +16,15 @@ when not defined(windows): logScope: topics = "lsquic" -const MaxBatch = 1024 - ## Upper bound on the stack WSABUF array in the Windows send path (`sendPacketsOut`). - ## In practice `iovlen` is small; if it exceeds this bound, `sendPacketsOut` - ## falls back to a heap-allocated seq. +type ControlBuffer {.union.} = object + alignment: uint + data: array[128, byte] + +when defined(windows): + const MaxBatch = 1024 + ## Upper bound on the stack WSABUF array in the Windows send path (`sendPacketsOut`). + ## In practice `iovlen` is small; if it exceeds this bound, `sendPacketsOut` + ## falls back to a heap-allocated seq. when defined(linux): {.passc: "-D_GNU_SOURCE".} @@ -34,26 +39,212 @@ when defined(linux): sockfd: SocketHandle, msgvec: ptr MMsgHdr, vlen: cuint, flags: cint ): cint {.importc, header: "".} +when defined(linux) or defined(macosx): + when defined(macosx): + {.passc: "-D__APPLE_USE_RFC_3542".} + + type + InPktInfo {.importc: "struct in_pktinfo", header: "", bycopy.} = object + ipi_ifindex: cint + ipi_spec_dst: InAddr + ipi_addr: InAddr + + In6PktInfo {.importc: "struct in6_pktinfo", header: "", bycopy.} = object + ipi6_addr: In6Addr + ipi6_ifindex: cuint + + var + IP_PKTINFO {.importc, header: "".}: cint + IPV6_PKTINFO {.importc, header: "".}: cint + + when defined(macosx): + var IP_RECVDSTADDR {.importc, header: "".}: cint + +when not defined(windows): + proc prepareSourceAddr( + localSa: ptr SockAddr, control: var ControlBuffer, msg: var Tmsghdr + ) = + when defined(linux) or defined(macosx): + var local = localSa.toTransportAddress() + if local.isV4Mapped(): + local = local.toIPv4() + if local.isAnyLocal(): + return + + zeroMem(addr control.data[0], control.data.len) + msg.msg_control = addr control.data[0] + + let cmsg = cast[ptr Tcmsghdr](msg.msg_control) + if local.family == AddressFamily.IPv4: + msg.msg_controllen = + typeof(msg.msg_controllen)(CMSG_SPACE(sizeof(InPktInfo).csize_t)) + cmsg.cmsg_len = typeof(cmsg.cmsg_len)(CMSG_LEN(sizeof(InPktInfo).csize_t)) + cmsg.cmsg_level = IPPROTO_IP + cmsg.cmsg_type = IP_PKTINFO + let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) + copyMem( + addr info.ipi_spec_dst, unsafeAddr local.address_v4[0], local.address_v4.len + ) + elif local.family == AddressFamily.IPv6: + msg.msg_controllen = + typeof(msg.msg_controllen)(CMSG_SPACE(sizeof(In6PktInfo).csize_t)) + cmsg.cmsg_len = typeof(cmsg.cmsg_len)(CMSG_LEN(sizeof(In6PktInfo).csize_t)) + cmsg.cmsg_level = IPPROTO_IPV6 + cmsg.cmsg_type = IPV6_PKTINFO + let info = cast[ptr In6PktInfo](CMSG_DATA(cmsg)) + copyMem( + addr info.ipi6_addr, unsafeAddr local.address_v6[0], local.address_v6.len + ) + +when defined(linux) or defined(macosx): + proc recvPacket*( + fd: SocketHandle, + buf: var seq[byte], + boundLocal: TransportAddress, + local, remote: var TransportAddress, + ): int {.raises: [].} = + var + remoteStorage: Sockaddr_storage + iov = IOVec(iov_base: addr buf[0], iov_len: buf.len.csize_t) + control: ControlBuffer + msg = Tmsghdr( + msg_name: addr remoteStorage, + msg_namelen: SockLen(sizeof(remoteStorage)), + msg_iov: addr iov, + msg_iovlen: 1, + msg_control: addr control.data[0], + ) + msg.msg_controllen = typeof(msg.msg_controllen)(control.data.len) + + let response = recvmsg(fd, addr msg, 0) + if response < 0: + return response + if (msg.msg_flags and MSG_CTRUNC) != 0: + trace "Dropping UDP datagram with truncated packet info", boundLocal + return -1 + + remote = toTransportAddress(cast[ptr SockAddr](addr remoteStorage)) + if boundLocal.family == AddressFamily.IPv6 and remote.isV4Mapped(): + remote = remote.toIPv4() + local = boundLocal + + var cmsg = CMSG_FIRSTHDR(addr msg) + while not cmsg.isNil: + when defined(linux): + if cmsg.cmsg_level == IPPROTO_IP and cmsg.cmsg_type == IP_PKTINFO: + let info = cast[ptr InPktInfo](CMSG_DATA(cmsg)) + local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) + copyMem(addr local.address_v4[0], addr info.ipi_addr, local.address_v4.len) + break + else: + if cmsg.cmsg_level == IPPROTO_IP and cmsg.cmsg_type == IP_RECVDSTADDR: + local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) + copyMem(addr local.address_v4[0], CMSG_DATA(cmsg), local.address_v4.len) + break + if cmsg.cmsg_level == IPPROTO_IPV6 and cmsg.cmsg_type == IPV6_PKTINFO: + let info = cast[ptr In6PktInfo](CMSG_DATA(cmsg)) + local = TransportAddress(family: AddressFamily.IPv6, port: boundLocal.port) + copyMem(addr local.address_v6[0], addr info.ipi6_addr, local.address_v6.len) + break + cmsg = CMSG_NXTHDR(addr msg, cmsg) + + local = local.matchSocketFamily(boundLocal) + response + when defined(windows): - import std/winlean - - {.pragma: wsa, stdcall, dynlib: "ws2_32.dll".} - - type WSABUF* = object - len*: culong - buf*: ptr char - - proc WSASendTo*( - s: SocketHandle, - lpBuffers: ptr WSABUF, - dwBufferCount: culong, - lpNumberOfBytesSent: ptr culong, - dwFlags: culong, - lpTo: ptr SockAddr, - iToLen: cint, - lpOverlapped: pointer, - lpCompletionRoutine: pointer, - ): cint {.wsa, importc: "WSASendTo".} + const WsaMsgCtrunc = DWORD(0x0200) + + func wsaCmsgAlign(value: uint): uint = + (value + uint(sizeof(uint) - 1)) and not uint(sizeof(uint) - 1) + + proc prepareSourceAddr( + localSa: ptr SockAddr, control: var ControlBuffer, msg: var osdefs.WSAMSG + ) = + var local = localSa.toTransportAddress() + if local.isV4Mapped(): + local = local.toIPv4() + if local.isAnyLocal(): + return + + zeroMem(addr control.data[0], control.data.len) + let + headerLen = wsaCmsgAlign(uint(sizeof(osdefs.WSACMSGHDR))) + header = cast[ptr osdefs.WSACMSGHDR](addr control.data[0]) + data = cast[pointer](cast[uint](header) + headerLen) + + msg.control.buf = cast[cstring](addr control.data[0]) + if local.family == AddressFamily.IPv4: + header.cmsg_len = headerLen + uint(sizeof(osdefs.WinInPktInfo)) + header.cmsg_level = osdefs.IPPROTO_IP + header.cmsg_type = osdefs.IP_PKTINFO + msg.control.len = ULONG(wsaCmsgAlign(header.cmsg_len)) + let info = cast[ptr osdefs.WinInPktInfo](data) + copyMem(addr info.ipi_addr, unsafeAddr local.address_v4[0], local.address_v4.len) + elif local.family == AddressFamily.IPv6: + header.cmsg_len = headerLen + uint(sizeof(osdefs.WinIn6PktInfo)) + header.cmsg_level = osdefs.IPPROTO_IPV6 + header.cmsg_type = osdefs.IPV6_PKTINFO + msg.control.len = ULONG(wsaCmsgAlign(header.cmsg_len)) + let info = cast[ptr osdefs.WinIn6PktInfo](data) + copyMem(addr info.ipi6_addr, unsafeAddr local.address_v6[0], local.address_v6.len) + + proc recvPacket*( + ctx: QuicContext, + fd: SocketHandle, + buf: var seq[byte], + boundLocal: TransportAddress, + local, remote: var TransportAddress, + ): int {.raises: [].} = + var + remoteStorage: Sockaddr_storage + control: ControlBuffer + data = osdefs.WSABUF(len: ULONG(buf.len), buf: cast[cstring](addr buf[0])) + msg = osdefs.WSAMSG( + name: cast[ptr SockAddr](addr remoteStorage), + namelen: cint(sizeof(remoteStorage)), + lpBuffers: addr data, + dwBufferCount: DWORD(1), + control: osdefs.WSABUF( + len: ULONG(control.data.len), buf: cast[cstring](addr control.data[0]) + ), + ) + bytesReceived: DWORD + + if ctx.recvMsg(fd, addr msg, addr bytesReceived) != 0: + return -1 + if (msg.dwFlags and WsaMsgCtrunc) != 0: + trace "Dropping UDP datagram with truncated packet info", boundLocal + return -1 + + remote = toTransportAddress(cast[ptr SockAddr](addr remoteStorage)) + if boundLocal.family == AddressFamily.IPv6 and remote.isV4Mapped(): + remote = remote.toIPv4() + local = boundLocal + + var + cursor = cast[uint](msg.control.buf) + finish = cursor + uint(msg.control.len) + while cursor + uint(sizeof(osdefs.WSACMSGHDR)) <= finish: + let header = cast[ptr osdefs.WSACMSGHDR](cursor) + if header.cmsg_len < uint(sizeof(osdefs.WSACMSGHDR)) or + cursor + header.cmsg_len > finish: + break + let data = cast[pointer](cursor + wsaCmsgAlign(uint(sizeof(osdefs.WSACMSGHDR)))) + if header.cmsg_level == osdefs.IPPROTO_IP and header.cmsg_type == osdefs.IP_PKTINFO: + let info = cast[ptr osdefs.WinInPktInfo](data) + local = TransportAddress(family: AddressFamily.IPv4, port: boundLocal.port) + copyMem(addr local.address_v4[0], addr info.ipi_addr, local.address_v4.len) + break + if header.cmsg_level == osdefs.IPPROTO_IPV6 and + header.cmsg_type == osdefs.IPV6_PKTINFO: + let info = cast[ptr osdefs.WinIn6PktInfo](data) + local = TransportAddress(family: AddressFamily.IPv6, port: boundLocal.port) + copyMem(addr local.address_v6[0], addr info.ipi6_addr, local.address_v6.len) + break + cursor += wsaCmsgAlign(header.cmsg_len) + + local = local.matchSocketFamily(boundLocal) + bytesReceived.int proc prepareDestAddr( localSa: ptr SockAddr, @@ -77,9 +268,11 @@ when not defined(windows): spec: struct_lsquic_out_spec, destStorage: var Sockaddr_storage, destAddrLen: SockLen, + control: var ControlBuffer, ): Tmsghdr = + var response: Tmsghdr when defined(linux) and defined(x86_64) and not defined(android): - Tmsghdr( + response = Tmsghdr( msg_name: cast[pointer](addr destStorage), msg_namelen: destAddrLen, msg_iov: cast[ptr IOVec](spec.iov), @@ -89,7 +282,7 @@ when not defined(windows): msg_flags: 0, ) else: - Tmsghdr( + response = Tmsghdr( msg_name: cast[pointer](addr destStorage), msg_namelen: destAddrLen, msg_iov: cast[ptr IOVec](spec.iov), @@ -98,6 +291,8 @@ when not defined(windows): msg_controllen: 0, msg_flags: 0, ) + prepareSourceAddr(spec.local_sa, control, response) + response proc packetIn*( ctx: QuicContext, @@ -144,6 +339,7 @@ proc sendPacketsOut*( when defined(linux): var destStorages {.noinit.}: array[SendmmsgBatchSize, Sockaddr_storage] + controls {.noinit.}: array[SendmmsgBatchSize, ControlBuffer] msgs {.noinit.}: array[SendmmsgBatchSize, MMsgHdr] sent = 0 @@ -153,8 +349,10 @@ proc sendPacketsOut*( let curr = specsArr[sent + i] var destAddrLen: SockLen prepareDestAddr(curr.local_sa, curr.dest_sa, destStorages[i], destAddrLen) - msgs[i] = - MMsgHdr(msg_hdr: makeMsgHdr(curr, destStorages[i], destAddrLen), msg_len: 0) + msgs[i] = MMsgHdr( + msg_hdr: makeMsgHdr(curr, destStorages[i], destAddrLen, controls[i]), + msg_len: 0, + ) let res = sendmmsg(SocketHandle(quicCtx.fd), addr msgs[0], nmsgs.cuint, 0) if res < 0: @@ -177,8 +375,8 @@ proc sendPacketsOut*( else: when defined(windows): var - bufs {.noinit.}: array[MaxBatch, WSABUF] - overflow: seq[WSABUF] + bufs {.noinit.}: array[MaxBatch, osdefs.WSABUF] + overflow: seq[osdefs.WSABUF] var sent = 0 for i in 0 ..< nspecs.int: let curr = specsArr[i] @@ -193,28 +391,27 @@ proc sendPacketsOut*( iovlen = curr.iovlen.int dst = if iovlen <= bufs.len: - cast[ptr UncheckedArray[WSABUF]](addr bufs[0]) + cast[ptr UncheckedArray[osdefs.WSABUF]](addr bufs[0]) else: overflow.setLen(iovlen) - cast[ptr UncheckedArray[WSABUF]](addr overflow[0]) + cast[ptr UncheckedArray[osdefs.WSABUF]](addr overflow[0]) for j in 0 ..< iovlen: let src = iovArr[j] - dst[j].len = culong(src.iov_len) - dst[j].buf = cast[ptr char](src.iov_base) - - var bytesSent: culong = 0 - let res = WSASendTo( - SocketHandle(quicCtx.fd), - addr dst[0], - culong(iovlen), - addr bytesSent, - 0, # flags - cast[ptr SockAddr](addr destStorage), - cint(destAddrLen), - nil, - nil, # no overlapped - ) + dst[j].len = ULONG(src.iov_len) + dst[j].buf = cast[cstring](src.iov_base) + + var + control: ControlBuffer + bytesSent: DWORD + msg = osdefs.WSAMSG( + name: cast[ptr SockAddr](addr destStorage), + namelen: cint(destAddrLen), + lpBuffers: addr dst[0], + dwBufferCount: DWORD(iovlen), + ) + prepareSourceAddr(curr.local_sa, control, msg) + let res = quicCtx.sendMsg(SocketHandle(quicCtx.fd), addr msg, addr bytesSent) if res != 0: let errorCode = osdefs.wsaGetLastError() trace "Failed to send UDP datagram", @@ -223,7 +420,8 @@ proc sendPacketsOut*( return -1 break else: - let msg = makeMsgHdr(curr, destStorage, destAddrLen) + var control: ControlBuffer + let msg = makeMsgHdr(curr, destStorage, destAddrLen, control) let res = sendmsg(SocketHandle(quicCtx.fd), msg.addr, 0) if res < 0: diff --git a/lsquic/endpoint.nim b/lsquic/endpoint.nim index f6a4416..2e0b17d 100644 --- a/lsquic/endpoint.nim +++ b/lsquic/endpoint.nim @@ -2,10 +2,12 @@ # Copyright (c) Status Research & Development GmbH import chronos, chronicles, results +import std/nativesockets except SOL_SOCKET, SO_RCVBUF when defined(windows): - from chronos/osdefs import SOL_SOCKET, SO_RCVBUF + from chronos/osdefs import SOL_SOCKET, SO_RCVBUF, getsockname else: from posix import SOL_SOCKET, SO_RCVBUF + from chronos/osdefs import getsockname import ./[ errors, connection, tlsconfig, connectionmanager, lsquic_ffi, certificateverifier, @@ -13,11 +15,6 @@ import ] import ./context/[server, client, context, io] import ./helpers/[logging, transportaddr] -from chronos/osdefs import Sockaddr_storage, SockAddr, SockLen, SocketHandle -when defined(windows): - from std/winlean import recvfrom -else: - from chronos/osdefs import recvfrom logScope: topics = "lsquic" @@ -46,7 +43,6 @@ type const CloseWait: Duration = 300.milliseconds - MaxDatagramsPerWakeup = 64 ## Capped so that a busy socket cannot starve the rest of the event loop. @@ -90,6 +86,10 @@ proc createServerContext( var context = ServerContext.new(tlsConfig, engineConfig).valueOr: raise newException(QuicError, error) context.fd = fd + when defined(windows): + if not context.initPacketIo(SocketHandle(fd)): + context.destroy() + raise newException(QuicError, "could not initialize Windows packet I/O") context proc createClientContext( @@ -98,6 +98,10 @@ proc createClientContext( var context = ClientContext.new(tlsConfig, engineConfig).valueOr: raise newException(QuicError, error) context.fd = fd + when defined(windows): + if not context.initPacketIo(SocketHandle(fd)): + context.destroy() + raise newException(QuicError, "could not initialize Windows packet I/O") context proc scidLen(endpoint: QuicEndpoint): cuint {.raises: [].} = @@ -194,33 +198,8 @@ proc routeDatagram( trace "Dropping packet with unknown connection ID", bytes = data.len, local, remote {} -proc recvDatagram( - fd: SocketHandle, - buf: var seq[byte], - remoteAddress: var Sockaddr_storage, - remoteAddrLen: var SockLen, -): int {.raises: [].} = - when defined(windows): - recvfrom( - fd, - cast[cstring](addr buf[0]), - cint(buf.len), - cint(0), - cast[ptr SockAddr](addr remoteAddress), - addr remoteAddrLen, - ).int - else: - recvfrom( - fd, - addr buf[0], - buf.len, - cint(0), - cast[ptr SockAddr](addr remoteAddress), - addr remoteAddrLen, - ).int - proc drainDatagrams( - endpoint: QuicEndpoint, udp: DatagramTransport, local: TransportAddress + endpoint: QuicEndpoint, udp: DatagramTransport ): set[RouteTarget] {.raises: [].} = ## Chronos hands this callback a single datagram per event-loop wakeup, so ## whatever else has already arrived is read here rather than one wakeup, and @@ -228,23 +207,38 @@ proc drainDatagrams( if endpoint.drainBuf.len == 0: endpoint.drainBuf = newSeq[byte](DefaultDatagramBufferSize) - var targets: set[RouteTarget] + var + targets: set[RouteTarget] + boundLocal: TransportAddress + try: + boundLocal = udp.localAddress() + except TransportOsError: + return + for _ in 0 ..< MaxDatagramsPerWakeup: - var - remoteAddress: Sockaddr_storage - remoteAddrLen = SockLen(sizeof(Sockaddr_storage)) - let res = recvDatagram( - SocketHandle(udp.fd), endpoint.drainBuf, remoteAddress, remoteAddrLen - ) + var local, remote: TransportAddress + let res = + when defined(windows): + let ctx = + if not endpoint.clientContext.isNil: + endpoint.clientContext + else: + endpoint.serverContext + if ctx.isNil: + -1 + else: + recvPacket( + ctx, SocketHandle(udp.fd), endpoint.drainBuf, boundLocal, local, remote + ) + else: + recvPacket(SocketHandle(udp.fd), endpoint.drainBuf, boundLocal, local, remote) if res < 0: # Empty, or an error the transport will report again on the next wakeup. break if res > 0: targets.incl endpoint.routeDatagram( - endpoint.drainBuf.toOpenArray(0, res - 1), - local, - toTransportAddress(cast[ptr SockAddr](addr remoteAddress)), + endpoint.drainBuf.toOpenArray(0, res - 1), local, remote ) targets @@ -272,7 +266,7 @@ proc receiveFromUdp( var msg: seq[byte] msgLen: int - local = udp.localAddress() + local = udp.receivedLocalAddress().matchSocketFamily(udp.localAddress()) readIncoming(udp, msg, msgLen) if msgLen > 0: targets = endpoint.routeDatagram(msg.toOpenArray(0, msgLen - 1), local, remote) @@ -280,7 +274,7 @@ proc receiveFromUdp( warn "Failed to read UDP datagram", error = shortLog(e.msg) return - targets = targets + endpoint.drainDatagrams(udp, local) + targets = targets + endpoint.drainDatagrams(udp) if rtClient in targets: endpoint.clientContext.processWhenReady() @@ -298,9 +292,11 @@ proc createUdp( let udp = case address.family of AddressFamily.IPv4: - newDatagramTransport(onReceive, local = address) + newDatagramTransport(onReceive, local = address, flags = {ServerFlags.PacketInfo}) of AddressFamily.IPv6: - newDatagramTransport6(onReceive, local = address) + newDatagramTransport6( + onReceive, local = address, flags = {ServerFlags.PacketInfo} + ) else: raise newException(QuicError, "only IPv4/IPv6 address is supported") @@ -380,6 +376,46 @@ proc ensureClientContext( endpoint.clientContext +proc selectSourceAddress(remote: TransportAddress): TransportAddress {.raises: [].} = + let domain = + case remote.family + of AddressFamily.IPv4: + AF_INET + of AddressFamily.IPv6: + AF_INET6 + else: + return + let fd = createNativeSocket(domain, SOCK_DGRAM, IPPROTO_UDP) + if fd.int == osInvalidSocket.int: + return + defer: + nativesockets.close(fd) + + var + remoteStorage, localStorage: Sockaddr_storage + remoteLen, localLen: SockLen + remote.toSAddr(remoteStorage, remoteLen) + if connect(fd, cast[ptr SockAddr](addr remoteStorage), remoteLen) != 0: + return + localLen = SockLen(sizeof(localStorage)) + if getsockname(fd, cast[ptr SockAddr](addr localStorage), addr localLen) == 0: + fromSAddr(addr localStorage, localLen, result) + +proc dialLocalAddress( + endpoint: QuicEndpoint, remote: TransportAddress +): TransportAddress {.raises: [TransportOsError].} = + let bound = endpoint.udp.localAddress() + if not bound.isAnyLocal(): + return bound + + var source = selectSourceAddress(remote) + if source.family == AddressFamily.None: + return bound + + source = source.matchSocketFamily(bound) + source.port = bound.port + source + proc accept*( endpoint: QuicEndpoint ): Future[Connection] {.async: (raises: [CancelledError, TransportError]).} = @@ -418,7 +454,7 @@ proc dial( .} = let ctx = endpoint.ensureClientContext() let connection = newOutgoingConnection( - ctx, endpoint.udp.localAddress(), address, serverName, certVerifier + ctx, endpoint.dialLocalAddress(address), address, serverName, certVerifier ) endpoint.connman.addConnection(connection) var connected = false diff --git a/lsquic/helpers/transportaddr.nim b/lsquic/helpers/transportaddr.nim index f7e1a08..dcffdde 100644 --- a/lsquic/helpers/transportaddr.nim +++ b/lsquic/helpers/transportaddr.nim @@ -37,3 +37,9 @@ proc isIPv6Family*(sock: ptr SockAddr): bool {.inline.} = proc isIPv4Family*(sock: ptr SockAddr): bool {.inline.} = sock.sa_family.int == AF_INET.int + +func matchSocketFamily*(address, bound: TransportAddress): TransportAddress {.inline.} = + if bound.family == AddressFamily.IPv6 and address.family == AddressFamily.IPv4: + address.toIPv6() + else: + address diff --git a/tests/test_all.nim b/tests/test_all.nim index d86d5cc..6631ca4 100644 --- a/tests/test_all.nim +++ b/tests/test_all.nim @@ -6,7 +6,7 @@ import test_connection, test_perf, test_tlsconfig, test_verifier, test_lifecycle, test_timeout, test_stress, test_runtime, test_routing, test_destaddr, test_send, - test_engine_config + test_engine_config, test_receive from ./helpers/trackers import finalCheckTrackers finalCheckTrackers() diff --git a/tests/test_connection.nim b/tests/test_connection.nim index 037d895..3c1e91a 100644 --- a/tests/test_connection.nim +++ b/tests/test_connection.nim @@ -44,6 +44,9 @@ proc runConnectionTest( incomingConn.localAddress().port == boundAddress.port outgoingConn.localAddress().port == incomingConn.remoteAddress().port + if boundAddress.isAnyLocal(): + check not incomingConn.localAddress().isAnyLocal() + let outgoingBehaviour = proc() {.async.} = let stream = await outgoingConn.openStream() @@ -180,6 +183,23 @@ proc runEndpointSharedSocketDialTest(address: TransportAddress) {.async.} = incomingConn.close() await allFutures(outgoingConn.closedFuture(), incomingConn.closedFuture()) +proc runEndpointDualStackDialTest() {.async.} = + let dialer = makeEndpoint(WildcardIP6) + let server = makeEndpoint(AutoAddressIP4, {CanListen}) + let dialerPort = dialer.localAddress().port + defer: + await allFutures(dialer.stop(), server.stop()) + + let accepting = server.accept() + let outgoingConn = await dialer.dial(server.localAddress()).wait(dialTimeout) + let incomingConn = await accepting.wait(dialTimeout) + + check incomingConn.remoteAddress().port == dialerPort + + outgoingConn.close() + incomingConn.close() + await allFutures(outgoingConn.closedFuture(), incomingConn.closedFuture()) + proc runEndpointDialOnlyTest(address: TransportAddress) {.async.} = let server = makeServer() let listener = server.listen(address) @@ -392,6 +412,9 @@ suite "connection": asyncTest "endpoint dials from listener socket": await runEndpointSharedSocketDialTest(AutoAddressIP4) + asyncTest "dual-stack endpoint dials IPv4 from listener socket": + await runEndpointDualStackDialTest() + asyncTest "dial-only endpoint works without listener": await runEndpointDialOnlyTest(AutoAddressIP4) diff --git a/tests/test_receive.nim b/tests/test_receive.nim new file mode 100644 index 0000000..3075427 --- /dev/null +++ b/tests/test_receive.nim @@ -0,0 +1,66 @@ +# SPDX-License-Identifier: Apache-2.0 OR MIT +# Copyright (c) Status Research & Development GmbH + +{.used.} + +import chronos, chronos/osdefs, chronos/unittest2/asynctests +import ../lsquic/context/io +when defined(windows): + import ../lsquic/context/context +import ./helpers/trackers + +suite "UDP receive": + teardown: + checkTrackers() + + asyncTest "drained IPv4 burst keeps the dual-stack socket family": + proc ignore( + udp: DatagramTransport, remote: TransportAddress + ): Future[void] {.async: (raises: []).} = + discard + + let + server = newDatagramTransport6( + ignore, + local = initTAddress("[::]:0"), + flags = {ServerFlags.NoAutoRead, ServerFlags.PacketInfo}, + ) + client = newDatagramTransport(ignore) + defer: + await allFutures(server.closeWait(), client.closeWait()) + + when defined(windows): + let ctx = QuicContext() + check ctx.initPacketIo(SocketHandle(server.fd)) + + var destination = initTAddress("127.0.0.1:0") + destination.port = server.localAddress().port + await client.sendTo(destination, @[1.byte]) + await client.sendTo(destination, @[2.byte]) + + var buf = newSeq[byte](DefaultDatagramBufferSize) + for expected in [1.byte, 2.byte]: + var + local, remote: TransportAddress + received = -1 + + for _ in 0 ..< 100: + received = + when defined(windows): + recvPacket( + ctx, SocketHandle(server.fd), buf, server.localAddress(), local, remote + ) + else: + recvPacket( + SocketHandle(server.fd), buf, server.localAddress(), local, remote + ) + if received >= 0: + break + # The socket is non-blocking and send completion may precede readability. + await sleepAsync(10.milliseconds) + + check: + received == 1 + buf[0] == expected + local == destination.toIPv6() + remote.family == AddressFamily.IPv4 diff --git a/tests/test_send.nim b/tests/test_send.nim index 96b52b6..7e4711d 100644 --- a/tests/test_send.nim +++ b/tests/test_send.nim @@ -4,8 +4,10 @@ {.used.} import chronos, chronos/osdefs, nativesockets, unittest2 +import chronos/osutils import lsquic/[lsquic_ffi, context/context, context/io] import ./helpers/[address, trackers] +import std/os when not defined(windows): from posix import EAGAIN, EBADF, errno @@ -20,6 +22,41 @@ proc makeOutSpec( dest_sa: cast[ptr SockAddr](dest), ) +proc makeContext(fd: SocketHandle): QuicContext = + var response = QuicContext(fd: cint(fd)) + when defined(windows): + doAssert response.initPacketIo(fd) + response + +proc receiveWithTimeout( + fd: SocketHandle, + received: var array[3, byte], + remoteStorage: var Sockaddr_storage, + remoteLen: var SockLen, +): int = + doAssert setDescriptorBlocking(fd, false).isOk() + let buffer = + when defined(windows): + cast[cstring](addr received[0]) + else: + addr received[0] + + var response: int = -1 + for _ in 0 ..< 100: + remoteLen = sizeof(remoteStorage).SockLen + response = recvfrom( + fd, + buffer, + received.len.cint, + 0, + cast[ptr SockAddr](addr remoteStorage), + addr remoteLen, + ).int + if response >= 0: + return response + os.sleep(10) + response + suite "packets out": teardown: checkTrackers() @@ -46,7 +83,7 @@ suite "packets out": defer: nativesockets.close(fd) - let ctx = QuicContext(fd: cint(fd)) + let ctx = makeContext(fd) var payload = @[1'u8, 2, 3] localStorage = toSockaddrStorage(initTAddress("127.0.0.1:1000")) @@ -74,7 +111,7 @@ suite "packets out": defer: nativesockets.close(fd) - let ctx = QuicContext(fd: cint(fd)) + let ctx = makeContext(fd) var payload = @[1'u8, 2, 3] localStorage = toSockaddrStorage(initTAddress("127.0.0.1:1000")) @@ -85,3 +122,90 @@ suite "packets out": spec = makeOutSpec(addr iov, addr localStorage, addr destStorage) check sendPacketsOut(cast[pointer](ctx), addr specs[0], SpecCount.cuint) == SpecCount + + test "packet info selects the requested IPv4 source address": + let + senderFd = createNativeSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) + receiverFd = createNativeSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) + requestedSource = when defined(linux): "127.0.0.2" else: "127.0.0.1" + defer: + nativesockets.close(senderFd) + nativesockets.close(receiverFd) + + var + senderBind = toSockaddrStorage(initTAddress("0.0.0.0:0")) + receiverBind = toSockaddrStorage(initTAddress(requestedSource & ":0")) + require bindSocket( + senderFd, cast[ptr SockAddr](addr senderBind), sizeof(Sockaddr_in).SockLen + ) == 0 + require bindSocket( + receiverFd, cast[ptr SockAddr](addr receiverBind), sizeof(Sockaddr_in).SockLen + ) == 0 + + var receiverLen = sizeof(receiverBind).SockLen + require getsockname( + receiverFd, cast[ptr SockAddr](addr receiverBind), addr receiverLen + ) == 0 + + let ctx = makeContext(senderFd) + var + payload = @[byte(1), 2, 3] + localStorage = toSockaddrStorage(initTAddress(requestedSource & ":0")) + iov = struct_iovec(iov_base: addr payload[0], iov_len: payload.len.csize_t) + spec = makeOutSpec(addr iov, addr localStorage, addr receiverBind) + + require sendPacketsOut(cast[pointer](ctx), addr spec, 1) == 1 + + var + received: array[3, byte] + remoteStorage: Sockaddr_storage + remoteLen = sizeof(remoteStorage).SockLen + check receiveWithTimeout(receiverFd, received, remoteStorage, remoteLen) == + received.len + + var remote: TransportAddress + fromSAddr(addr remoteStorage, remoteLen, remote) + check remote.toIpAddress() == parseIpAddress(requestedSource) + + test "packet info selects the requested IPv6 source address": + let + senderFd = createNativeSocket(AF_INET6, SOCK_DGRAM, IPPROTO_UDP) + receiverFd = createNativeSocket(AF_INET6, SOCK_DGRAM, IPPROTO_UDP) + defer: + nativesockets.close(senderFd) + nativesockets.close(receiverFd) + + var + senderBind = toSockaddrStorage(initTAddress("[::]:0")) + receiverBind = toSockaddrStorage(initTAddress("[::1]:0")) + require bindSocket( + senderFd, cast[ptr SockAddr](addr senderBind), sizeof(Sockaddr_in6).SockLen + ) == 0 + require bindSocket( + receiverFd, cast[ptr SockAddr](addr receiverBind), sizeof(Sockaddr_in6).SockLen + ) == 0 + + var receiverLen = sizeof(receiverBind).SockLen + require getsockname( + receiverFd, cast[ptr SockAddr](addr receiverBind), addr receiverLen + ) == 0 + + let ctx = makeContext(senderFd) + var + payload = @[byte(1), 2, 3] + localStorage = toSockaddrStorage(initTAddress("[::1]:0")) + iov = struct_iovec(iov_base: addr payload[0], iov_len: payload.len.csize_t) + spec = makeOutSpec(addr iov, addr localStorage, addr receiverBind) + + require sendPacketsOut(cast[pointer](ctx), addr spec, 1) == 1 + + var + received: array[3, byte] + remoteStorage: Sockaddr_storage + remoteLen = sizeof(remoteStorage).SockLen + check receiveWithTimeout(receiverFd, received, remoteStorage, remoteLen) == + received.len + + var remote: TransportAddress + fromSAddr(addr remoteStorage, remoteLen, remote) + check remote.toIpAddress() == parseIpAddress("::1")