From ddca59173ad3e208a8b4be2b60940df69a43680e Mon Sep 17 00:00:00 2001 From: Brad Fitzpatrick Date: Thu, 24 Sep 2026 17:54:25 +0000 Subject: [PATCH] tagpeers, lansport, rendezvous: add LAN peering among tagged tailnet nodes These are the reusable layers under gocached's upcoming cache pooling for the colo Macs, moved here so anything with the same shape can use them: a set of tailnet nodes on one LAN that want to exchange bulk traffic directly, without pushing every byte through magicsock/wireguard, while still getting their trust from the tailnet. tagpeers watches tailscaled's IPN bus and keeps the set of nodes carrying a tag, as the control plane sees it; no reachability analysis. lansport turns those nodes into working connections: each process serves a plain-HTTP advert over the tailnet naming its LAN TLS address and per-process certificate hash, fetches its peers' adverts, pins their certificates, and maintains a pinned *http.Transport to each reachable peer. Requests from peers arrive on a TLS listener that admits only pinned certificates. rendezvous weighted rendezvous hashing over a changing member set, and a router that picks which lansport peer, if any, should handle a key. tagpeerstest has a fake tailscaled serving the IPN bus watch endpoint for testing code built on these. Updates tailscale/corp#48514 Signed-off-by: Brad Fitzpatrick Change-Id: I7b2f9c41d8e3a65f0c2d4b8e1a9f3c7d5e6b0a12 --- go.mod | 33 ++ go.sum | 170 ++++++++ lansport/identity.go | 158 +++++++ lansport/lansport.go | 771 ++++++++++++++++++++++++++++++++++ lansport/lansport_test.go | 368 ++++++++++++++++ lansport/lansporttest/lan.go | 118 ++++++ lansport/source.go | 138 ++++++ rendezvous/rendezvous.go | 105 +++++ rendezvous/rendezvous_test.go | 83 ++++ rendezvous/router.go | 94 +++++ tagpeers/tagpeerstest/fake.go | 215 ++++++++++ tagpeers/tracker.go | 461 ++++++++++++++++++++ tagpeers/tracker_test.go | 183 ++++++++ 13 files changed, 2897 insertions(+) create mode 100644 go.mod create mode 100644 go.sum create mode 100644 lansport/identity.go create mode 100644 lansport/lansport.go create mode 100644 lansport/lansport_test.go create mode 100644 lansport/lansporttest/lan.go create mode 100644 lansport/source.go create mode 100644 rendezvous/rendezvous.go create mode 100644 rendezvous/rendezvous_test.go create mode 100644 rendezvous/router.go create mode 100644 tagpeers/tagpeerstest/fake.go create mode 100644 tagpeers/tracker.go create mode 100644 tagpeers/tracker_test.go diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..3d1a87f --- /dev/null +++ b/go.mod @@ -0,0 +1,33 @@ +module github.com/tailscale/tb + +go 1.27.1 + +require ( + github.com/cespare/xxhash/v2 v2.3.0 + tailscale.com v1.103.0-pre.0.20260920011920-7d96cf5a62ef +) + +require ( + filippo.io/edwards25519 v1.2.0 // indirect + github.com/akutz/memconn v0.1.0 // indirect + github.com/coder/websocket v1.8.15 // indirect + github.com/dblohm7/wingoes v0.0.0-20260526185140-fb298caac7ca // indirect + github.com/fxamacker/cbor/v2 v2.9.3 // indirect + github.com/go-json-experiment/json v0.0.0-20260820222146-c27c302e5fc3 // indirect + github.com/google/go-cmp v0.7.0 // indirect + github.com/hdevalence/ed25519consensus v0.2.0 // indirect + github.com/jsimonetti/rtnetlink v1.4.2 // indirect + github.com/mdlayher/netlink v1.11.2 // indirect + github.com/mdlayher/socket v0.7.0 // indirect + github.com/mitchellh/go-ps v1.0.0 // indirect + github.com/tailscale/go-winio v0.0.0-20231025203758-c4f33415bf55 // indirect + github.com/x448/float16 v0.8.4 // indirect + go4.org/mem v0.0.0-20240501181205-ae6ca9944745 // indirect + go4.org/netipx v0.0.0-20260823151212-3075585bcbeb // indirect + golang.org/x/crypto v0.57.0 // indirect + golang.org/x/exp v0.0.0-20260908205506-85c1c2202aba // indirect + golang.org/x/net v0.59.0 // indirect + golang.org/x/sync v0.23.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.zx2c4.com/wireguard/windows v1.0.1 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..41f69f0 --- /dev/null +++ b/go.sum @@ -0,0 +1,170 @@ +filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo= +filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc= +github.com/akutz/memconn v0.1.0 h1:NawI0TORU4hcOMsMr11g7vwlCdkYeLKXBcxWu2W/P8A= +github.com/akutz/memconn v0.1.0/go.mod h1:Jo8rI7m0NieZyLI5e2CDlRdRqRRB4S7Xp77ukDjH+Fw= +github.com/alexbrainman/sspi v0.0.0-20250919150558-7d374ff0d59e h1:4dAU9FXIyQktpoUAgOJK3OTFc/xug0PCXYCqU0FgDKI= +github.com/alexbrainman/sspi v0.0.0-20250919150558-7d374ff0d59e/go.mod h1:cEWa1LVoE5KvSD9ONXsZrj0z6KqySlCCNKHlLzbqAt4= +github.com/aws/aws-sdk-go-v2 v1.46.0 h1:1kt7m/EKcEHt5mlyyxx9cSlMddRPIKbjb6DIQsu4HPk= +github.com/aws/aws-sdk-go-v2 v1.46.0/go.mod h1:bttEH6JqnUL8LepvDVfdrds/fZ5bCIxzpe3abyUrhDU= +github.com/aws/aws-sdk-go-v2/config v1.33.3 h1:h090b3O5S17bF87/0ysHZuIT/7DCb4EBRFQX2PMVPCw= +github.com/aws/aws-sdk-go-v2/config v1.33.3/go.mod h1:YYDB1kTejxbfAbEVUqgCtkVp26xvNCHev9cLKABMGAk= +github.com/aws/aws-sdk-go-v2/credentials v1.20.3 h1:tToOYM/LXev4NpfWlIYGDvBvjHmJ3HXpRU9ppl+pM6k= +github.com/aws/aws-sdk-go-v2/credentials v1.20.3/go.mod h1:wfGneWyncO7p67wqXV2IQhPk14JqIc25woKlaArT3WI= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.19.2 h1:Ldv7RPHs7qwwTscRjAl3YBud32f3BvdAGRmSvAx5L38= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.19.2/go.mod h1:XyK6UV8xbo66ysVqLd2783C09pBYHOm8aKTRV5DVJ30= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.2 h1:q/PSLGuRWCChWg+dLnb9dWOnrCxJtnboXbBtFoqqRrI= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.2/go.mod h1:TD1jvU2LvXkJexct5vBqcd8QlNXh5EmRUeL/Z32p0n4= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.2 h1:6fl86IPqKEXoySqiOWdfgbEp9OVbn44zTfEICNEBDhY= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.2/go.mod h1:63HDfhFkdzBpI8WGXTSKUHPKS6mqldj4u3LJW7RZtSU= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.2 h1:XMgIRS+uW9F3yFKnXGRrI9pkHi99CXTmoz2kz2/TGBA= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.2/go.mod h1:vorxDzK+n3jiv9a5ST/LG0Eu9cSv1CRdKTpG6pDMs+M= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19 h1:bAdDl/HkGCcGPoe25ToSHEw23VIxt6CT5fLcg111BKg= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19/go.mod h1:KaUzbLxv4CeSxh6ZCl9B4m7CuFenS8kUEaDs+f/DQr4= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.2 h1:ZtHYnumr6QyxhzEzNZwzQTFJEXOswrZqTTkRxthwvr4= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.2/go.mod h1:a1NXrYpBd311gBzn1UI5UzJyyvXktM4xNh/ydPiPpqY= +github.com/aws/aws-sdk-go-v2/service/signin v1.9.0 h1:c3k+k/CS4L+sAIH6fxikL+g5g2LpeNczaoyjjw1iMKI= +github.com/aws/aws-sdk-go-v2/service/signin v1.9.0/go.mod h1:AGIoQg99fBrOIQnF78TLx4lj18mc4gZ0hJx1UaLIFM4= +github.com/aws/aws-sdk-go-v2/service/ssm v1.77.0 h1:JqKSBJlS68F4nxoPLwD7tB3lJ3gWEK9p+7eOiBE/lsM= +github.com/aws/aws-sdk-go-v2/service/ssm v1.77.0/go.mod h1:5TMMHmQGjCODMeIopSNfxkan11jG0SmFpyjcMvCLzHA= +github.com/aws/aws-sdk-go-v2/service/sso v1.37.0 h1:+rqBaOq7jzInjY8M12hr+zEe85JpRll9BjMx38r33Ok= +github.com/aws/aws-sdk-go-v2/service/sso v1.37.0/go.mod h1:XFlVwUsw3sYh8Hw37umYVJnrcWrwXWRxbNw/JY0bblw= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.42.0 h1:hzM3GslEAOBcLn3DHH6ENToFi+vXP+n02W+x6zejAIM= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.42.0/go.mod h1:588e7skMkYIYkSUseT8E3WKFfbAfrA1bj3Zf+qxJtJY= +github.com/aws/aws-sdk-go-v2/service/sts v1.49.0 h1:N7Ey8obY3uSui+cxl0OUzFlFmkxSucoJnrniFhw+cLc= +github.com/aws/aws-sdk-go-v2/service/sts v1.49.0/go.mod h1:zMBwjSf4Pt8a1OHYiZ5rPD0PJRK1kQrUaxhS/Dbld8E= +github.com/aws/smithy-go v1.28.1 h1:R/nXH00c8qcfCzQVELtRw+eLQWtzv+VAIEFJ1/xxXlQ= +github.com/aws/smithy-go v1.28.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/cilium/ebpf v0.22.0 h1:v2ktp0roffpMOj2MMf3idtCQZOsAoC4BJbAJN+ke2bY= +github.com/cilium/ebpf v0.22.0/go.mod h1:CDzZbe2hC5JjlDC+CY3KFCzlYwN4gbxppYM+Z10bQt4= +github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA= +github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= +github.com/coreos/go-iptables v0.8.0 h1:MPc2P89IhuVpLI7ETL/2tx3XZ61VeICZjYqDEgNsPRc= +github.com/coreos/go-iptables v0.8.0/go.mod h1:Qe8Bv2Xik5FyTXwgIbLAnv2sWSBmvWdFETJConOQ//Q= +github.com/creachadair/msync v0.10.1 h1:14jT5Ujob64Zp2D3ZcUidayDMoXikhmYN8xBxEuPKPk= +github.com/creachadair/msync v0.10.1/go.mod h1:dFf4Z0cxE2nibFmAdq5OqTuGn++GgEqA9JuKH+7ceSA= +github.com/creachadair/taskgroup v0.14.4 h1:ttR8StLWmYA1O6x96YlTpQLXjKyRpO4RBo+OcO6W6C0= +github.com/creachadair/taskgroup v0.14.4/go.mod h1:uhCtIEsa7zpeMAFixddhCYjbJi7e4JMyU7OXUygkSsg= +github.com/dblohm7/wingoes v0.0.0-20260526185140-fb298caac7ca h1:h1Awca4lQOspNR/2eeo04Ricn5NixDX9mb17WSAgLhQ= +github.com/dblohm7/wingoes v0.0.0-20260526185140-fb298caac7ca/go.mod h1:2TGl1jRJrRpbzykmg7asHm3h08TqutUgQqY5v9k/g3c= +github.com/digitalocean/go-smbios v0.0.0-20180907143718-390a4f403a8e h1:vUmf0yezR0y7jJ5pceLHthLaYf4bA5T14B6q39S4q2Q= +github.com/digitalocean/go-smbios v0.0.0-20180907143718-390a4f403a8e/go.mod h1:YTIHhz/QFSYnu/EhlF2SpU2Uk+32abacUYA5ZPljz1A= +github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= +github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= +github.com/fxamacker/cbor/v2 v2.9.3 h1:oQBnFATpNdY8gJHTndDDv5Xl4QqNaz51G5LLEPhng3Q= +github.com/fxamacker/cbor/v2 v2.9.3/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= +github.com/gaissmai/bart v0.29.0 h1:wO6HGE8g9YE0Wm0bCpYxwRzfQ4+fbJKOhL64e5ACGCI= +github.com/gaissmai/bart v0.29.0/go.mod h1:GREWQfTLRWz/c5FTOsIw+KkscuFkIV5t8Rp7Nd1Td5c= +github.com/go-json-experiment/json v0.0.0-20260820222146-c27c302e5fc3 h1:UADEEmDKgfXbtnGJZ97beY5XLo9ZechG1nlU4KnRrkE= +github.com/go-json-experiment/json v0.0.0-20260820222146-c27c302e5fc3/go.mod h1:tphK2c80bpPhMOI4v6bIc2xWywPfbqi1Z06+RcrMkDg= +github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE= +github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78= +github.com/godbus/dbus/v5 v5.2.2 h1:TUR3TgtSVDmjiXOgAAyaZbYmIeP3DPkld3jgKGV8mXQ= +github.com/godbus/dbus/v5 v5.2.2/go.mod h1:3AAv2+hPq5rdnr5txxxRwiGjPXamgoIHgz9FPBfOp3c= +github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 h1:f+oWsMOmNPc8JmEHVZIycC7hBoQxHH9pNKQORJNozsQ= +github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8/go.mod h1:wcDNUvekVysuuOpQKo3191zZyTpiI6se1N1ULghS0sw= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= +github.com/google/nftables v0.3.0 h1:bkyZ0cbpVeMHXOrtlFc8ISmfVqq5gPJukoYieyVmITg= +github.com/google/nftables v0.3.0/go.mod h1:BCp9FsrbF1Fn/Yu6CLUc9GGZFw/+hsxfluNXXmxBfRM= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/hdevalence/ed25519consensus v0.2.0 h1:37ICyZqdyj0lAZ8P4D1d1id3HqbbG1N3iBb1Tb4rdcU= +github.com/hdevalence/ed25519consensus v0.2.0/go.mod h1:w3BHWjwJbFU29IRHL1Iqkw3sus+7FctEyM4RqDxYNzo= +github.com/huin/goupnp v1.3.0 h1:UvLUlWDNpoUdYzb2TCn+MuTWtcjXKSza2n6CBdQ0xXc= +github.com/huin/goupnp v1.3.0/go.mod h1:gnGPsThkYa7bFi/KWmEysQRf48l2dvR5bxr2OFckNX8= +github.com/illarion/gonotify/v3 v3.0.2 h1:O7S6vcopHexutmpObkeWsnzMJt/r1hONIEogeVNmJMk= +github.com/illarion/gonotify/v3 v3.0.2/go.mod h1:HWGPdPe817GfvY3w7cx6zkbzNZfi3QjcBm/wgVvEL1U= +github.com/insomniacslk/dhcp v0.0.0-20260901064844-234b97448fae h1:nXGg65fXsylSUTNNWwvHuQsXev7mhIzQTnZREPTbWzs= +github.com/insomniacslk/dhcp v0.0.0-20260901064844-234b97448fae/go.mod h1:tGfUTcnFYGYvVNCaZZhwlJySU/fQQxh9TmpsFzWXnnY= +github.com/jsimonetti/rtnetlink v1.4.2 h1:Df9w9TZ3npHTyDn0Ev9e1uzmN2odmXd0QX+J5GTEn90= +github.com/jsimonetti/rtnetlink v1.4.2/go.mod h1:92s6LJdE+1iOrw+F2/RO7LYI2Qd8pPpFNNUYW06gcoM= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/kortschak/wol v0.0.0-20200729010619-da482cc4850a h1:+RR6SqnTkDLWyICxS1xpjCi/3dhyV+TgZwA6Ww3KncQ= +github.com/kortschak/wol v0.0.0-20200729010619-da482cc4850a/go.mod h1:YTtCCM3ryyfiu4F7t8HQ1mxvp1UBdWM2r6Xa+nGWvDk= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/mdlayher/genetlink v1.4.0 h1:f/Xs7Y2T+GyX9b3dbiUhnLE9InGs5F9RxJ2JwBMl71o= +github.com/mdlayher/genetlink v1.4.0/go.mod h1:d1hrKr8fwZU2JkcAtQUAzeTrI7nbgQSl+5k1cC0biSA= +github.com/mdlayher/netlink v1.11.2 h1:HKh2jqe+omdSWcQ88nrT7INE61B0NXfiSPFdgL4YbNI= +github.com/mdlayher/netlink v1.11.2/go.mod h1:uT2Yc/QLaZubzDpZIBi9d4GoeLwtp3x1AMeqSRrK2sA= +github.com/mdlayher/sdnotify v1.0.0 h1:Ma9XeLVN/l0qpyx1tNeMSeTjCPH6NtuD6/N9XdTlQ3c= +github.com/mdlayher/sdnotify v1.0.0/go.mod h1:HQUmpM4XgYkhDLtd+Uad8ZFK1T9D5+pNxnXQjCeJlGE= +github.com/mdlayher/socket v0.7.0 h1:qVREPVwtUMg17pwvveQxpurq0PVisMxi3FGgpsorMYQ= +github.com/mdlayher/socket v0.7.0/go.mod h1:f7iKql2EK/rfsWYDKofKjk2Ig7I+6HQWdMIdn1tO4A4= +github.com/mitchellh/go-ps v1.0.0 h1:i6ampVEEF4wQFF+bkYfwYgY+F/uYJDktmvLPf7qIgjc= +github.com/mitchellh/go-ps v1.0.0/go.mod h1:J4lOc8z8yJs6vUwklHw2XEIiT4z4C40KtWVN3nvg8Pg= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/pierrec/lz4/v4 v4.1.26 h1:GrpZw1gZttORinvzBdXPUXATeqlJjqUG/D87TKMnhjY= +github.com/pierrec/lz4/v4 v4.1.26/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= +github.com/pires/go-proxyproto v0.15.0 h1:dTshmNbFm/D+0+sbrxUuddPOZ5Y0B7c5NhtsBkm6LqI= +github.com/pires/go-proxyproto v0.15.0/go.mod h1:OXsCrKwrK2tXS9YrI5tkHx5xaQlO8FH3lFW76orFh24= +github.com/prometheus/client_model v0.6.3 h1:O0jaTVAYNxTHYInEPFJt5I3+sN8zqBtVMPTB1qyxiEo= +github.com/prometheus/client_model v0.6.3/go.mod h1:gpN5P9S7Rr6Yr92PiQ+Ixvhf6JZEkF1dnxsYL2aPBEM= +github.com/prometheus/common v0.71.0 h1:9KDAKb7Mj3HEVKyFCK6Dc/HIwlBzZIN2l7/lrHl3KK8= +github.com/prometheus/common v0.71.0/go.mod h1:CLJ5H8TEsGX8bl31BdMkfhIZ+QmZ9tBPPotUxUbfcmk= +github.com/rogpeppe/go-internal v1.16.0 h1:O9DK+vNMDVGLr2BeZqmpLeMjiMNkuXfcqntWbZV6S5g= +github.com/rogpeppe/go-internal v1.16.0/go.mod h1:DrUVZyrJU+txYW5/1kwtXQSMFio52ZOxX7yM1VHvnxs= +github.com/safchain/ethtool v0.7.0 h1:rlJzfDetsVvT61uz8x1YIcFn12akMfuPulHtZjtb7Is= +github.com/safchain/ethtool v0.7.0/go.mod h1:MenQKEjXdfkjD3mp2QdCk8B/hwvkrlOTm/FD4gTpFxQ= +github.com/tailscale/certstore v0.1.1-0.20260409135935-3638fb84b77d h1:JcGKBZAL7ePLwOhUdN8qGQZlP5GueEiIZwY7R62pejE= +github.com/tailscale/certstore v0.1.1-0.20260409135935-3638fb84b77d/go.mod h1:XrBNfAFN+pwoWuksbFS9Ccxnopa15zJGgXRFN90l3K4= +github.com/tailscale/go-winio v0.0.0-20231025203758-c4f33415bf55 h1:Gzfnfk2TWrk8Jj4P4c1a3CtQyMaTVCznlkLZI++hok4= +github.com/tailscale/go-winio v0.0.0-20231025203758-c4f33415bf55/go.mod h1:4k4QO+dQ3R5FofL+SanAUZe+/QfeK0+OIuwDIRu2vSg= +github.com/tailscale/hujson v0.0.0-20260727124030-b80ff77dac4f h1:9hiVElpCmKzsBKQHkBqZ8LGzt82iLfM8egxr4sew+Ys= +github.com/tailscale/hujson v0.0.0-20260727124030-b80ff77dac4f/go.mod h1:8/zr1Tv0+cKpVtGCEB/7YfRXr2TszsMxMXLaT8YuBgU= +github.com/tailscale/netlink v1.1.1-0.20240822203006-4d49adab4de7 h1:uFsXVBE9Qr4ZoF094vE6iYTLDl0qCiKzYXlL6UeWObU= +github.com/tailscale/netlink v1.1.1-0.20240822203006-4d49adab4de7/go.mod h1:NzVQi3Mleb+qzq8VmcWpSkcSYxXIg0DkI6XDzpVkhJ0= +github.com/tailscale/peercred v0.0.0-20250107143737-35a0c7bd7edc h1:24heQPtnFR+yfntqhI3oAu9i27nEojcQ4NuBQOo5ZFA= +github.com/tailscale/peercred v0.0.0-20250107143737-35a0c7bd7edc/go.mod h1:f93CXfllFsO9ZQVq+Zocb1Gp4G5Fz0b0rXHLOzt/Djc= +github.com/tailscale/web-client-prebuilt v0.0.0-20260917222731-e0ed2d0d0fea h1:ah5jt6ZnxMvNclt6kBBoLcMpLLjMA44PTsnov2y7gpU= +github.com/tailscale/web-client-prebuilt v0.0.0-20260917222731-e0ed2d0d0fea/go.mod h1:agQPE6y6ldqCOui2gkIh7ZMztTkIQKH049tv8siLuNQ= +github.com/tailscale/wireguard-go v0.0.0-20260911194433-e3222a3340cd h1:7l7Aru/QMwjoWvfvnTEiX3gG+ixhN5WAFENxHcA9NlI= +github.com/tailscale/wireguard-go v0.0.0-20260911194433-e3222a3340cd/go.mod h1:rUelGmuK4UnSJYM5gl5Mknp6YbwwcL8+VAPMhNYe+jg= +github.com/u-root/uio v0.0.0-20240224005618-d2acac8f3701 h1:pyC9PaHYZFgEKFdlp3G8RaCKgVpHZnecvArXvPXcFkM= +github.com/u-root/uio v0.0.0-20240224005618-d2acac8f3701/go.mod h1:P3a5rG4X7tI17Nn3aOIAYr5HbIMukwXG0urG0WuL8OA= +github.com/vishvananda/netns v0.0.5 h1:DfiHV+j8bA32MFM7bfEunvT8IAqQ/NzSJHtcmW5zdEY= +github.com/vishvananda/netns v0.0.5/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM= +github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= +github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= +go4.org/mem v0.0.0-20240501181205-ae6ca9944745 h1:Tl++JLUCe4sxGu8cTpDzRLd3tN7US4hOxG5YpKCzkek= +go4.org/mem v0.0.0-20240501181205-ae6ca9944745/go.mod h1:reUoABIJ9ikfM5sgtSF3Wushcza7+WeD01VB9Lirh3g= +go4.org/netipx v0.0.0-20260823151212-3075585bcbeb h1:XBM4hvfwGAttkkiTIFfeigdfcL1xIfdKXqFdgiHGtDs= +go4.org/netipx v0.0.0-20260823151212-3075585bcbeb/go.mod h1:PLyyIXexvUFg3Owu6p/WfdlivPbZJsZdgWZlrGope/Y= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/exp v0.0.0-20260908205506-85c1c2202aba h1:Ck8QetSgk912qxWLMCKxd0in+aiyBQyDSMae6e/xmpU= +golang.org/x/exp v0.0.0-20260908205506-85c1c2202aba/go.mod h1:50RgIsmK7OwqzTTeqcSXQW8SswW0o8fRcDxmqGluJ8E= +golang.org/x/mod v0.41.0 h1:qJmnOUb4YB+FsEuM3HcWucdZASCPGhsX6uljO6pog0c= +golang.org/x/mod v0.41.0/go.mod h1:Ek9pY8RKWXwsWvd3rQiHYtMqkjSUV+s1Rj7j4H5Ur6o= +golang.org/x/net v0.59.0 h1:5zfYln+w5XCxwrnMMJPufRgNoXEaGxl0wo5GqPXyues= +golang.org/x/net v0.59.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE= +golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc= +golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= +golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE= +golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s= +golang.org/x/tools v0.50.0 h1:c2ifzfcuY7L90lZ2aKd8S4K2NpASF08SZx9ZuJkHmSU= +golang.org/x/tools v0.50.0/go.mod h1:7ulVMw3831Mwi5EZD6RomGyffr4VFjuNYXf2BbCEAV0= +golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 h1:B82qJJgjvYKsXS9jeunTOisW56dUokqW/FOteYJJ/yg= +golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2/go.mod h1:deeaetjYA+DHMHg+sMSMI58GrEteJUUzzw7en6TJQcI= +golang.zx2c4.com/wireguard/windows v1.0.1 h1:eOxiDVbywPC+ZQqvdCK7x+ZwWXKbYv50TtH8ysFIbw8= +golang.zx2c4.com/wireguard/windows v1.0.1/go.mod h1:+fbT3FFdX4zzYDLwJh5+HPEcNN/3HyNdzhNSVsQM+zs= +google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= +google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gvisor.dev/gvisor v0.0.0-20260915211658-a6f909f08a72 h1:EytKYr5WrVs+Aah/0xDO8ZAZZzS/iTEVC8ln4ipDbL0= +gvisor.dev/gvisor v0.0.0-20260915211658-a6f909f08a72/go.mod h1:8aLQqUBHDH8fY5y60lzmwDpMMbQCcT3EBfoSwhfaGCY= +tailscale.com v1.103.0-pre.0.20260920011920-7d96cf5a62ef h1:N5JizbSj4+lLBUSwIc89JAjDnxYbxEKyNhIhNwBvBHo= +tailscale.com v1.103.0-pre.0.20260920011920-7d96cf5a62ef/go.mod h1:oUmJ/nA/pTiEi4J7pOaPdEL6Bxi+OhgIXGkz1KdPAjA= diff --git a/lansport/identity.go b/lansport/identity.go new file mode 100644 index 0000000..7baa254 --- /dev/null +++ b/lansport/identity.go @@ -0,0 +1,158 @@ +// Copyright (c) Tailscale Inc & AUTHORS +// SPDX-License-Identifier: BSD-3-Clause + +package lansport + +import ( + "crypto/ed25519" + "crypto/rand" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/hex" + "errors" + "math/big" + "net/http" + "time" +) + +// Pin identifies a TLS certificate: the SHA-256 of its DER encoding. Since an +// [Identity] is generated fresh on every process start, a pin identifies one +// run of one process. +type Pin [sha256.Size]byte + +// PinOf returns the pin of cert. +func PinOf(cert *x509.Certificate) Pin { return sha256.Sum256(cert.Raw) } + +// ParsePin parses the hex form of a pin, as produced by [Pin.String]. +func ParsePin(s string) (Pin, error) { + var p Pin + b, err := hex.DecodeString(s) + if err != nil { + return p, err + } + if len(b) != len(p) { + return p, errors.New("lansport: pin has wrong length") + } + copy(p[:], b) + return p, nil +} + +// String returns the pin as 64 lowercase hex characters. +func (p Pin) String() string { return hex.EncodeToString(p[:]) } + +// Short returns the first 16 hex characters of the pin, for display. +func (p Pin) Short() string { return p.String()[:16] } + +// Identity is a process's TLS identity: a self-signed certificate over a +// freshly generated Ed25519 key, used both as the server certificate it +// presents to peers and as the client certificate on connections it makes +// to them. It lives only in memory. +type Identity struct { + cert tls.Certificate + pin Pin +} + +// NewIdentity generates an identity whose certificate names name; the name +// is informational, since peers verify by pin. +func NewIdentity(name string) (*Identity, error) { + pub, priv, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + return nil, err + } + serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128)) + if err != nil { + return nil, err + } + now := time.Now() + tmpl := &x509.Certificate{ + SerialNumber: serial, + Subject: pkix.Name{CommonName: name}, + // Peers verify by pin, not by name or time, so the validity window + // only has to satisfy the TLS libraries on both ends. Backdate a + // little for clock skew. + NotBefore: now.Add(-time.Hour), + NotAfter: now.Add(10 * 365 * 24 * time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, pub, priv) + if err != nil { + return nil, err + } + leaf, err := x509.ParseCertificate(der) + if err != nil { + return nil, err + } + return &Identity{ + cert: tls.Certificate{Certificate: [][]byte{der}, PrivateKey: priv, Leaf: leaf}, + pin: PinOf(leaf), + }, nil +} + +// Pin returns the pin of the identity's certificate. +func (id *Identity) Pin() Pin { return id.pin } + +// Certificate returns the identity's TLS certificate. +func (id *Identity) Certificate() tls.Certificate { return id.cert } + +// ServerTLSConfig returns a TLS configuration for a server presenting id. +// Any client certificate is accepted at the handshake, so that a peer which +// hasn't verified this process yet can still connect to do so; the request +// handler decides what the presented certificate's pin (see [RequestPin]) +// is allowed to do. +func (id *Identity) ServerTLSConfig() *tls.Config { + return &tls.Config{ + Certificates: []tls.Certificate{id.cert}, + ClientAuth: tls.RequireAnyClientCert, + MinVersion: tls.VersionTLS13, + } +} + +// ErrPinMismatch is returned by a pinned client TLS configuration when the +// server presented a certificate other than the pinned one, which is what a +// peer restart looks like. +var ErrPinMismatch = errors.New("lansport: peer certificate does not match its pin") + +// ClientTLSConfig returns a TLS configuration for connecting to a peer while +// presenting id. If pin is nil, any server certificate is accepted; use that +// for the one connection made over the tailnet to learn a peer's pin, where +// WireGuard has already authenticated the endpoint (see [ResponsePin]). +// Otherwise the server certificate must match the pin exactly. +func (id *Identity) ClientTLSConfig(pin *Pin) *tls.Config { + cfg := &tls.Config{ + Certificates: []tls.Certificate{id.cert}, + MinVersion: tls.VersionTLS13, + // Verification is by pin, not by CA or name; see VerifyConnection. + InsecureSkipVerify: true, + } + if pin != nil { + want := *pin + cfg.VerifyConnection = func(cs tls.ConnectionState) error { + if len(cs.PeerCertificates) == 0 || PinOf(cs.PeerCertificates[0]) != want { + return ErrPinMismatch + } + return nil + } + } + return cfg +} + +// RequestPin returns the pin of the client certificate r was made with, if +// any. +func RequestPin(r *http.Request) (Pin, bool) { + if r.TLS == nil || len(r.TLS.PeerCertificates) == 0 { + return Pin{}, false + } + return PinOf(r.TLS.PeerCertificates[0]), true +} + +// ResponsePin returns the pin of the server certificate res was received +// over, if any. +func ResponsePin(res *http.Response) (Pin, bool) { + if res.TLS == nil || len(res.TLS.PeerCertificates) == 0 { + return Pin{}, false + } + return PinOf(res.TLS.PeerCertificates[0]), true +} diff --git a/lansport/lansport.go b/lansport/lansport.go new file mode 100644 index 0000000..bffc6cf --- /dev/null +++ b/lansport/lansport.go @@ -0,0 +1,771 @@ +// Copyright (c) Tailscale Inc & AUTHORS +// SPDX-License-Identifier: BSD-3-Clause + +// Package lansport gives a set of tailnet peers on one LAN a way to talk to +// each other directly over TLS, without pushing their bulk traffic through +// WireGuard, while still getting their trust from the tailnet. +// +// Each participating process runs a [Server]. It has a per-process TLS +// [Identity] and two listeners. The advert listener speaks plain HTTP and +// answers only connections that arrived through the tailnet (or from a +// statically trusted host); at [AdvertPath] it returns the process's +// [Advert]: the LAN address and port of its TLS listener, the [Pin] of its +// certificate, and a routing weight. The TLS listener serves the caller's +// handler to peers that present a verified peer's client certificate. +// +// A Server learns which nodes might be peers from a [Source], typically a +// tagpeers.Tracker through [TailnetSource]. Every few seconds it fetches each +// candidate's advert over the tailnet; when a candidate's pin or address is +// new or changed, it builds a fresh pinned *http.Transport to the LAN +// address, and it checks that the LAN path works through that transport. A +// peer is reachable when both steps succeeded on the last round. A +// restarted peer shows up as a new pin; a broken LAN path shows up in the +// LAN check. +// +// Callers get reachable peers from [Server.Peers], each with a Transport +// that always connects to that peer's LAN address regardless of the request +// URL's host, follow changes to that set with [Server.Subscribe], and learn +// which peer sent a request with [FromPeer]. Choosing which peer should +// handle a given key is left to the rendezvous package. +package lansport + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "maps" + "net" + "net/http" + "net/netip" + "os" + "sync" + "sync/atomic" + "time" + + "tailscale.com/net/netx" + "tailscale.com/net/tsaddr" + "tailscale.com/types/logger" +) + +// AdvertPath is the path at which a [Server] serves its [Advert], on both +// the advert listener (to tailnet-origin connections) and the TLS listener +// (to verified peers, as a LAN liveness check). +const AdvertPath = "/lansport" + +// defaultProbeInterval is how often each candidate's advert is fetched and +// its LAN path checked when [Config.ProbeInterval] is zero. +const defaultProbeInterval = 3 * time.Second + +// probeTimeout bounds one advert fetch or LAN check. +const probeTimeout = 3 * time.Second + +// Advert is what a [Server] publishes about itself at [AdvertPath]. +type Advert struct { + // Name is the process's display name. + Name string `json:"name"` + + // IPPort is the LAN address and port of the process's TLS listener. + IPPort string `json:"ipPort"` + + // TLSCertHash is the hex [Pin] of the certificate that listener + // presents. + TLSCertHash string `json:"tlsCertHash"` + + // Weight is the process's relative share of whatever peers divide + // among themselves, for the rendezvous package. 1 is normal. + Weight float64 `json:"weight"` +} + +// Peer is a reachable peer: something a caller can send requests to. +type Peer struct { + // Key is the peer's routing [Key]: its [Candidate.Key], or its + // advertised name if the candidate had none. + Key Key + + // Name is the peer's display name, from its advert. + Name string + + // Addr is the peer's LAN TLS address, from its advert. + Addr string + + // Weight is the peer's advertised routing weight. + Weight float64 + + // Transport makes HTTPS requests to the peer, pinned to its + // certificate and presenting ours. It connects to Addr regardless of + // the host in the request URL; use [Peer.URL] to build one. + Transport *http.Transport +} + +// URL returns an https URL for path on the peer. +func (p Peer) URL(path string) string { + return "https://" + p.Addr + path +} + +// Config configures a [Server]. +type Config struct { + // Source says which nodes might be peers. Required. + Source Source + + // Name is the advertised display name. If empty, the hostname is used. + Name string + + // Weight is the advertised routing weight. If zero, 1 is used. + Weight float64 + + // AdvertAddr is the listen address of the plain-HTTP advert server, + // such as ":7890". Peers must be told the same port. Required unless + // AdvertListener is set. + AdvertAddr string + AdvertListener net.Listener + + // TLSAddr is the listen address of the LAN TLS server, such as + // ":31367" or ":0"; peers learn the actual port from the advert. + // Required unless TLSListener is set. + TLSAddr string + TLSListener net.Listener + + // LANIP is the IPv4 address advertised for the TLS listener. If unset, + // the address of the interface carrying the default route is used. + LANIP netip.Addr + + // Handler serves requests from verified peers on the TLS listener. + // Required. + Handler http.Handler + + // Dial, if non-nil, replaces how adverts are fetched and how peers' + // LAN addresses are connected to. The real network is used otherwise. + // Tests use it, with the listeners above, to run several servers on an + // in-memory network; see the lansporttest package. + Dial netx.DialFunc + + // ResponseHeaderTimeout bounds how long requests through a peer's + // Transport wait for response headers. If zero, 30 seconds. + ResponseHeaderTimeout time.Duration + + // ProbeInterval is how often adverts are fetched and LAN paths + // checked. If zero, a few seconds. + ProbeInterval time.Duration + + // Logf receives log lines; nil discards them. + Logf logger.Logf +} + +// PeerState is everything a [Server] knows about one candidate, for +// debugging pages. +type PeerState struct { + Candidate Candidate + Advert *Advert // last successfully fetched; nil if none + Pin Pin // zero until an advert has been fetched + Reachable bool // advert fetched and LAN check passed on the last round + Since time.Time + LastAdvert time.Time // last successful advert fetch + AdvertErr string // from the last failed advert fetch; empty on success + LastLAN time.Time // last successful LAN check + LANRTT time.Duration + LANErr string // from the last failed LAN check; empty on success + Rebuilds int // transports built for this peer (pin or address changes) +} + +// Key returns the routing key the peer has, or will have once its advert is +// fetched: the candidate's key, else the advertised name, else empty. +func (st PeerState) Key() Key { + if st.Candidate.Key != "" { + return st.Candidate.Key + } + if st.Advert != nil { + return Key(st.Advert.Name) + } + return "" +} + +// Name returns the peer's advertised name, or the candidate's until then. +func (st PeerState) Name() string { + if st.Advert != nil { + return st.Advert.Name + } + return st.Candidate.Name +} + +// peerState is a candidate's state plus its transport. The copies in +// Server.peers are guarded by Server.mu; probe works on a copy of its own. +type peerState struct { + PeerState + transport *http.Transport +} + +// Server publishes this process's advert, serves peers over LAN TLS, and +// maintains pinned transports to reachable peers. Use [Listen]. +type Server struct { + cfg Config + logf logger.Logf + identity *Identity + advertLn net.Listener + tlsLn net.Listener + advertSrv *http.Server + tlsSrv *http.Server + advert Advert + self Peer // Key, Name, Addr, Weight of this process + advertJSON []byte + probeClient *http.Client + + // reachable is the current set of reachable peers by key, replaced + // wholesale after each probe round. subs are the channels handed out + // by Subscribe, each signaled when it changes. + reachable atomic.Pointer[map[Key]Peer] + subsMu sync.Mutex + subs []chan struct{} + + mu sync.Mutex + peers map[string]*peerState // by Candidate.AdvertAddr + + cancel context.CancelFunc + done chan struct{} +} + +// peerCtxKey marks request contexts on the TLS listener with the sending +// peer. +type peerCtxKey struct{} + +// FromPeer returns the verified peer that sent r, if r arrived on a +// [Server]'s TLS listener. +func FromPeer(r *http.Request) (Peer, bool) { + p, ok := r.Context().Value(peerCtxKey{}).(Peer) + return p, ok +} + +// Listen starts a Server: it opens both listeners, publishes the advert, +// and begins probing the source's candidates. +func Listen(cfg Config) (*Server, error) { + if cfg.Source == nil { + return nil, errors.New("lansport: Source is required") + } + if cfg.Handler == nil { + return nil, errors.New("lansport: Handler is required") + } + if cfg.Logf == nil { + cfg.Logf = logger.Discard + } + if cfg.Name == "" { + h, err := os.Hostname() + if err != nil { + return nil, fmt.Errorf("lansport: Name not set and hostname unavailable: %w", err) + } + cfg.Name = h + } + if cfg.Weight == 0 { + cfg.Weight = 1 + } + if cfg.ResponseHeaderTimeout == 0 { + cfg.ResponseHeaderTimeout = 30 * time.Second + } + if cfg.ProbeInterval == 0 { + cfg.ProbeInterval = defaultProbeInterval + } + if cfg.Dial == nil { + cfg.Dial = (&net.Dialer{Timeout: probeTimeout, KeepAlive: 30 * time.Second}).DialContext + } + identity, err := NewIdentity(cfg.Name) + if err != nil { + return nil, fmt.Errorf("lansport: generating identity: %w", err) + } + + advertLn := cfg.AdvertListener + if advertLn == nil { + if cfg.AdvertAddr == "" { + return nil, errors.New("lansport: AdvertAddr or AdvertListener is required") + } + advertLn, err = net.Listen("tcp", cfg.AdvertAddr) + if err != nil { + return nil, fmt.Errorf("lansport: advert listener: %w", err) + } + } + tlsLn := cfg.TLSListener + if tlsLn == nil { + if cfg.TLSAddr == "" { + advertLn.Close() + return nil, errors.New("lansport: TLSAddr or TLSListener is required") + } + tlsLn, err = net.Listen("tcp", cfg.TLSAddr) + if err != nil { + advertLn.Close() + return nil, fmt.Errorf("lansport: TLS listener: %w", err) + } + } + lanIP := cfg.LANIP + if !lanIP.IsValid() { + lanIP, err = defaultLANIP() + if err != nil { + advertLn.Close() + tlsLn.Close() + return nil, fmt.Errorf("lansport: choosing LAN address to advertise: %w", err) + } + } + tlsPort := tlsLn.Addr().(*net.TCPAddr).Port + s := &Server{ + cfg: cfg, + logf: cfg.Logf, + identity: identity, + advertLn: advertLn, + tlsLn: tlsLn, + advert: Advert{ + Name: cfg.Name, + IPPort: netip.AddrPortFrom(lanIP, uint16(tlsPort)).String(), + TLSCertHash: identity.Pin().String(), + Weight: cfg.Weight, + }, + probeClient: &http.Client{ + Transport: &http.Transport{ + DialContext: cfg.Dial, + ResponseHeaderTimeout: probeTimeout, + MaxIdleConnsPerHost: 1, + IdleConnTimeout: time.Minute, + }, + }, + peers: make(map[string]*peerState), + done: make(chan struct{}), + } + s.advertJSON, _ = json.Marshal(s.advert) + s.self = Peer{Name: cfg.Name, Addr: s.advert.IPPort, Weight: cfg.Weight} + if key, ok := cfg.Source.SelfKey(); ok { + s.self.Key = key + } + empty := map[Key]Peer{} + s.reachable.Store(&empty) + + s.advertSrv = &http.Server{ + Handler: http.HandlerFunc(s.serveAdvert), + ReadHeaderTimeout: 10 * time.Second, + ErrorLog: logger.StdLogger(cfg.Logf), + } + s.tlsSrv = &http.Server{ + Handler: http.HandlerFunc(s.serveTLS), + TLSConfig: identity.ServerTLSConfig(), + ReadHeaderTimeout: 10 * time.Second, + ErrorLog: logger.StdLogger(cfg.Logf), + } + go s.advertSrv.Serve(advertLn) + go s.tlsSrv.ServeTLS(tlsLn, "", "") + + ctx, cancel := context.WithCancel(context.Background()) + s.cancel = cancel + go s.run(ctx) + s.logf("lansport: %q advertising %s (pin %s) on %v; TLS for peers on %v", + cfg.Name, s.advert.IPPort, identity.Pin().Short(), advertLn.Addr(), tlsLn.Addr()) + return s, nil +} + +// Close stops probing and both listeners. +func (s *Server) Close() error { + s.cancel() + <-s.done + err := errors.Join(s.advertSrv.Close(), s.tlsSrv.Close()) + s.probeClient.CloseIdleConnections() + s.mu.Lock() + defer s.mu.Unlock() + for _, p := range s.peers { + if p.transport != nil { + p.transport.CloseIdleConnections() + } + } + return err +} + +// Advert returns this process's advert. +func (s *Server) Advert() Advert { return s.advert } + +// Identity returns this process's TLS identity. +func (s *Server) Identity() *Identity { return s.identity } + +// Self describes this process as a Peer: its Key (if the source knows it), +// Name, Addr, and Weight. Its Transport is nil. +func (s *Server) Self() Peer { + if s.self.Key == "" { + if key, ok := s.cfg.Source.SelfKey(); ok { + s.self.Key = key + } + } + return s.self +} + +// TLSAddr returns the address of the LAN TLS listener. +func (s *Server) TLSAddr() net.Addr { return s.tlsLn.Addr() } + +// AdvertAddr returns the address of the advert listener. +func (s *Server) AdvertAddr() net.Addr { return s.advertLn.Addr() } + +// Peers returns the currently reachable peers by [Key]. The map must not be +// modified. +func (s *Server) Peers() map[Key]Peer { + return *s.reachable.Load() +} + +// Subscribe returns a channel that receives a value when the reachable +// set, or a reachable peer's Transport or Weight, changes. Signals are +// coalesced: a slow receiver sees one signal for many changes and should +// re-read [Server.Peers]. Each call returns its own channel, so several +// parties can follow the server. +func (s *Server) Subscribe() <-chan struct{} { + ch := make(chan struct{}, 1) + s.subsMu.Lock() + s.subs = append(s.subs, ch) + s.subsMu.Unlock() + return ch +} + +// Snapshot returns the state of every candidate, for debugging pages. +func (s *Server) Snapshot() []PeerState { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]PeerState, 0, len(s.peers)) + for _, p := range s.peers { + out = append(out, p.PeerState) + } + return out +} + +// serveAdvert handles the plain-HTTP advert listener. +func (s *Server) serveAdvert(w http.ResponseWriter, r *http.Request) { + ap, err := netip.ParseAddrPort(r.RemoteAddr) + if err != nil || !s.cfg.Source.TrustedAddr(ap.Addr()) { + http.Error(w, "forbidden", http.StatusForbidden) + return + } + if r.URL.Path != AdvertPath { + http.Error(w, "not found", http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + w.Write(s.advertJSON) +} + +// serveTLS handles the LAN TLS listener: the advert for verified peers' +// liveness checks, and the caller's handler for everything else they send. +func (s *Server) serveTLS(w http.ResponseWriter, r *http.Request) { + pin, ok := RequestPin(r) + if !ok { + http.Error(w, "forbidden: no client certificate", http.StatusForbidden) + return + } + peer, ok := s.peerByPin(pin) + if !ok { + http.Error(w, "forbidden: unknown peer certificate", http.StatusForbidden) + return + } + if r.URL.Path == AdvertPath { + w.Header().Set("Content-Type", "application/json") + w.Write(s.advertJSON) + return + } + s.cfg.Handler.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), peerCtxKey{}, peer))) +} + +// peerByPin returns the peer whose last fetched advert declared pin, if +// any. A peer counts here as soon as its advert has been fetched, whether +// or not its LAN path has been checked yet, so that a peer which learned of +// us first can reach us. +func (s *Server) peerByPin(pin Pin) (Peer, bool) { + s.mu.Lock() + defer s.mu.Unlock() + for _, p := range s.peers { + if p.Advert != nil && p.Pin == pin { + return p.peer(), true + } + } + return Peer{}, false +} + +func (p *peerState) peer() Peer { + return Peer{ + Key: p.Key(), + Name: p.Advert.Name, + Addr: p.Advert.IPPort, + Weight: p.Advert.Weight, + Transport: p.transport, + } +} + +// run probes every candidate each ProbeInterval, and sooner when the source +// reports a change. +func (s *Server) run(ctx context.Context) { + defer close(s.done) + t := time.NewTicker(s.cfg.ProbeInterval) + defer t.Stop() + for { + s.probeAll(ctx) + select { + case <-ctx.Done(): + return + case <-t.C: + case <-s.cfg.Source.Changes(): + } + } +} + +// probeAll reconciles the candidate list with the source, probes every +// online candidate concurrently, and publishes the new reachable set. It is +// the only place peer state is mutated: probe works on a copy and hands +// back the result, which is applied here under mu. +func (s *Server) probeAll(ctx context.Context) { + cands := s.cfg.Source.Candidates() + selfKey, _ := s.cfg.Source.SelfKey() + + s.mu.Lock() + seen := make(map[string]bool) + var toProbe []peerState + for _, c := range cands { + if c.Key != "" && c.Key == selfKey { + continue + } + seen[c.AdvertAddr] = true + p, ok := s.peers[c.AdvertAddr] + if !ok { + p = &peerState{PeerState: PeerState{Candidate: c}} + s.peers[c.AdvertAddr] = p + s.logf("lansport: candidate %s at %s", c.Name, c.AdvertAddr) + } + p.Candidate = c + if c.Online { + toProbe = append(toProbe, *p) + } else { + p.Reachable = false + p.AdvertErr = "offline per source" + } + } + for addr, p := range s.peers { + if !seen[addr] { + s.logf("lansport: %s is no longer a candidate", p.Name()) + if p.transport != nil { + p.transport.CloseIdleConnections() + } + delete(s.peers, addr) + } + } + s.mu.Unlock() + + results := make([]peerState, len(toProbe)) + var wg sync.WaitGroup + for i, p := range toProbe { + wg.Go(func() { results[i] = s.probe(ctx, p) }) + } + wg.Wait() + + s.mu.Lock() + for _, res := range results { + p, ok := s.peers[res.Candidate.AdvertAddr] + if !ok { + // Removed by a concurrent round; nothing to update. + if res.transport != nil { + res.transport.CloseIdleConnections() + } + continue + } + if p.transport != nil && p.transport != res.transport { + p.transport.CloseIdleConnections() + } + *p = res + } + s.mu.Unlock() + s.publish() +} + +// probe fetches p's advert over the source's trusted path, rebuilds its +// transport if the pin or address changed, and checks the LAN path. It works +// on its copy of p and returns the updated state; it touches no shared +// state, so it needs no lock. +func (s *Server) probe(ctx context.Context, p peerState) peerState { + name := p.Name() + adv, err := s.fetchAdvert(ctx, p.Candidate.AdvertAddr) + now := time.Now() + if err != nil { + if p.Reachable { + s.logf("lansport: %s: fetching advert from %s: %v", name, p.Candidate.AdvertAddr, err) + } + p.Reachable = false + p.AdvertErr = err.Error() + return p + } + pin, err := ParsePin(adv.TLSCertHash) + if err != nil { + p.Reachable = false + p.AdvertErr = "bad tlsCertHash in advert: " + err.Error() + return p + } + if pin == s.identity.Pin() { + // The candidate list included this process (a static list shared + // by every member, say). + p.Reachable = false + p.AdvertErr = "that's me" + return p + } + p.AdvertErr = "" + p.LastAdvert = now + if p.Advert == nil || p.Pin != pin || p.Advert.IPPort != adv.IPPort { + if p.transport != nil { + s.logf("lansport: %s changed: now %s with pin %s", adv.Name, adv.IPPort, pin.Short()) + } else { + s.logf("lansport: %s is %s with pin %s", adv.Name, adv.IPPort, pin.Short()) + } + p.transport = s.newTransport(adv.IPPort, pin) + p.Rebuilds++ + p.Reachable = false + } + p.Advert = adv + p.Pin = pin + + start := time.Now() + err = s.checkLAN(ctx, p.transport, adv) + rtt := time.Since(start) + if err != nil { + if p.Reachable { + s.logf("lansport: %s: LAN check to %s failed: %v", adv.Name, adv.IPPort, err) + } + p.Reachable = false + p.LANErr = err.Error() + return p + } + if !p.Reachable { + s.logf("lansport: %s reachable at %s (%v)", adv.Name, adv.IPPort, rtt.Round(100*time.Microsecond)) + p.Since = now + } + p.Reachable = true + p.LastLAN = now + p.LANRTT = rtt + p.LANErr = "" + return p +} + +// fetchAdvert GETs a candidate's advert with plain HTTP. +func (s *Server) fetchAdvert(ctx context.Context, addr string) (*Advert, error) { + ctx, cancel := context.WithTimeout(ctx, probeTimeout) + defer cancel() + req, err := http.NewRequestWithContext(ctx, "GET", "http://"+addr+AdvertPath, nil) + if err != nil { + return nil, err + } + res, err := s.probeClient.Do(req) + if err != nil { + return nil, err + } + defer res.Body.Close() + if res.StatusCode != http.StatusOK { + return nil, fmt.Errorf("status %s", res.Status) + } + adv := new(Advert) + if err := json.NewDecoder(io.LimitReader(res.Body, 64<<10)).Decode(adv); err != nil { + return nil, fmt.Errorf("decoding advert: %w", err) + } + if _, err := netip.ParseAddrPort(adv.IPPort); err != nil { + return nil, fmt.Errorf("bad ipPort %q in advert", adv.IPPort) + } + if adv.Weight <= 0 { + return nil, fmt.Errorf("bad weight %v in advert", adv.Weight) + } + return adv, nil +} + +// checkLAN fetches the peer's advert over the LAN through its pinned +// transport and checks it agrees with the one fetched over the tailnet. +func (s *Server) checkLAN(ctx context.Context, transport *http.Transport, want *Advert) error { + ctx, cancel := context.WithTimeout(ctx, probeTimeout) + defer cancel() + req, err := http.NewRequestWithContext(ctx, "GET", "https://"+want.IPPort+AdvertPath, nil) + if err != nil { + return err + } + res, err := transport.RoundTrip(req) + if err != nil { + return err + } + defer res.Body.Close() + if res.StatusCode != http.StatusOK { + // Most likely the peer hasn't fetched our advert yet and so + // doesn't know our certificate; it will within its next round. + msg, _ := io.ReadAll(io.LimitReader(res.Body, 200)) + return fmt.Errorf("status %s: %s", res.Status, msg) + } + var got Advert + if err := json.NewDecoder(io.LimitReader(res.Body, 64<<10)).Decode(&got); err != nil { + return fmt.Errorf("decoding advert over LAN: %w", err) + } + if got.TLSCertHash != want.TLSCertHash { + return fmt.Errorf("LAN address %s answered with pin %s, expected %s", want.IPPort, got.TLSCertHash, want.TLSCertHash) + } + return nil +} + +// newTransport returns a transport that connects to ipPort, pinned to pin, +// presenting our identity. +func (s *Server) newTransport(ipPort string, pin Pin) *http.Transport { + return &http.Transport{ + DialContext: func(ctx context.Context, network, _ string) (net.Conn, error) { + return s.cfg.Dial(ctx, "tcp", ipPort) + }, + TLSClientConfig: s.identity.ClientTLSConfig(&pin), + MaxIdleConnsPerHost: 64, + IdleConnTimeout: 90 * time.Second, + ResponseHeaderTimeout: s.cfg.ResponseHeaderTimeout, + // Peers negotiate their own encodings; don't add gzip on top. + DisableCompression: true, + } +} + +// publish recomputes the reachable set and signals if it differs from the +// last one. Two reachable peers with the same routing key is a +// misconfiguration (two static peers advertising the same name); neither +// is published, and both say why. +func (s *Server) publish() { + s.mu.Lock() + next := make(map[Key]Peer) + byKey := make(map[Key][]*peerState) + for _, p := range s.peers { + if p.Reachable { + byKey[p.Key()] = append(byKey[p.Key()], p) + } + } + for key, ps := range byKey { + if len(ps) > 1 { + for _, p := range ps { + p.Reachable = false + p.LANErr = fmt.Sprintf("routing key %q is shared by %d peers; names must be unique", key, len(ps)) + } + continue + } + next[key] = ps[0].peer() + } + s.mu.Unlock() + prev := *s.reachable.Load() + if maps.EqualFunc(prev, next, func(a, b Peer) bool { + return a.Name == b.Name && a.Addr == b.Addr && a.Weight == b.Weight && a.Transport == b.Transport + }) { + return + } + s.reachable.Store(&next) + s.subsMu.Lock() + defer s.subsMu.Unlock() + for _, ch := range s.subs { + select { + case ch <- struct{}{}: + default: + } + } +} + +// defaultLANIP returns the IPv4 address of the interface carrying the +// default route, found by asking the kernel which source address it would +// use for an outbound packet; nothing is sent. Tailnet addresses are +// rejected, since the point is to bypass the tailnet. +func defaultLANIP() (netip.Addr, error) { + c, err := net.Dial("udp4", "192.0.2.1:9") // TEST-NET-1; never actually contacted + if err != nil { + return netip.Addr{}, err + } + defer c.Close() + ip := c.LocalAddr().(*net.UDPAddr).AddrPort().Addr().Unmap() + if !ip.Is4() || ip.IsLoopback() || tsaddr.IsTailscaleIP(ip) { + return netip.Addr{}, fmt.Errorf("default route uses %v; set Config.LANIP", ip) + } + return ip, nil +} diff --git a/lansport/lansport_test.go b/lansport/lansport_test.go new file mode 100644 index 0000000..f62b583 --- /dev/null +++ b/lansport/lansport_test.go @@ -0,0 +1,368 @@ +// Copyright (c) Tailscale Inc & AUTHORS +// SPDX-License-Identifier: BSD-3-Clause + +package lansport + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "log" + "net" + "net/http" + "net/http/httptest" + "net/netip" + "testing" + "testing/synctest" + "time" + + "github.com/tailscale/tb/lansport/lansporttest" + "github.com/tailscale/tb/tagpeers" + "github.com/tailscale/tb/tagpeers/tagpeerstest" +) + +// probeInterval is the probe interval the tests configure. Under synctest +// it is virtual, so it costs nothing to wait out. +const probeInterval = time.Second + +// rounds lets n probe rounds happen and everything they trigger settle. +func rounds(n int) { + for range n { + time.Sleep(probeInterval) + synctest.Wait() + } +} + +var ( + ipA = netip.MustParseAddr("127.0.0.1") + ipB = netip.MustParseAddr("127.0.0.2") +) + +// echoName is a handler that answers with the name it was given, so tests +// can tell which server they reached through a transport, and which peer +// the server thought sent the request. +func echoName(name string) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + from, _ := FromPeer(r) + fmt.Fprintf(w, "%s saw %s", name, from.Name) + }) +} + +// newServer starts a lansport server named name at ip on lan, with its +// advert listener at advertAddr, and closes it when the test ends. +func newServer(t testing.TB, lan *lansporttest.LAN, name string, ip netip.Addr, advertAddr string, src Source) *Server { + t.Helper() + s, err := Listen(Config{ + Source: src, + Name: name, + AdvertListener: lan.Listen(advertAddr), + TLSListener: lan.Listen(net.JoinHostPort(ip.String(), "0")), + LANIP: ip, + Dial: lan.DialFrom(ip), + Handler: echoName(name), + ProbeInterval: probeInterval, + Logf: func(f string, args ...any) { t.Logf(name+": "+f, args...) }, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { s.Close() }) + return s +} + +// newStaticPair starts two servers named a and b that know each other +// through static sources and lets them reach each other. a is on ipA, b on +// ipB, both with adverts on port 7890. +func newStaticPair(t testing.TB, lan *lansporttest.LAN) (a, b *Server) { + t.Helper() + advA, advB := net.JoinHostPort(ipA.String(), "7890"), net.JoinHostPort(ipB.String(), "7890") + a = newServer(t, lan, "a", ipA, advA, StaticSource("a", advB)) + b = newServer(t, lan, "b", ipB, advB, StaticSource("b", advA)) + // Round 0 runs at Listen: a can't reach b's advert yet (b didn't + // exist), b fetches a's advert but a doesn't know b's pin. Round 1: a + // fetches b's advert and its LAN check passes, since b knows a. b's + // LAN check races a's fetch, so allow one more round for it. + rounds(2) + if len(a.Peers()) != 1 || len(b.Peers()) != 1 { + t.Fatalf("after two rounds: a sees %v, b sees %v", a.Peers(), b.Peers()) + } + return a, b +} + +// get fetches path from peer p and returns the status and body. +func get(t testing.TB, p Peer, path string) (int, string) { + t.Helper() + req, _ := http.NewRequest("GET", p.URL(path), nil) + res, err := p.Transport.RoundTrip(req) + if err != nil { + t.Fatalf("GET %s: %v", p.URL(path), err) + } + defer res.Body.Close() + body, _ := io.ReadAll(res.Body) + return res.StatusCode, string(body) +} + +func TestStaticPair(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + lan := new(lansporttest.LAN) + a, b := newStaticPair(t, lan) + + // Static peers are keyed by their advertised names, matching what + // each side uses for itself, so routing tables agree. + pb, ok := a.Peers()["b"] + if !ok { + t.Fatalf("a's peers = %v, want b", a.Peers()) + } + if a.Self().Key != "a" || pb.Weight != 1 || pb.Addr != b.TLSAddr().String() { + t.Errorf("a.Self = %+v, a's view of b = %+v", a.Self(), pb) + } + + // Requests through the transport reach the peer's handler, which + // knows who sent them. + if code, body := get(t, pb, "/hello"); code != 200 || body != "b saw a" { + t.Errorf("GET via transport: %d %q", code, body) + } + + // Anyone else on the LAN gets nothing from the TLS port, not even + // the advert. + rogue, _ := NewIdentity("rogue") + c := &http.Client{Transport: &http.Transport{ + DialContext: lan.DialFrom(netip.MustParseAddr("127.0.0.9")), + TLSClientConfig: rogue.ClientTLSConfig(nil), + }} + for _, path := range []string{AdvertPath, "/hello"} { + res, err := c.Get("https://" + b.TLSAddr().String() + path) + if err != nil { + t.Fatal(err) + } + res.Body.Close() + if res.StatusCode != http.StatusForbidden { + t.Errorf("rogue GET %s on TLS port: %d, want 403", path, res.StatusCode) + } + } + c.CloseIdleConnections() + + // A peer going away drops out at the next round. + b.Close() + rounds(1) + if len(a.Peers()) != 0 { + t.Errorf("a still sees %v after b closed", a.Peers()) + } + snap := a.Snapshot() + if len(snap) != 1 || snap[0].Reachable || snap[0].AdvertErr == "" { + t.Errorf("snapshot after b left = %+v", snap) + } + }) +} + +// TestRestartRepins covers a peer restarting: a new certificate behind the +// same addresses is detected through the advert, the transport is rebuilt, +// and requests flow again. +func TestRestartRepins(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + lan := new(lansporttest.LAN) + a, b := newStaticPair(t, lan) + oldTransport := a.Peers()["b"].Transport + advB, tlsB := b.AdvertAddr().String(), b.TLSAddr().String() + b.Close() + rounds(1) + if len(a.Peers()) != 0 { + t.Fatalf("a still sees %v after b closed", a.Peers()) + } + + b2, err := Listen(Config{ + Source: StaticSource("b", a.AdvertAddr().String()), + Name: "b", + AdvertListener: lan.Listen(advB), + TLSListener: lan.Listen(tlsB), + LANIP: ipB, + Dial: lan.DialFrom(ipB), + Handler: echoName("b2"), + ProbeInterval: probeInterval, + Logf: t.Logf, + }) + if err != nil { + t.Fatal(err) + } + defer b2.Close() + rounds(2) + pb, ok := a.Peers()["b"] + if !ok { + t.Fatalf("a doesn't see the restarted b: %v", a.Peers()) + } + if pb.Transport == oldTransport { + t.Error("transport not rebuilt after restart") + } + if code, body := get(t, pb, "/x"); code != 200 || body != "b2 saw a" { + t.Errorf("GET after restart: %d %q", code, body) + } + for _, st := range a.Snapshot() { + if st.Rebuilds != 2 { + t.Errorf("rebuilds = %d, want 2 (initial and after restart)", st.Rebuilds) + } + } + }) +} + +// TestLANPathBroken covers an advert that is fetched fine over the trusted +// path but names a LAN address nothing answers at: the peer is not +// reachable, and the snapshot says why. +func TestLANPathBroken(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + lan := new(lansporttest.LAN) + liar, _ := NewIdentity("liar") + liarLn := lan.Listen("127.0.0.3:7890") + liarSrv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprintf(w, `{"name":"liar","ipPort":"127.0.0.3:1","tlsCertHash":%q,"weight":1}`, liar.Pin()) + })} + go liarSrv.Serve(liarLn) + defer liarSrv.Close() + + a := newServer(t, lan, "a", ipA, "127.0.0.1:7890", StaticSource("a", liarLn.Addr().String())) + synctest.Wait() // round 0 + snap := a.Snapshot() + if len(snap) != 1 || snap[0].Advert == nil || snap[0].LANErr == "" || snap[0].Reachable { + t.Errorf("snapshot = %+v", snap) + } + if len(a.Peers()) != 0 { + t.Error("peer with broken LAN path is reachable") + } + }) +} + +// TestAdvertUntrusted checks that the advert port refuses addresses the +// source doesn't vouch for. +func TestAdvertUntrusted(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + lan := new(lansporttest.LAN) + a, _ := newStaticPair(t, lan) + for _, tt := range []struct { + from string + want int + }{ + {"127.0.0.2", http.StatusOK}, // b's host, in a's static list + {"127.0.0.9", http.StatusForbidden}, // nobody a knows + } { + c := &http.Client{Transport: &http.Transport{DialContext: lan.DialFrom(netip.MustParseAddr(tt.from))}} + res, err := c.Get("http://" + a.AdvertAddr().String() + AdvertPath) + if err != nil { + t.Fatal(err) + } + res.Body.Close() + c.CloseIdleConnections() + if res.StatusCode != tt.want { + t.Errorf("advert fetched from %s: %d, want %d", tt.from, res.StatusCode, tt.want) + } + } + }) +} + +// TestTailnetSource runs two servers whose peers come from fake tailscaleds +// through TailnetSource: adverts are fetched at the nodes' tailnet addresses +// on a shared advert port, and routing keys are the stable node IDs. +func TestTailnetSource(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + const tag = "tag:test" + lan := new(lansporttest.LAN) + nodeA := tagpeerstest.Node{ID: 1, Name: "a", IP: ipA, Tags: []string{tag}, Online: true} + nodeB := tagpeerstest.Node{ID: 2, Name: "b", IP: ipB, Tags: []string{tag}, Online: true} + tsA := tagpeerstest.New(t, nodeA, nodeB) + tsB := tagpeerstest.New(t, nodeB, nodeA) + + mk := func(name string, ts *tagpeerstest.Tailscaled, ip netip.Addr) *Server { + tr, err := tagpeers.Start(context.Background(), tagpeers.Config{Tag: tag, Dial: ts.Dial, Logf: t.Logf}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(tr.Close) + return newServer(t, lan, name, ip, net.JoinHostPort(ip.String(), "7890"), TailnetSource(tr, 7890)) + } + a := mk("a", tsA, ipA) + b := mk("b", tsB, ipB) + rounds(2) + pb, ok := a.Peers()["stable-2"] + if !ok || pb.Name != "b" || pb.Addr != b.TLSAddr().String() { + t.Fatalf("a's peers = %+v", a.Peers()) + } + if a.Self().Key != "stable-1" { + t.Errorf("a.Self().Key = %q", a.Self().Key) + } + if len(b.Peers()) != 1 { + t.Errorf("b's peers = %+v", b.Peers()) + } + if code, body := get(t, pb, "/y"); code != 200 || body != "b saw a" { + t.Errorf("GET via tailnet-sourced transport: %d %q", code, body) + } + + // Losing the tag drops the peer at once: the tracker signals, the + // server re-probes, and b is no longer a candidate. + untagged := nodeB + untagged.Tags = []string{"tag:other"} + tsA.ChangeNode(untagged) + synctest.Wait() + if len(a.Peers()) != 0 || len(a.Snapshot()) != 0 { + t.Errorf("after untagging b: peers %v, snapshot %+v", a.Peers(), a.Snapshot()) + } + }) +} + +// TestIdentityPinning checks the TLS configurations: a pinned client +// accepts the pinned server and rejects another, the server learns the +// client's pin, and an unpinned client reports the server's pin. +func TestIdentityPinning(t *testing.T) { + server, err := NewIdentity("server") + if err != nil { + t.Fatal(err) + } + client, err := NewIdentity("client") + if err != nil { + t.Fatal(err) + } + if server.Pin() == client.Pin() { + t.Fatal("two identities share a pin") + } + if p, err := ParsePin(server.Pin().String()); err != nil || p != server.Pin() { + t.Errorf("ParsePin round trip: %v, %v", p, err) + } + + var gotClientPin Pin + hs := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotClientPin, _ = RequestPin(r) + })) + hs.TLS = server.ServerTLSConfig() + hs.Config.ErrorLog = log.New(io.Discard, "", 0) // the wrong-pin handshake below is expected to fail + hs.StartTLS() + defer hs.Close() + + c := &http.Client{Transport: &http.Transport{TLSClientConfig: client.ClientTLSConfig(nil)}} + res, err := c.Get(hs.URL) + if err != nil { + t.Fatal(err) + } + res.Body.Close() + if pin, ok := ResponsePin(res); !ok || pin != server.Pin() { + t.Errorf("ResponsePin = %v, %v; want %v", pin, ok, server.Pin()) + } + if gotClientPin != client.Pin() { + t.Errorf("server saw client pin %v, want %v", gotClientPin, client.Pin()) + } + + want := server.Pin() + c = &http.Client{Transport: &http.Transport{TLSClientConfig: client.ClientTLSConfig(&want)}} + if res, err := c.Get(hs.URL); err != nil { + t.Errorf("pinned GET: %v", err) + } else { + res.Body.Close() + } + + wrong := client.Pin() + d := tls.Dialer{Config: client.ClientTLSConfig(&wrong)} + if conn, err := d.Dial("tcp", hs.Listener.Addr().String()); err == nil { + conn.Close() + t.Error("connection with the wrong pin succeeded") + } else if !errors.Is(err, ErrPinMismatch) { + t.Errorf("wrong pin: %v, want ErrPinMismatch", err) + } +} diff --git a/lansport/lansporttest/lan.go b/lansport/lansporttest/lan.go new file mode 100644 index 0000000..1a3769b --- /dev/null +++ b/lansport/lansporttest/lan.go @@ -0,0 +1,118 @@ +// Copyright (c) Tailscale Inc & AUTHORS +// SPDX-License-Identifier: BSD-3-Clause + +// Package lansporttest provides an in-memory network for testing code built +// on [lansport]: listeners at IPv4 addresses, and dialers that connect from a +// chosen source address, so servers see the remote addresses they would on +// a real LAN. Its connections are memnet pipes, which block in ways +// testing/synctest recognizes, so tests can run whole peer topologies under +// synctest with virtual time. +package lansporttest + +import ( + "context" + "errors" + "net" + "net/netip" + "sync" + + "tailscale.com/net/memnet" + "tailscale.com/net/netx" +) + +// LAN is an in-memory network. The zero value is ready to use. +type LAN struct { + mu sync.Mutex + lns map[netip.AddrPort]*listener + nextPort uint16 // for listeners asking for port 0 + nextSrc uint16 // ephemeral source ports for dialers +} + +// Listen returns a listener at addr, an ip:port; port 0 picks an unused +// one. It panics on a malformed address or one already listening, which in +// a test is a bug. +func (l *LAN) Listen(addr string) net.Listener { + ap, err := netip.ParseAddrPort(addr) + if err != nil { + panic("lansporttest: Listen(" + addr + "): " + err.Error()) + } + l.mu.Lock() + defer l.mu.Unlock() + if l.lns == nil { + l.lns = make(map[netip.AddrPort]*listener) + } + if ap.Port() == 0 { + for { + l.nextPort++ + ap = netip.AddrPortFrom(ap.Addr(), 40000+l.nextPort) + if _, taken := l.lns[ap]; !taken { + break + } + } + } else if _, taken := l.lns[ap]; taken { + panic("lansporttest: Listen(" + addr + "): address in use") + } + ln := &listener{lan: l, addr: ap, ch: make(chan net.Conn), closed: make(chan struct{})} + l.lns[ap] = ln + return ln +} + +// DialFrom returns a dialer whose connections appear to come from src. +func (l *LAN) DialFrom(src netip.Addr) netx.DialFunc { + return func(ctx context.Context, network, addr string) (net.Conn, error) { + dst, err := netip.ParseAddrPort(addr) + if err != nil { + return nil, &net.OpError{Op: "dial", Net: network, Err: err} + } + l.mu.Lock() + ln := l.lns[dst] + l.nextSrc++ + srcAP := netip.AddrPortFrom(src, 50000+l.nextSrc) + l.mu.Unlock() + if ln == nil { + return nil, &net.OpError{Op: "dial", Net: network, Addr: net.TCPAddrFromAddrPort(dst), Err: errors.New("connection refused")} + } + c, s := memnet.NewTCPConn(srcAP, dst, 1<<20) + select { + case ln.ch <- s: + return c, nil + case <-ln.closed: + c.Close() + s.Close() + return nil, &net.OpError{Op: "dial", Net: network, Addr: net.TCPAddrFromAddrPort(dst), Err: errors.New("connection refused")} + case <-ctx.Done(): + c.Close() + s.Close() + return nil, &net.OpError{Op: "dial", Net: network, Addr: net.TCPAddrFromAddrPort(dst), Err: ctx.Err()} + } + } +} + +type listener struct { + lan *LAN + addr netip.AddrPort + ch chan net.Conn + closed chan struct{} + once sync.Once +} + +func (ln *listener) Accept() (net.Conn, error) { + select { + case c := <-ln.ch: + return c, nil + case <-ln.closed: + return nil, net.ErrClosed + } +} + +func (ln *listener) Close() error { + ln.once.Do(func() { + close(ln.closed) + ln.lan.mu.Lock() + delete(ln.lan.lns, ln.addr) + ln.lan.mu.Unlock() + }) + return nil +} + +func (ln *listener) Addr() net.Addr { return net.TCPAddrFromAddrPort(ln.addr) } diff --git a/lansport/source.go b/lansport/source.go new file mode 100644 index 0000000..930ce26 --- /dev/null +++ b/lansport/source.go @@ -0,0 +1,138 @@ +// Copyright (c) Tailscale Inc & AUTHORS +// SPDX-License-Identifier: BSD-3-Clause + +package lansport + +import ( + "net" + "net/netip" + "slices" + "strconv" + + "github.com/tailscale/tb/tagpeers" +) + +// Key identifies a peer for routing purposes. Every member of a pool must +// have a distinct Key, and each member's Key must be the same on every +// other member, since the rendezvous package hashes on it to decide which +// member owns what: two members that disagree about a Key route some keys +// differently. It should also survive a node renaming or changing address, +// or those events move ownership around. A tailnet node's stable node ID +// is the usual choice; a static peer's advertised name is the fallback. +type Key string + +// Candidate is a node that might be a peer, as reported by a [Source]: where +// to fetch its advert from, over a path the source vouches for. +type Candidate struct { + // Key is the node's routing key. If empty, the name the node + // advertises is used, which must then be unique. + Key Key + + // Name is for display until the node's advert supplies its own. + Name string + + // AdvertAddr is the host:port to fetch the node's advert from with + // plain HTTP. The source guarantees that whatever answers there is the + // node (for a tailnet source, WireGuard does). + AdvertAddr string + + // Online reports whether the source believes the node is up. Offline + // candidates are kept but not probed. + Online bool +} + +// Source tells a [Server] which nodes might be peers and which connections +// may fetch its advert. [TailnetSource] and [StaticSource] implement it. +type Source interface { + // Candidates returns the current candidates. + Candidates() []Candidate + + // Changes returns a channel that receives a value when Candidates may + // have changed. Signals may be coalesced. A Source has one consumer, + // its Server, so one channel is enough. + Changes() <-chan struct{} + + // TrustedAddr reports whether a connection from ip arrived over a path + // the source vouches for, and so may fetch this node's advert. + TrustedAddr(ip netip.Addr) bool + + // SelfKey returns this node's own routing [Key], if known: what other + // members use for this node, so the rendezvous table here agrees with + // theirs about which keys this node owns. + SelfKey() (_ Key, ok bool) +} + +// TailnetSource adapts a [tagpeers.Tracker] into a [Source]: every tracked +// node is a candidate whose advert is fetched at its tailnet address on +// advertPort, over WireGuard. +func TailnetSource(tr *tagpeers.Tracker, advertPort int) Source { + return &tailnetSource{tr: tr, port: advertPort, changes: tr.Subscribe()} +} + +type tailnetSource struct { + tr *tagpeers.Tracker + port int + changes <-chan struct{} +} + +func (s *tailnetSource) Candidates() []Candidate { + nodes := s.tr.Nodes() + out := make([]Candidate, 0, len(nodes)) + for _, n := range nodes { + ip, ok := n.IP() + if !ok { + continue + } + out = append(out, Candidate{ + Key: Key(n.StableID), + Name: n.Hostname, + AdvertAddr: netip.AddrPortFrom(ip, uint16(s.port)).String(), + Online: n.Online, + }) + } + return out +} + +func (s *tailnetSource) Changes() <-chan struct{} { return s.changes } +func (s *tailnetSource) TrustedAddr(ip netip.Addr) bool { return s.tr.IsTailnetAddr(ip) } + +func (s *tailnetSource) SelfKey() (_ Key, ok bool) { + self, ok := s.tr.Self() + if !ok || self.StableID == "" { + return "", false + } + return Key(self.StableID), true +} + +// StaticSource returns a [Source] with a fixed list of advert addresses, +// trusted because the operator configured them. It is for tests and for +// hosts without a tailscaled. Peers are keyed by the names they advertise, +// and the local node by selfName, so every member must have a distinct +// name. +func StaticSource(selfName string, advertAddrs ...string) Source { + s := &staticSource{selfKey: Key(selfName), changes: make(chan struct{})} + for _, a := range advertAddrs { + s.cands = append(s.cands, Candidate{Name: a, AdvertAddr: a, Online: true}) + if host, _, err := net.SplitHostPort(a); err == nil { + if ip, err := netip.ParseAddr(host); err == nil { + s.hosts = append(s.hosts, ip.Unmap()) + } + } + } + return s +} + +type staticSource struct { + selfKey Key + cands []Candidate + hosts []netip.Addr + changes chan struct{} // never signaled +} + +func (s *staticSource) Candidates() []Candidate { return slices.Clone(s.cands) } +func (s *staticSource) Changes() <-chan struct{} { return s.changes } +func (s *staticSource) TrustedAddr(ip netip.Addr) bool { return slices.Contains(s.hosts, ip.Unmap()) } +func (s *staticSource) SelfKey() (_ Key, ok bool) { return s.selfKey, s.selfKey != "" } +func (s *staticSource) String() string { + return "static list of " + strconv.Itoa(len(s.cands)) + " peers" +} diff --git a/rendezvous/rendezvous.go b/rendezvous/rendezvous.go new file mode 100644 index 0000000..75922fb --- /dev/null +++ b/rendezvous/rendezvous.go @@ -0,0 +1,105 @@ +// Copyright (c) Tailscale Inc & AUTHORS +// SPDX-License-Identifier: BSD-3-Clause + +// Package rendezvous assigns keys to members of a changing set with weighted +// rendezvous (highest random weight) hashing: every party with the same +// member list picks the same member for a key, adding or removing a member +// only moves the keys that member gains or loses, and a member's share of +// the keyspace is proportional to its weight. +// +// Throughout, a member is one of the parties that can own things (named by +// a string, such as a node ID) and a key is a thing to be owned (such as a +// cache entry's ID). +// +// The scoring is the weighted variant of rendezvous hashing from +// Schindelhauer and Schomaker, "Weighted Distributed Hash Tables" (SPAA +// 2005); see https://en.wikipedia.org/wiki/Rendezvous_hashing for the +// unweighted original by Thaler and Ravishankar and the weighted form. +// +// [Table] is the pure hashing. [Router] keeps a Table in step with a +// lansport.Server's reachable peers, so a caller can ask which peer, if +// any, should handle a key. +package rendezvous + +import ( + "math" + "slices" + "sync" + + "github.com/cespare/xxhash/v2" +) + +// Table picks a member for a key by weighted rendezvous hashing. +type Table struct { + mu sync.RWMutex + members []member +} + +// member is one party that can own keys. +type member struct { + name string + weight float64 +} + +// Set replaces the members, a map from member name to weight. Weights must +// be positive; a member with weight 2 receives about twice the keys of one +// with weight 1. Members with non-positive weight are ignored. +func (t *Table) Set(members map[string]float64) { + ms := make([]member, 0, len(members)) + for name, w := range members { + if w > 0 { + ms = append(ms, member{name: name, weight: w}) + } + } + t.mu.Lock() + t.members = ms + t.mu.Unlock() +} + +// Pick returns the name of the member that owns key, or false if there are +// no members. +func (t *Table) Pick(key string) (string, bool) { + t.mu.RLock() + defer t.mu.RUnlock() + best := -1 + var bestScore float64 + for i, m := range t.members { + if s := Score(m.name, m.weight, key); best < 0 || s > bestScore { + best, bestScore = i, s + } + } + if best < 0 { + return "", false + } + return t.members[best].name, true +} + +// Members returns the names of the current members, sorted. +func (t *Table) Members() []string { + t.mu.RLock() + defer t.mu.RUnlock() + out := make([]string, len(t.members)) + for i, m := range t.members { + out[i] = m.name + } + slices.Sort(out) + return out +} + +// Score returns the weighted rendezvous score of key for the member with +// the given name and weight. The member with the highest score owns the +// key. Every party computes the same score for the same inputs, so they +// agree on owners without coordination. +func Score(member string, weight float64, key string) float64 { + var d xxhash.Digest + d.Reset() + d.WriteString(member) + d.WriteString("\x00") + d.WriteString(key) + // The weighted form: -weight / ln(u), for u uniform in (0, 1) derived + // from hashing member and key together. Map the 64-bit hash to (0, 1); + // the +1 and +2 keep u strictly inside the interval so the logarithm + // is finite and negative. + u := (float64(d.Sum64()>>11) + 1) / float64(1<<53+2) + return -weight / math.Log(u) +} diff --git a/rendezvous/rendezvous_test.go b/rendezvous/rendezvous_test.go new file mode 100644 index 0000000..45b1052 --- /dev/null +++ b/rendezvous/rendezvous_test.go @@ -0,0 +1,83 @@ +// Copyright (c) Tailscale Inc & AUTHORS +// SPDX-License-Identifier: BSD-3-Clause + +package rendezvous + +import ( + "fmt" + "testing" +) + +func keys(n int) []string { + out := make([]string, n) + for i := range out { + out[i] = fmt.Sprintf("key-%d", i) + } + return out +} + +func TestTable(t *testing.T) { + var tb Table + if _, ok := tb.Pick("x"); ok { + t.Fatal("empty table picked something") + } + + tb.Set(map[string]float64{"a": 1, "b": 1, "c": 1}) + ks := keys(30000) + counts := map[string]int{} + owner := map[string]string{} + for _, k := range ks { + m, ok := tb.Pick(k) + if !ok { + t.Fatal("no pick") + } + if again, _ := tb.Pick(k); again != m { + t.Fatalf("pick of %s not deterministic", k) + } + counts[m]++ + owner[k] = m + } + for m, n := range counts { + // About a third each; wide margins so this never flakes while + // still catching a broken hash. + if n < 8500 || n > 11500 { + t.Errorf("%s owns %d of %d, want about a third", m, n, len(ks)) + } + } + + // Removing c moves only c's keys. + tb.Set(map[string]float64{"a": 1, "b": 1}) + moved := 0 + for _, k := range ks { + m, _ := tb.Pick(k) + switch { + case owner[k] == "c": + if m == "c" { + t.Fatal("removed member still picked") + } + moved++ + case m != owner[k]: + t.Fatalf("key %s moved from surviving member %s to %s", k, owner[k], m) + } + } + if moved != counts["c"] { + t.Errorf("moved %d keys, want %d", moved, counts["c"]) + } +} + +func TestWeights(t *testing.T) { + var tb Table + tb.Set(map[string]float64{"light": 1, "heavy": 3, "ignored": 0}) + counts := map[string]int{} + for _, k := range keys(40000) { + m, _ := tb.Pick(k) + counts[m]++ + } + if counts["ignored"] != 0 { + t.Errorf("zero-weight member picked %d times", counts["ignored"]) + } + ratio := float64(counts["heavy"]) / float64(counts["light"]) + if ratio < 2.7 || ratio > 3.3 { + t.Errorf("heavy/light = %.2f (heavy %d, light %d), want about 3", ratio, counts["heavy"], counts["light"]) + } +} diff --git a/rendezvous/router.go b/rendezvous/router.go new file mode 100644 index 0000000..289ceda --- /dev/null +++ b/rendezvous/router.go @@ -0,0 +1,94 @@ +// Copyright (c) Tailscale Inc & AUTHORS +// SPDX-License-Identifier: BSD-3-Clause + +package rendezvous + +import ( + "context" + "sync/atomic" + + "github.com/tailscale/tb/lansport" +) + +// Router picks which of a lansport.Server's peers, or the local process, +// should handle a key. It follows the server's reachable set, so a peer that +// goes away stops being picked within one probe round, and includes the +// local process with its own advertised weight. +type Router struct { + s *lansport.Server + cur atomic.Pointer[routing] + cancel context.CancelFunc + done chan struct{} +} + +// routing is one consistent view of the reachable peers and the table built +// from them. Router replaces it wholesale on each change, so a Pick never +// sees a table and a peer map from different rounds. +type routing struct { + table *Table + peers map[lansport.Key]lansport.Peer +} + +// NewRouter returns a Router following s. Close it when done. +func NewRouter(s *lansport.Server) *Router { + ctx, cancel := context.WithCancel(context.Background()) + r := &Router{s: s, cancel: cancel, done: make(chan struct{})} + changes := s.Subscribe() + r.update() + go r.run(ctx, changes) + return r +} + +// Close stops following the server. +func (r *Router) Close() { + r.cancel() + <-r.done +} + +func (r *Router) run(ctx context.Context, changes <-chan struct{}) { + defer close(r.done) + for { + select { + case <-ctx.Done(): + return + case <-changes: + r.update() + } + } +} + +// update rebuilds the table from the server's reachable peers and itself. +func (r *Router) update() { + peers := r.s.Peers() + members := make(map[string]float64, len(peers)+1) + for key, p := range peers { + members[string(key)] = p.Weight + } + self := r.s.Self() + if self.Key != "" { + members[string(self.Key)] = self.Weight + } + table := new(Table) + table.Set(members) + r.cur.Store(&routing{table: table, peers: peers}) +} + +// Pick returns the reachable peer that should handle key. It returns false +// when the local process should: because it is the owner, or because there +// are no reachable peers. +func (r *Router) Pick(key string) (lansport.Peer, bool) { + cur := r.cur.Load() + owner, ok := cur.table.Pick(key) + if !ok { + return lansport.Peer{}, false + } + p, ok := cur.peers[lansport.Key(owner)] + return p, ok +} + +// Members returns the keys currently in the table, sorted, including the +// local process's if known. It is for debugging pages: two routers whose +// members differ route some keys differently. +func (r *Router) Members() []string { + return r.cur.Load().table.Members() +} diff --git a/tagpeers/tagpeerstest/fake.go b/tagpeers/tagpeerstest/fake.go new file mode 100644 index 0000000..a740f13 --- /dev/null +++ b/tagpeers/tagpeerstest/fake.go @@ -0,0 +1,215 @@ +// Copyright (c) Tailscale Inc & AUTHORS +// SPDX-License-Identifier: BSD-3-Clause + +// Package tagpeerstest provides a fake tailscaled for testing code built on +// [tagpeers]: it serves the LocalAPI IPN bus watch endpoint over an +// in-memory connection, starting each stream with a status the test controls +// and pushing peer deltas as the test changes nodes. It works under +// testing/synctest. +package tagpeerstest + +import ( + "context" + "encoding/json" + "fmt" + "maps" + "net" + "net/http" + "net/netip" + "sync" + "testing" + + "tailscale.com/ipn" + "tailscale.com/ipn/ipnstate" + "tailscale.com/net/memnet" + "tailscale.com/tailcfg" + "tailscale.com/types/key" + "tailscale.com/types/views" + "tailscale.com/util/set" +) + +// Node describes a tailnet node for the fake. +type Node struct { + ID tailcfg.NodeID + Name string // hostname; the DNS name is derived from it + IP netip.Addr + Tags []string + Online bool +} + +func (d Node) peerStatus() *ipnstate.PeerStatus { + tags := views.SliceOf(d.Tags) + return &ipnstate.PeerStatus{ + ID: tailcfg.StableNodeID(fmt.Sprintf("stable-%d", d.ID)), + NodeID: d.ID, + HostName: d.Name, + DNSName: d.Name + ".example.ts.net.", + TailscaleIPs: []netip.Addr{d.IP}, + Tags: &tags, + Online: d.Online, + } +} + +func (d Node) netmapNode() *tailcfg.Node { + online := d.Online + return &tailcfg.Node{ + ID: d.ID, + StableID: tailcfg.StableNodeID(fmt.Sprintf("stable-%d", d.ID)), + Name: d.Name + ".example.ts.net.", + Addresses: []netip.Prefix{netip.PrefixFrom(d.IP, d.IP.BitLen())}, + Hostinfo: (&tailcfg.Hostinfo{Hostname: d.Name}).View(), + Tags: d.Tags, + Online: &online, + } +} + +// listenerName is the memnet address the fake listens at. It never leaves +// the process, so any name will do. +const listenerName = "tailscaled" + +// Tailscaled is a fake tailscaled. Hand its Dial method to the code under +// test (as tagpeers.Config.Dial) to make it talk to the fake. +type Tailscaled struct { + t testing.TB + + mu sync.Mutex + ln *memnet.Listener // nil while stopped + hs *http.Server + status ipnstate.Status + watchers set.Set[chan ipn.Notify] +} + +// New starts a fake tailscaled whose local node is self and whose tailnet +// peers are peers. It is stopped when the test ends. +func New(t testing.TB, self Node, peers ...Node) *Tailscaled { + t.Helper() + f := &Tailscaled{ + t: t, + watchers: make(set.Set[chan ipn.Notify]), + } + f.status = ipnstate.Status{Self: self.peerStatus(), Peer: map[key.NodePublic]*ipnstate.PeerStatus{}} + for _, p := range peers { + f.status.Peer[key.NewNode().Public()] = p.peerStatus() + } + f.Start() + t.Cleanup(f.Stop) + return f +} + +// Dial connects to the fake. Pass it as tagpeers.Config.Dial. While the +// fake is stopped it fails, as connecting to a dead tailscaled would. +func (f *Tailscaled) Dial(ctx context.Context, _, _ string) (net.Conn, error) { + f.mu.Lock() + ln := f.ln + f.mu.Unlock() + if ln == nil { + return nil, &net.OpError{Op: "dial", Net: "unix", Err: net.ErrClosed} + } + return ln.Dial(ctx, "tcp", listenerName) +} + +// Start serves the IPN bus watch endpoint. New calls it; tests call it +// again after Stop to simulate tailscaled coming back. +func (f *Tailscaled) Start() { + f.mu.Lock() + defer f.mu.Unlock() + if f.ln != nil { + f.t.Fatal("tagpeerstest: Start called while running") + } + f.ln = memnet.Listen(listenerName) + f.hs = &http.Server{Handler: http.HandlerFunc(f.serveWatch)} + go f.hs.Serve(f.ln) +} + +// Stop shuts the fake down, closing its listener and any open streams, as a +// tailscaled restart or crash would. It is a no-op while stopped. +func (f *Tailscaled) Stop() { + f.mu.Lock() + hs := f.hs + f.ln, f.hs = nil, nil + f.mu.Unlock() + if hs != nil { + hs.Close() + } +} + +func (f *Tailscaled) serveWatch(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/localapi/v0/watch-ipn-bus" { + http.Error(w, "not found", http.StatusNotFound) + return + } + ch := make(chan ipn.Notify, 16) + f.mu.Lock() + f.watchers.Add(ch) + initial := f.status + initial.Peer = maps.Clone(f.status.Peer) + f.mu.Unlock() + defer func() { + f.mu.Lock() + f.watchers.Delete(ch) + f.mu.Unlock() + }() + + w.Header().Set("Content-Type", "application/json") + enc := json.NewEncoder(w) + rc := http.NewResponseController(w) + if err := enc.Encode(ipn.Notify{InitialStatus: &initial}); err != nil { + return + } + rc.Flush() + for { + select { + case n := <-ch: + if err := enc.Encode(n); err != nil { + return + } + rc.Flush() + case <-r.Context().Done(): + return + } + } +} + +// Notify sends n on every open stream. +func (f *Tailscaled) Notify(n ipn.Notify) { + f.mu.Lock() + defer f.mu.Unlock() + for ch := range f.watchers { + ch <- n + } +} + +// ChangeNode announces d as added or replaced, as control does when a node +// appears or changes (without the peer patches bit, that includes coming +// online or offline). +func (f *Tailscaled) ChangeNode(d Node) { + f.mu.Lock() + for k, p := range f.status.Peer { + if p.NodeID == d.ID { + delete(f.status.Peer, k) + } + } + f.status.Peer[key.NewNode().Public()] = d.peerStatus() + f.mu.Unlock() + f.Notify(ipn.Notify{PeersChanged: []*tailcfg.Node{d.netmapNode()}}) +} + +// RemoveNode announces that node id left the tailnet. +func (f *Tailscaled) RemoveNode(id tailcfg.NodeID) { + f.mu.Lock() + for k, p := range f.status.Peer { + if p.NodeID == id { + delete(f.status.Peer, k) + } + } + f.mu.Unlock() + f.Notify(ipn.Notify{PeersRemoved: []tailcfg.NodeID{id}}) +} + +// ChangeSelf announces that the local node changed. +func (f *Tailscaled) ChangeSelf(d Node) { + f.mu.Lock() + f.status.Self = d.peerStatus() + f.mu.Unlock() + f.Notify(ipn.Notify{SelfChange: d.netmapNode()}) +} diff --git a/tagpeers/tracker.go b/tagpeers/tracker.go new file mode 100644 index 0000000..c2cec43 --- /dev/null +++ b/tagpeers/tracker.go @@ -0,0 +1,461 @@ +// Copyright (c) Tailscale Inc & AUTHORS +// SPDX-License-Identifier: BSD-3-Clause + +// Package tagpeers tracks the tailnet nodes that carry a given tag, using the +// local tailscaled's IPN bus. +// +// A [Tracker] tells a daemon which nodes carry the tag, as the control plane +// sees it: their tailnet addresses, hostnames, and whether tailscaled thinks +// they're online, updated as nodes appear, change, and leave. It says nothing +// about whether a node is reachable or what it's running; that is for the +// layer above, such as the lansport package, which turns tagged nodes into +// working LAN connections. +package tagpeers + +import ( + "context" + "errors" + "fmt" + "net/netip" + "slices" + "strings" + "sync" + "time" + + "tailscale.com/client/local" + "tailscale.com/ipn" + "tailscale.com/ipn/ipnstate" + "tailscale.com/net/netx" + "tailscale.com/net/tsaddr" + "tailscale.com/tailcfg" + "tailscale.com/types/logger" +) + +// watchMask selects what the IPN bus watcher receives from tailscaled: the +// current status once, then peer additions, replacements, and removals. +// Narrow per-field changes (online state, endpoints) arrive promoted to full +// nodes, which is fine: only the fields in [Node] are compared. +const watchMask = ipn.NotifyInitialStatus | ipn.NotifyPeerChanges + +// maxBackoff caps the wait between attempts to reconnect to tailscaled's IPN +// bus after the stream fails. +const maxBackoff = 30 * time.Second + +// Node is a tailnet node carrying the tracked tag, or the local node. +type Node struct { + ID tailcfg.NodeID + StableID tailcfg.StableNodeID + Hostname string // the node's own hostname, not necessarily unique + DNSName string // MagicDNS FQDN with trailing dot, if any + Addrs []netip.Addr + Tags []string + Online bool +} + +// IP returns the node's IPv4 tailnet address, else its first address. +func (n Node) IP() (netip.Addr, bool) { + for _, ip := range n.Addrs { + if ip.Is4() { + return ip, true + } + } + if len(n.Addrs) > 0 { + return n.Addrs[0], true + } + return netip.Addr{}, false +} + +// equal reports whether n and o agree on every field. +func (n Node) equal(o Node) bool { + return n.ID == o.ID && n.StableID == o.StableID && n.Hostname == o.Hostname && + n.DNSName == o.DNSName && n.Online == o.Online && + slices.Equal(n.Addrs, o.Addrs) && slices.Equal(n.Tags, o.Tags) +} + +func nodeFromStatus(ps *ipnstate.PeerStatus) Node { + n := Node{ + ID: ps.NodeID, + StableID: ps.ID, + Hostname: ps.HostName, + DNSName: ps.DNSName, + Addrs: slices.Clone(ps.TailscaleIPs), + Online: ps.Online, + } + if ps.Tags != nil { + n.Tags = ps.Tags.AsSlice() + } + return n +} + +// nodeFromNetmap converts a netmap node. Online is unknown for some nodes +// (nil); prev supplies the value to keep in that case, defaulting to true. +func nodeFromNetmap(tn *tailcfg.Node, prev *Node) Node { + n := Node{ + ID: tn.ID, + StableID: tn.StableID, + DNSName: tn.Name, + Tags: slices.Clone(tn.Tags), + Online: true, + } + if tn.Hostinfo.Valid() && tn.Hostinfo.Hostname() != "" { + n.Hostname = tn.Hostinfo.Hostname() + } else if i := strings.IndexByte(tn.Name, '.'); i > 0 { + n.Hostname = tn.Name[:i] + } else { + n.Hostname = tn.Name + } + for _, pfx := range tn.Addresses { + n.Addrs = append(n.Addrs, pfx.Addr()) + } + if tn.Online != nil { + n.Online = *tn.Online + } else if prev != nil { + n.Online = prev.Online + } + return n +} + +// Config configures a [Tracker]. +type Config struct { + // Tag is the tag (such as "tag:my-pool") whose nodes to track. + Tag string + + // Socket is the path of tailscaled's LocalAPI socket. If empty, the + // platform's default is used, including the macOS GUI variants'. + Socket string + + // Dial, if non-nil, replaces how tailscaled's LocalAPI is reached and + // Socket is ignored. Tests use it to reach a fake tailscaled; see the + // tagpeerstest package. + Dial netx.DialFunc + + // Logf receives log lines; nil discards them. + Logf logger.Logf +} + +// State describes the tracker's connection to tailscaled. +type State struct { + // Connected reports whether an IPN bus stream is open and has + // delivered its initial status. Since is when that happened. + Connected bool + Since time.Time + + // LastError is why the most recent stream ended, or the most recent + // failure to open one; empty while connected. + LastError string + + // Events counts notifications applied across all streams, and + // LastEvent is when the most recent one arrived. + Events int64 + LastEvent time.Time +} + +// Tracker maintains the set of tailnet nodes carrying a tag by watching the +// local tailscaled's IPN bus. Use [Start] to create one. +type Tracker struct { + cfg Config + lc *local.Client + logf logger.Logf // never nil + + // subs are the channels handed out by Subscribe, each signaled + // (coalesced) whenever the tracked set or the local node changes. + // subsMu is never held together with mu. + subsMu sync.Mutex + subs []chan struct{} + + cancel context.CancelFunc + done chan struct{} + + // mu guards the fields below. It is never held together with subsMu: + // the merge functions decide under mu whether anything changed and + // signal subscribers only after releasing it. + mu sync.Mutex + nodes map[tailcfg.NodeID]Node + self Node + haveSelf bool + state State +} + +// Start begins tracking cfg.Tag's nodes. It returns before the first +// connection to tailscaled completes; see [Tracker.State] and +// [Tracker.Subscribe]. The tracker runs until ctx ends or [Tracker.Close] is +// called. +func Start(ctx context.Context, cfg Config) (*Tracker, error) { + if cfg.Tag == "" { + return nil, errors.New("tagpeers: Tag is required") + } + if !strings.HasPrefix(cfg.Tag, "tag:") { + return nil, fmt.Errorf("tagpeers: tag %q must start with \"tag:\"", cfg.Tag) + } + if cfg.Logf == nil { + cfg.Logf = logger.Discard + } + ctx, cancel := context.WithCancel(ctx) + t := &Tracker{ + cfg: cfg, + logf: cfg.Logf, + lc: &local.Client{ + Socket: cfg.Socket, + Dial: cfg.Dial, + // An explicit socket path means exactly that; only the default + // gets the platform's fallbacks (such as the macOS GUI + // variants' port lookup). + UseSocketOnly: cfg.Socket != "", + }, + cancel: cancel, + done: make(chan struct{}), + nodes: make(map[tailcfg.NodeID]Node), + } + go t.run(ctx) + return t, nil +} + +// Close stops the tracker and waits for its goroutine to exit. +func (t *Tracker) Close() { + t.cancel() + <-t.done +} + +// Subscribe returns a channel that receives a value whenever the set of +// tracked nodes, any tracked node's fields, or the local node changes. +// Signals are coalesced: a receiver that is slow sees one signal for many +// changes and should re-read [Tracker.Nodes]. Each call returns its own +// channel, so several parties can follow the tracker. +func (t *Tracker) Subscribe() <-chan struct{} { + ch := make(chan struct{}, 1) + t.subsMu.Lock() + t.subs = append(t.subs, ch) + t.subsMu.Unlock() + return ch +} + +// Nodes returns the nodes currently carrying the tag, sorted by ID. The +// local node is not included; see [Tracker.Self]. +func (t *Tracker) Nodes() []Node { + t.mu.Lock() + defer t.mu.Unlock() + out := make([]Node, 0, len(t.nodes)) + for _, n := range t.nodes { + out = append(out, n) + } + slices.SortFunc(out, func(a, b Node) int { return int(a.ID - b.ID) }) + return out +} + +// Self returns the local node as tailscaled reports it, if known yet. +func (t *Tracker) Self() (Node, bool) { + t.mu.Lock() + defer t.mu.Unlock() + return t.self, t.haveSelf +} + +// State returns the tracker's connection state. +func (t *Tracker) State() State { + t.mu.Lock() + defer t.mu.Unlock() + return t.state +} + +// IsTailnetAddr reports whether ip is a tailnet address: in the ranges +// Tailscale assigns from, or an address tailscaled reported for the local +// node or a tracked node. A TCP connection whose remote address satisfies +// this arrived through WireGuard, since a spoofed tailnet source on another +// network couldn't complete the handshake. +func (t *Tracker) IsTailnetAddr(ip netip.Addr) bool { + ip = ip.Unmap() + if tsaddr.IsTailscaleIP(ip) { + return true + } + t.mu.Lock() + defer t.mu.Unlock() + if t.haveSelf && slices.Contains(t.self.Addrs, ip) { + return true + } + for _, n := range t.nodes { + if slices.Contains(n.Addrs, ip) { + return true + } + } + return false +} + +func (t *Tracker) signal() { + t.subsMu.Lock() + defer t.subsMu.Unlock() + for _, ch := range t.subs { + select { + case ch <- struct{}{}: + default: + } + } +} + +// run keeps an IPN bus stream open, reconnecting with backoff when it +// fails. +func (t *Tracker) run(ctx context.Context) { + defer close(t.done) + backoff := time.Second + for ctx.Err() == nil { + start := time.Now() + err := t.watch(ctx) + if ctx.Err() != nil { + return + } + t.mu.Lock() + if t.state.LastError == "" { + t.logf("tagpeers: watching tailscaled: %v", err) + } + t.state.LastError = err.Error() + t.state.Connected = false + t.state.Since = time.Time{} + t.mu.Unlock() + if time.Since(start) > time.Minute { + backoff = time.Second + } + select { + case <-ctx.Done(): + return + case <-time.After(backoff): + } + backoff = min(backoff*2, maxBackoff) + } +} + +// watch runs one IPN bus stream until it fails or ctx ends. +func (t *Tracker) watch(ctx context.Context) error { + w, err := t.lc.WatchIPNBus(ctx, watchMask) + if err != nil { + return err + } + defer w.Close() + n, err := w.Next() + if err != nil { + return err + } + if n.InitialStatus == nil { + return errors.New("first notification carried no status; tailscaled may be too old for peer change notifications") + } + t.applyStatus(n.InitialStatus) + t.applyNotify(&n) + for { + n, err := w.Next() + if err != nil { + return err + } + t.applyNotify(&n) + } +} + +// applyStatus replaces the tracked set with the tagged nodes in st, the +// full status that starts every stream, and signals subscribers if anything +// changed. +func (t *Tracker) applyStatus(st *ipnstate.Status) { + if t.mergeStatus(st) { + t.signal() + } +} + +// mergeStatus does applyStatus's work under mu and reports whether the +// tracked set or the local node changed. +func (t *Tracker) mergeStatus(st *ipnstate.Status) (changed bool) { + now := time.Now() + t.mu.Lock() + defer t.mu.Unlock() + if t.state.LastError != "" { + t.logf("tagpeers: watching tailscaled works again") + } + t.state = State{Connected: true, Since: now, Events: t.state.Events + 1, LastEvent: now} + if st.Self != nil { + self := nodeFromStatus(st.Self) + if !t.haveSelf || !self.equal(t.self) { + t.self, t.haveSelf = self, true + changed = true + } + } + fresh := make(map[tailcfg.NodeID]Node) + for _, ps := range st.Peer { + n := nodeFromStatus(ps) + if slices.Contains(n.Tags, t.cfg.Tag) { + fresh[n.ID] = n + } + } + for id, n := range fresh { + if old, ok := t.nodes[id]; !ok || !old.equal(n) { + changed = true + if !ok { + t.logf("tagpeers: %s (%v) carries %s", n.Hostname, n.Addrs, t.cfg.Tag) + } + } + } + for id, old := range t.nodes { + if _, ok := fresh[id]; !ok { + t.logf("tagpeers: %s (%v) no longer carries %s", old.Hostname, old.Addrs, t.cfg.Tag) + changed = true + } + } + t.nodes = fresh + return changed +} + +// applyNotify applies the deltas in one notification and signals +// subscribers if anything changed. +func (t *Tracker) applyNotify(n *ipn.Notify) { + if len(n.PeersChanged) == 0 && len(n.PeersRemoved) == 0 && n.SelfChange == nil { + return + } + if t.mergeNotify(n) { + t.signal() + } +} + +// mergeNotify does applyNotify's work under mu and reports whether the +// tracked set or the local node changed. +func (t *Tracker) mergeNotify(n *ipn.Notify) (changed bool) { + now := time.Now() + t.mu.Lock() + defer t.mu.Unlock() + t.state.Events++ + t.state.LastEvent = now + if n.SelfChange != nil { + var prev *Node + if t.haveSelf { + prev = &t.self + } + self := nodeFromNetmap(n.SelfChange, prev) + if !t.haveSelf || !self.equal(t.self) { + t.self, t.haveSelf = self, true + changed = true + } + } + for _, tn := range n.PeersChanged { + old, had := t.nodes[tn.ID] + var prev *Node + if had { + prev = &old + } + node := nodeFromNetmap(tn, prev) + if !slices.Contains(node.Tags, t.cfg.Tag) { + if had { + t.logf("tagpeers: %s (%v) no longer carries %s", old.Hostname, old.Addrs, t.cfg.Tag) + delete(t.nodes, tn.ID) + changed = true + } + continue + } + if !had { + t.logf("tagpeers: %s (%v) carries %s", node.Hostname, node.Addrs, t.cfg.Tag) + } + if !had || !old.equal(node) { + t.nodes[tn.ID] = node + changed = true + } + } + for _, id := range n.PeersRemoved { + if old, ok := t.nodes[id]; ok { + t.logf("tagpeers: %s (%v) left the tailnet", old.Hostname, old.Addrs) + delete(t.nodes, id) + changed = true + } + } + return changed +} diff --git a/tagpeers/tracker_test.go b/tagpeers/tracker_test.go new file mode 100644 index 0000000..f1feb7d --- /dev/null +++ b/tagpeers/tracker_test.go @@ -0,0 +1,183 @@ +// Copyright (c) Tailscale Inc & AUTHORS +// SPDX-License-Identifier: BSD-3-Clause + +package tagpeers + +import ( + "context" + "net/netip" + "slices" + "testing" + "testing/synctest" + "time" + + "github.com/tailscale/tb/tagpeers/tagpeerstest" + "tailscale.com/tailcfg" +) + +const testTag = "tag:test-pool" + +var ( + self = tagpeerstest.Node{ID: 1, Name: "self", IP: netip.MustParseAddr("100.64.0.1"), Tags: []string{testTag}, Online: true} + nodeB = tagpeerstest.Node{ID: 2, Name: "b", IP: netip.MustParseAddr("100.64.0.2"), Tags: []string{testTag}, Online: true} + untagged = tagpeerstest.Node{ID: 3, Name: "untagged", IP: netip.MustParseAddr("100.64.0.3"), Tags: []string{"tag:other"}, Online: true} + offline = tagpeerstest.Node{ID: 4, Name: "offline", IP: netip.MustParseAddr("100.64.0.4"), Tags: []string{testTag}, Online: false} +) + +// ids returns the tracked node IDs, sorted. +func ids(tr *Tracker) []tailcfg.NodeID { + var out []tailcfg.NodeID + for _, n := range tr.Nodes() { + out = append(out, n.ID) + } + return out +} + +// expectSignal fails unless a change signal is pending on ch. It is called +// after synctest.Wait, so the tracker has already done everything it will. +func expectSignal(t testing.TB, ch <-chan struct{}, what string) { + t.Helper() + select { + case <-ch: + default: + t.Errorf("no change signal after %s", what) + } +} + +// expectNoSignal fails if a change signal is pending on ch. +func expectNoSignal(t testing.TB, ch <-chan struct{}, what string) { + t.Helper() + select { + case <-ch: + t.Errorf("unexpected change signal after %s", what) + default: + } +} + +func TestTracker(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + f := tagpeerstest.New(t, self, nodeB, untagged, offline) + tr, err := Start(context.Background(), Config{Tag: testTag, Dial: f.Dial, Logf: t.Logf}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(tr.Close) + changes := tr.Subscribe() + + synctest.Wait() + if !tr.State().Connected { + t.Fatalf("not connected: %+v", tr.State()) + } + expectSignal(t, changes, "initial status") + if got, want := ids(tr), []tailcfg.NodeID{2, 4}; !slices.Equal(got, want) { + t.Fatalf("nodes = %v, want %v", got, want) + } + s, ok := tr.Self() + if !ok || s.ID != 1 || s.Hostname != "self" { + t.Errorf("self = %+v, %v", s, ok) + } + nodes := tr.Nodes() + if nodes[0].Hostname != "b" || !nodes[0].Online || nodes[1].Hostname != "offline" || nodes[1].Online { + t.Errorf("nodes = %+v", nodes) + } + if ip, _ := nodes[0].IP(); ip != nodeB.IP { + t.Errorf("b's IP = %v", ip) + } + + // Coming online is a change worth signaling. + on := offline + on.Online = true + f.ChangeNode(on) + synctest.Wait() + expectSignal(t, changes, "node coming online") + if !tr.Nodes()[1].Online { + t.Error("offline node still offline") + } + + // Endpoint churn arrives as an identical full node and is not. + f.ChangeNode(on) + synctest.Wait() + expectNoSignal(t, changes, "unchanged node") + + // Losing the tag drops the node; regaining it brings it back. + retagged := nodeB + retagged.Tags = []string{"tag:other"} + f.ChangeNode(retagged) + synctest.Wait() + expectSignal(t, changes, "tag removal") + if got := ids(tr); !slices.Equal(got, []tailcfg.NodeID{4}) { + t.Errorf("nodes after tag removal = %v", got) + } + f.ChangeNode(nodeB) + synctest.Wait() + expectSignal(t, changes, "tag restored") + if got := ids(tr); !slices.Equal(got, []tailcfg.NodeID{2, 4}) { + t.Errorf("nodes after tag restored = %v", got) + } + + // Leaving the tailnet drops it too. + f.RemoveNode(nodeB.ID) + synctest.Wait() + expectSignal(t, changes, "node removal") + if got := ids(tr); !slices.Equal(got, []tailcfg.NodeID{4}) { + t.Errorf("nodes after removal = %v", got) + } + + // The local node's changes are tracked as well. + moved := self + moved.IP = netip.MustParseAddr("100.64.0.11") + f.ChangeSelf(moved) + synctest.Wait() + expectSignal(t, changes, "self change") + if s, _ := tr.Self(); !slices.Contains(s.Addrs, moved.IP) { + t.Errorf("self after change = %+v", s) + } + + // IsTailnetAddr knows the ranges and the listed addresses. + for _, tt := range []struct { + ip string + want bool + }{ + {"100.64.0.4", true}, // tracked node + {"100.64.0.11", true}, + {"100.99.1.1", true}, // in range + {"fd7a:115c:a1e0::1", true}, // in range + {"10.0.0.1", false}, + {"127.0.0.1", false}, + } { + if got := tr.IsTailnetAddr(netip.MustParseAddr(tt.ip)); got != tt.want { + t.Errorf("IsTailnetAddr(%s) = %v, want %v", tt.ip, got, tt.want) + } + } + + // tailscaled going away is reported and leaves the set alone; + // coming back re-seeds from the new stream's initial status, after + // the reconnect backoff. + f.Stop() + synctest.Wait() + if st := tr.State(); st.Connected || st.LastError == "" { + t.Errorf("state after tailscaled stopped = %+v", st) + } + if got := ids(tr); !slices.Equal(got, []tailcfg.NodeID{4}) { + t.Errorf("nodes after disconnect = %v", got) + } + f.ChangeNode(nodeB) // only the fake's status changes; no stream to deliver it + f.Start() + time.Sleep(time.Second) // the first reconnect backoff + synctest.Wait() + if !tr.State().Connected { + t.Fatalf("not reconnected: %+v", tr.State()) + } + if got := ids(tr); !slices.Equal(got, []tailcfg.NodeID{2, 4}) { + t.Errorf("nodes after reconnect = %v", got) + } + }) +} + +func TestStartRejectsBadConfig(t *testing.T) { + for _, cfg := range []Config{{}, {Tag: "ci-mac"}} { + if _, err := Start(context.Background(), cfg); err == nil { + t.Errorf("Start(%+v) succeeded, want error", cfg) + } + } +}