Compare commits

..

2 Commits

Author SHA1 Message Date
JackDoan 2df43dc218 Add experimental host_query API for local identity lookups
Programs running alongside nebula have no simple way to ask "who is this
vpn address?" when making authorization decisions, e.g. a nebula-aware
webapp that wants to identify an inbound connection by its source
address instead of presenting a login form. The existing surfaces are
the sshd admin interface (not scriptable from app code) and the
lighthouse-only DNS TXT lookup, which returns raw cert JSON over an
awkward transport.

This adds an opt-in `host_query` config section that serves a small
HTTP+JSON API on a unix socket or tcp address, requiring no client
library to consume:

  GET /v1/host?addr=<vpn addr>  identity of the host owning the address
                                (an established peer, or this node).
                                addr may include a port so a server can
                                pass a connection's RemoteAddr through
                                unparsed.
  GET /v1/self                  this node's own identity.

Responses carry the certificate-derived identity only: name, vpn
addresses, networks, unsafe networks, groups, fingerprint, issuer,
validity window, and cert version.

The self-vs-peer lookup logic is shared with the DNS TXT handler via a
new findCertificateForVpnAddr helper, which also swaps the panicking
GetDefaultCertificate call for the nil-returning accessor so a missing
certificate yields an empty answer instead of a crash.

The listener follows the statsServer lifecycle: the whole section is
reloadable via SIGHUP, including moving between socket paths and tcp
addresses. Unix sockets default to mode 0600, stale sockets left by an
unclean exit are removed at bind time, and a non-socket file at the
configured path is never replaced.
2026-07-24 10:06:25 -05:00
Nate Brown 72bf111209 Add an e2e Drop exit type and a roaming recovery measurement (#1819)
smoke-extra / freebsd-amd64 (push) Failing after 15s
smoke-extra / linux-amd64-ipv6disable (push) Failing after 15s
smoke-extra / netbsd-amd64 (push) Failing after 14s
smoke-extra / openbsd-amd64 (push) Failing after 15s
smoke-extra / linux-386 (push) Failing after 16s
smoke / Run multi node smoke test (push) Failing after 1m37s
Build and test / Static checks (push) Successful in 18s
Build and test / Test linux (push) Failing after 58s
Build and test / Test linux-boringcrypto (push) Failing after 2m45s
Build and test / Test linux-pkcs11 (push) Failing after 2m10s
Build and test / Cross-build linux-arm (push) Successful in 3m11s
Build and test / Cross-build linux-mips (push) Successful in 3m53s
Build and test / Cross-build linux-other (push) Successful in 3m16s
Build and test / Cross-build windows (push) Successful in 1m2s
Build and test / Cross-build freebsd (push) Successful in 1m36s
Build and test / Cross-build netbsd (push) Successful in 1m36s
Build and test / Cross-build openbsd (push) Successful in 1m37s
Build and test / Cross-build mobile (push) Successful in 3m23s
smoke-extra / Run windows smoke test (push) Has been cancelled
Build and test / Test macos (push) Has been cancelled
Build and test / Test windows (push) Has been cancelled
Build and test / CI status (push) Has been cancelled
2026-07-23 17:02:02 -05:00
28 changed files with 1078 additions and 735 deletions
-8
View File
@@ -60,12 +60,4 @@ jobs:
working-directory: ./.github/workflows/smoke
run: NAME="smoke-p256" ./smoke.sh
- name: setup docker image for multiport
working-directory: ./.github/workflows/smoke
run: NAME="smoke-multiport" MULTIPORT_TX=true MULTIPORT_RX=true MULTIPORT_HANDSHAKE=true ./build.sh
- name: run smoke
working-directory: ./.github/workflows/smoke
run: NAME="smoke-multiport" ./smoke.sh
timeout-minutes: 10
-4
View File
@@ -48,10 +48,6 @@ listen:
tun:
dev: ${TUN_DEV:-tun0}
multiport:
tx_enabled: ${MULTIPORT_TX:-false}
rx_enabled: ${MULTIPORT_RX:-false}
tx_handshake: ${MULTIPORT_HANDSHAKE:-false}
firewall:
inbound_action: reject
-4
View File
@@ -268,10 +268,6 @@ smoke-relay-docker: bin-docker
cd .github/workflows/smoke/ && ./build-relay.sh
cd .github/workflows/smoke/ && ./smoke-relay.sh
smoke-multiport-docker: bin-docker
cd .github/workflows/smoke/ && NAME="smoke-multiport" MULTIPORT_TX=true MULTIPORT_RX=true MULTIPORT_HANDSHAKE=true ./build.sh
cd .github/workflows/smoke/ && NAME="smoke-multiport" ./smoke.sh
smoke-docker-ipv6: export SMOKE_OVERLAY_IPV6 = 1
smoke-docker-ipv6: smoke-docker
-10
View File
@@ -1,10 +0,0 @@
package config
type MultiPortConfig struct {
Tx bool
Rx bool
TxBasePort uint16
TxPorts int
TxHandshake bool
TxHandshakeDelay int64
}
-3
View File
@@ -8,7 +8,6 @@ import (
"github.com/flynn/noise"
"github.com/slackhq/nebula/cert"
ct "github.com/slackhq/nebula/cert_test"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/handshake"
"github.com/slackhq/nebula/header"
"github.com/stretchr/testify/assert"
@@ -56,7 +55,6 @@ func runTestHandshake(t *testing.T) (initR, respR *handshake.Result) {
cert.Version2, initCreds, verifier,
func() (uint32, error) { return 1000, nil },
true, header.HandshakeIXPSK0,
config.MultiPortConfig{},
)
require.NoError(t, err)
@@ -64,7 +62,6 @@ func runTestHandshake(t *testing.T) (initR, respR *handshake.Result) {
cert.Version2, respCreds, verifier,
func() (uint32, error) { return 2000, nil },
false, header.HandshakeIXPSK0,
config.MultiPortConfig{},
)
require.NoError(t, err)
+4
View File
@@ -52,6 +52,7 @@ type Control struct {
sshStart func()
statsStart func()
dnsStart func()
infoAPIStart func()
lighthouseStart func()
networkChangeStart func(rebind func())
connectionManagerStart func(context.Context)
@@ -108,6 +109,9 @@ func (c *Control) Start() error {
if c.networkChangeStart != nil {
go c.networkChangeStart(c.RebindUDPServer)
}
if c.infoAPIStart != nil {
go c.infoAPIStart()
}
if c.connectionManagerStart != nil {
go c.connectionManagerStart(c.ctx)
}
+3 -22
View File
@@ -258,31 +258,12 @@ func (d *dnsServer) QueryCert(data string) string {
return ""
}
// The hostmap only ever contains peers we have handshaked with, so it never carries an entry for ourselves.
// Answer self lookups straight from the local cert state.
if cs := d.certState(); cs != nil && cs.myVpnAddrsTable != nil && cs.myVpnAddrsTable.Contains(ip) {
c := cs.GetDefaultCertificate()
if c == nil {
return ""
}
b, err := c.MarshalJSON()
if err != nil {
return ""
}
return string(b)
}
hostinfo := d.hostMap.QueryVpnAddr(ip)
if hostinfo == nil {
crt := findCertificateForVpnAddr(d.certState(), d.hostMap, ip)
if crt == nil {
return ""
}
q := hostinfo.GetCert()
if q == nil {
return ""
}
b, err := q.Certificate.MarshalJSON()
b, err := crt.MarshalJSON()
if err != nil {
return ""
}
+136
View File
@@ -0,0 +1,136 @@
//go:build e2e_testing
// +build e2e_testing
package e2e
import (
"testing"
"time"
"github.com/slackhq/nebula"
"github.com/slackhq/nebula/cert"
"github.com/slackhq/nebula/cert_test"
"github.com/slackhq/nebula/e2e/router"
"github.com/slackhq/nebula/udp"
)
// TestRecoveryTiming measures how long a tunnel takes to come back after the peer stops accepting our traffic,
// which is what a laptop waking on a new network looks like from the peer's side: its NAT has no state for where
// we are now, so everything we send disappears.
//
// It is a measurement, not a pass/fail assertion. Recovery is timed to the moment the peer punches back at us,
// since that is when its NAT opens and the tunnel is usable again.
//
// go test -tags e2e_testing -v -run TestRecoveryTiming ./e2e/
func TestRecoveryTiming(t *testing.T) {
for _, tc := range []struct {
name string
rebind bool
}{
{"no trigger", false},
{"rebind counter", true},
} {
t.Run(tc.name, func(t *testing.T) {
d, lost := measureRecovery(t, tc.rebind)
t.Logf("RESULT %-16s recovered in %-9v (%d packets lost)", tc.name, d.Round(time.Millisecond), lost)
})
}
}
// measureRecovery returns how long until the peer punched back, and how many of our packets died meanwhile. When
// rebind is true we call RebindUDPServer once the tunnel goes dark, which is what the darwin network change
// monitor does and what iOS has always done. When false, nothing tells nebula anything is wrong.
func measureRecovery(t *testing.T, rebind bool) (time.Duration, int) {
t.Helper()
ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version2, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
lhControl, lhVpnIpNet, lhUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "lh", "10.128.0.1/24", m{
"lighthouse": m{"am_lighthouse": true},
})
peerCfg := m{
"lighthouse": m{
"hosts": []any{lhVpnIpNet[0].Addr().String()},
"interval": 600,
"local_allow_list": m{
"10.0.0.0/24": true,
"::/0": false,
},
},
"static_host_map": m{
lhVpnIpNet[0].Addr().String(): []any{lhUdpAddr.String()},
},
}
myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "me", "10.128.0.2/24", peerCfg)
theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "them", "10.128.0.3/24", peerCfg)
r := router.NewR(t, lhControl, myControl, theirControl)
defer r.RenderFlow()
defer func() {
lhControl.Stop()
myControl.Stop()
theirControl.Stop()
}()
lhControl.Start()
myControl.Start()
theirControl.Start()
r.RouteFor(time.Millisecond * 500)
myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
myControl.InjectTunPacket(BuildTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("establish")))
r.RouteFor(time.Second)
if myControl.GetHostInfoByVpnAddr(theirVpnIpNet[0].Addr(), false) == nil {
t.Fatal("failed to establish the tunnel we are measuring")
}
r.RouteFor(time.Millisecond * 500)
// From here the peer's NAT has no state for us, everything we send it disappears
start := time.Now()
blackholed := 0
var recovered time.Duration
if rebind {
myControl.RebindUDPServer()
}
// Keep the tun busy the way someone retrying a stalled connection would
stop := make(chan struct{})
defer close(stop)
go func() {
tick := time.NewTicker(time.Millisecond * 200)
defer tick.Stop()
for {
select {
case <-stop:
return
case <-tick.C:
myControl.InjectTunPacket(BuildTunUDPPacket(
theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("retry")))
}
}
}()
r.RouteForAllExitFuncOrTimeout(time.Second*30, func(p *udp.Packet, c *nebula.Control) router.ExitType {
if c == theirControl && p.From == myControl.GetUDPAddr() {
blackholed++
return router.Drop
}
// The peer reaching us directly is the moment its NAT opened, whether that is a punch or a handshake
if c == myControl && p.From == theirUdpAddr {
recovered = time.Since(start)
return router.RouteAndExit
}
return router.KeepRouting
})
if recovered == 0 {
t.Fatalf("no recovery within 30s (%d packets blackholed)", blackholed)
}
return recovered, blackholed
}
+19 -2
View File
@@ -153,6 +153,9 @@ const (
ExitNow ExitType = 1
// RouteAndExit routes this packet and exits immediately afterwards
RouteAndExit ExitType = 2
// Drop discards this packet without delivering it and keeps routing. Use it to simulate a blackhole, such as
// a restrictive NAT refusing traffic from an address it has not seen.
Drop ExitType = 3
)
type ExitFunc func(packet *udp.Packet, receiver *nebula.Control) ExitType
@@ -163,7 +166,9 @@ type ExitFunc func(packet *udp.Packet, receiver *nebula.Control) ExitType
func NewR(t testing.TB, controls ...*nebula.Control) *R {
ctx, cancel := context.WithCancel(context.Background())
if err := os.MkdirAll("mermaid", 0755); err != nil {
// t.Name() contains a slash for subtests, so the flow log can land in a nested directory
fn := filepath.Join("mermaid", fmt.Sprintf("%s.md", t.Name()))
if err := os.MkdirAll(filepath.Dir(fn), 0755); err != nil {
panic(err)
}
@@ -174,7 +179,7 @@ func NewR(t testing.TB, controls ...*nebula.Control) *R {
outNat: make(map[outNatKey]netip.AddrPort),
flow: []flowEntry{},
ignoreFlows: []ignoreFlow{},
fn: filepath.Join("mermaid", fmt.Sprintf("%s.md", t.Name())),
fn: fn,
t: t,
cancelRender: cancel,
}
@@ -687,6 +692,10 @@ func (r *R) RouteExitFunc(sender *nebula.Control, whatDo ExitFunc) {
p.Release()
return
case Drop:
// Record it so the flow log shows the attempt, but never hand it to the receiver
r.unlockedInjectFlow(sender, receiver, p, false)
case KeepRouting:
fp := r.unlockedInjectFlow(sender, receiver, p, false)
receiver.InjectUDPPacket(p)
@@ -779,6 +788,10 @@ func (r *R) RouteForAllExitFuncOrTimeout(timeout time.Duration, whatDo ExitFunc)
p.Release()
return true
case Drop:
// Record it so the flow log shows the attempt, but never hand it to the receiver
r.unlockedInjectFlow(cm[x], receiver, p, false)
case KeepRouting:
fp := r.unlockedInjectFlow(cm[x], receiver, p, false)
receiver.InjectUDPPacket(p)
@@ -884,6 +897,10 @@ func (r *R) RouteForAllExitFunc(whatDo ExitFunc) {
p.Release()
return
case Drop:
// Record it so the flow log shows the attempt, but never hand it to the receiver
r.unlockedInjectFlow(cm[x], receiver, p, false)
case KeepRouting:
fp := r.unlockedInjectFlow(cm[x], receiver, p, false)
receiver.InjectUDPPacket(p)
+29 -41
View File
@@ -231,6 +231,35 @@ punchy:
# Overriding this to "" is the same as "/" and will allow overwriting any path on the host.
#sandbox_dir: /var/tmp/nebula-debug
# EXPERIMENTAL: this feature may change or disappear in the future.
# info_api exposes a small local HTTP+JSON API that lets other programs on
# this machine resolve a vpn address to its certificate identity (name, vpn
# addresses, groups, fingerprint, validity), e.g. for making authorization
# decisions about an inbound connection:
# GET /v1/host?addr=<vpn addr> - identity of the host owning the address: a
# peer with an active tunnel, or this node itself. `addr` may include a
# port (`192.168.100.7:54321`), which is ignored, so a connection's remote
# address can be passed through as is. Returns 404 when the address is
# unknown or has no active tunnel.
# GET /v1/self - this node's own identity.
# Identity answers can be trusted because nebula drops inbound packets whose
# source vpn address is not contained in the sender's certificate, so the
# source address of a connection arriving over the nebula interface is
# guaranteed to map to the certificate reported here.
# There is no authentication in this API; restrict access with unix socket
# file permissions.
# This whole section is reloadable.
#info_api:
# Toggles the feature
#enabled: false
# listen accepts a unix socket path as a unix:// URL with an absolute path:
#listen: unix:///var/run/nebula-info-api.sock
# File mode for the unix socket, as an octal string.
# The socket is created by nebula's user; to grant a group of local services
# access, place the socket in a directory with appropriate permissions
# (e.g. a systemd RuntimeDirectory) and relax this to "0660".
#socket_mode: "0600"
# EXPERIMENTAL: relay support for networks that can't establish direct connections.
relay:
# Relays are a list of Nebula IP's that peers can use to relay packets to me.
@@ -328,47 +357,6 @@ tun:
# SO_RCVBUFFORCE is used to avoid having to raise the system wide max
#use_system_route_table_buffer_size: 0
# EXPERIMENTAL: This option may change or disappear in the future.
# Multiport spreads outgoing UDP packets across multiple UDP send ports,
# which allows nebula to work around any issues on the underlay network.
# Some example issues this could work around:
# - UDP rate limits on a per flow basis.
# - Partial underlay network failure in which some flows work and some don't
# Agreement is done during the handshake to decide if multiport mode will
# be used for a given tunnel (one side must have tx_enabled set, the other
# side must have rx_enabled set)
#
# NOTE: you cannot use multiport on a host if you are relying on UDP hole
# punching to get through a NAT or firewall.
#
# NOTE: Linux only (uses raw sockets to send). Also currently only works
# with IPv4 underlay network remotes.
#
# The default values are listed below:
#multiport:
# This host support sending via multiple UDP ports.
#tx_enabled: false
#
# This host supports receiving packets sent from multiple UDP ports.
#rx_enabled: false
#
# How many UDP ports to use when sending. The lowest source port will be
# listen.port and go up to (but not including) listen.port + tx_ports.
#tx_ports: 100
#
# NOTE: All of your hosts must be running a version of Nebula that supports
# multiport if you want to enable this feature. Older versions of Nebula
# will be confused by these multiport handshakes.
#
# If handshakes are not getting a response, attempt to transmit handshakes
# using random UDP source ports (to get around partial underlay network
# failures).
#tx_handshake: false
#
# How many unresponded handshakes we should send before we attempt to
# send multiport handshakes.
#tx_handshake_delay: 2
# Configure logging level
logging:
# trace, debug, info, warn, or error. Default is info and is reloadable.
-28
View File
@@ -3,7 +3,6 @@ package firewall
import (
"encoding/json"
"fmt"
mathrand "math/rand"
"net/netip"
)
@@ -66,30 +65,3 @@ func (fp Packet) MarshalJSON() ([]byte, error) {
"Fragment": fp.Fragment,
})
}
// UDPSendPort calculates the UDP port to send from when using multiport mode.
// The result will be from [0, numBuckets)
func (fp Packet) UDPSendPort(numBuckets int) uint16 {
if numBuckets <= 1 {
return 0
}
// If there is no port (like an ICMP packet), pick a random UDP send port
if fp.LocalPort == 0 {
return uint16(mathrand.Intn(numBuckets))
}
// A decent enough 32bit hash function
// Prospecting for Hash Functions
// - https://nullprogram.com/blog/2018/07/31/
// - https://github.com/skeeto/hash-prospector
// [16 21f0aaad 15 d35a2d97 15] = 0.10760229515479501
x := (uint32(fp.LocalPort) << 16) | uint32(fp.RemotePort)
x ^= x >> 16
x *= 0x21f0aaad
x ^= x >> 15
x *= 0xd35a2d97
x ^= x >> 15
return uint16(x) % uint16(numBuckets)
}
+2 -10
View File
@@ -24,14 +24,6 @@ message NebulaHandshakeDetails {
uint64 Cookie = 4 [deprecated = true];
uint64 Time = 5;
uint32 CertVersion = 8;
MultiPortDetails InitiatorMultiPort = 6;
MultiPortDetails ResponderMultiPort = 7;
}
message MultiPortDetails {
bool RxSupported = 1;
bool TxSupported = 2;
uint32 BasePort = 3;
uint32 TotalPorts = 4;
// reserved for WIP multiport
reserved 6, 7;
}
-2
View File
@@ -8,7 +8,6 @@ import (
"github.com/flynn/noise"
"github.com/slackhq/nebula/cert"
ct "github.com/slackhq/nebula/cert_test"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/header"
"github.com/stretchr/testify/require"
)
@@ -72,7 +71,6 @@ func newTestMachine(
cs.version, cs.getCredential,
verifier, func() (uint32, error) { return localIndex, nil },
initiator, header.HandshakeIXPSK0,
config.MultiPortConfig{},
)
require.NoError(t, err)
return m
+1 -43
View File
@@ -3,13 +3,11 @@ package handshake
import (
"bytes"
"fmt"
"math"
"slices"
"time"
"github.com/flynn/noise"
"github.com/slackhq/nebula/cert"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/header"
)
@@ -41,10 +39,6 @@ type Result struct {
HandshakeTime uint64
MessageIndex uint64 // number of messages exchanged during the handshake
Initiator bool
MultiportRx bool
MultiportTx bool
MultiportBasePort uint16
}
// Machine drives a Noise handshake through N messages. It handles Noise
@@ -73,8 +67,6 @@ type Machine struct {
remoteCertSet bool
payloadSet bool
failed bool
multiport config.MultiPortConfig
}
// NewMachine creates a handshake state machine. The subtype determines both
@@ -88,7 +80,6 @@ func NewMachine(
allocIndex IndexAllocator,
initiator bool,
subtype header.MessageSubType,
multiport config.MultiPortConfig,
) (*Machine, error) {
info, err := subtypeInfoFor(subtype)
if err != nil {
@@ -117,8 +108,6 @@ func NewMachine(
Initiator: initiator,
Cipher: cred.cipherSuite,
},
multiport: multiport,
}, nil
}
@@ -309,7 +298,7 @@ func (m *Machine) processPayload(msg []byte, flags msgFlags) error {
}
// Assert the payload contains exactly what we expect
hasPayloadData := payload.InitiatorIndex != 0 || payload.ResponderIndex != 0 || payload.Time != 0 || payload.InitiatorMultiPort != nil || payload.ResponderMultiPort != nil
hasPayloadData := payload.InitiatorIndex != 0 || payload.ResponderIndex != 0 || payload.Time != 0
if hasPayloadData != flags.expectsPayload {
m.failed = true
return ErrUnexpectedContent
@@ -326,22 +315,8 @@ func (m *Machine) processPayload(msg []byte, flags msgFlags) error {
var remoteIndex uint32
if m.result.Initiator {
remoteIndex = payload.ResponderIndex
if payload.ResponderMultiPort != nil {
m.result.MultiportRx = payload.ResponderMultiPort.RxSupported
m.result.MultiportTx = payload.ResponderMultiPort.TxSupported
if payload.ResponderMultiPort.BasePort <= math.MaxUint16 {
m.result.MultiportBasePort = uint16(payload.ResponderMultiPort.BasePort)
}
}
} else {
remoteIndex = payload.InitiatorIndex
if payload.InitiatorMultiPort != nil {
m.result.MultiportRx = payload.InitiatorMultiPort.RxSupported
m.result.MultiportTx = payload.InitiatorMultiPort.TxSupported
if payload.InitiatorMultiPort.BasePort <= math.MaxUint16 {
m.result.MultiportBasePort = uint16(payload.InitiatorMultiPort.BasePort)
}
}
}
// The payload presence check above can be satisfied by Time alone, so a payload
// could still carry a zero index here. We need to reject it.
@@ -422,28 +397,11 @@ func (m *Machine) marshalOutgoing(flags msgFlags) ([]byte, error) {
if m.result.Initiator {
p.InitiatorIndex = m.result.LocalIndex
if m.multiport.Rx || m.multiport.Tx {
p.InitiatorMultiPort = &PayloadMultiPortDetails{
RxSupported: m.multiport.Rx,
TxSupported: m.multiport.Tx,
BasePort: uint32(m.multiport.TxBasePort),
TotalPorts: uint32(m.multiport.TxPorts),
}
}
} else {
p.ResponderIndex = m.result.LocalIndex
p.InitiatorIndex = m.result.RemoteIndex
if m.multiport.Rx || m.multiport.Tx {
p.ResponderMultiPort = &PayloadMultiPortDetails{
RxSupported: m.multiport.Rx,
TxSupported: m.multiport.Tx,
BasePort: uint32(m.multiport.TxBasePort),
TotalPorts: uint32(m.multiport.TxPorts),
}
}
}
p.Time = uint64(time.Now().UnixNano())
}
if flags.expectsCert {
cred := m.getCred(m.myVersion)
-3
View File
@@ -8,7 +8,6 @@ import (
"github.com/flynn/noise"
"github.com/slackhq/nebula/cert"
ct "github.com/slackhq/nebula/cert_test"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/header"
"github.com/slackhq/nebula/noiseutil"
"github.com/stretchr/testify/assert"
@@ -445,7 +444,6 @@ func TestMachineThreeMessagePattern(t *testing.T) {
initCS.getCredential, v,
func() (uint32, error) { return 1000, nil },
true, header.HandshakeXXPSK0,
config.MultiPortConfig{},
)
require.NoError(t, err)
@@ -454,7 +452,6 @@ func TestMachineThreeMessagePattern(t *testing.T) {
respCS.getCredential, v,
func() (uint32, error) { return 2000, nil },
false, header.HandshakeXXPSK0,
config.MultiPortConfig{},
)
require.NoError(t, err)
-139
View File
@@ -20,16 +20,6 @@ type Payload struct {
ResponderIndex uint32
Time uint64
CertVersion uint32
InitiatorMultiPort *PayloadMultiPortDetails
ResponderMultiPort *PayloadMultiPortDetails
}
type PayloadMultiPortDetails struct {
RxSupported bool
TxSupported bool
BasePort uint32
TotalPorts uint32
}
// Proto field numbers for NebulaHandshakeDetails
@@ -39,17 +29,6 @@ const (
fieldResponderIndex = 3 // uint32
fieldTime = 5 // uint64
fieldCertVersion = 8 // uint32
fieldInitiatorMultiPort = 6 // MultiPortDetails
fieldResponderMultiPort = 7 // MultiPortDetails
)
// Proto field numbers for MultiPortDetails
const (
fieldMultiportRxSupported = 1 // bool
fieldMultiportTxSupported = 2 // bool
fieldMultiportBasePort = 3 // uint32
fieldMultiportTotalPorts = 4 // uint32
)
// MarshalPayload encodes a handshake payload in protobuf wire format compatible
@@ -78,16 +57,6 @@ func MarshalPayload(out []byte, p Payload) []byte {
details = protowire.AppendTag(details, fieldCertVersion, protowire.VarintType)
details = protowire.AppendVarint(details, uint64(p.CertVersion))
}
if p.InitiatorMultiPort != nil {
details = protowire.AppendTag(details, fieldInitiatorMultiPort, protowire.BytesType)
details = protowire.AppendVarint(details, uint64(p.InitiatorMultiPort.size()))
details = p.InitiatorMultiPort.marshal(details)
}
if p.ResponderMultiPort != nil {
details = protowire.AppendTag(details, fieldResponderMultiPort, protowire.BytesType)
details = protowire.AppendVarint(details, uint64(p.ResponderMultiPort.size()))
details = p.ResponderMultiPort.marshal(details)
}
out = protowire.AppendTag(out, 1, protowire.BytesType)
out = protowire.AppendBytes(out, details)
@@ -95,23 +64,6 @@ func MarshalPayload(out []byte, p Payload) []byte {
return out
}
func (p PayloadMultiPortDetails) marshal(details []byte) []byte {
details = protowire.AppendTag(details, fieldMultiportRxSupported, protowire.VarintType)
details = protowire.AppendVarint(details, protowire.EncodeBool(p.RxSupported))
details = protowire.AppendTag(details, fieldMultiportTxSupported, protowire.VarintType)
details = protowire.AppendVarint(details, protowire.EncodeBool(p.TxSupported))
details = protowire.AppendTag(details, fieldMultiportBasePort, protowire.VarintType)
details = protowire.AppendVarint(details, uint64(p.BasePort))
details = protowire.AppendTag(details, fieldMultiportTotalPorts, protowire.VarintType)
details = protowire.AppendVarint(details, uint64(p.TotalPorts))
return details
}
func (p PayloadMultiPortDetails) size() int {
return 4 + 2 + protowire.SizeVarint(uint64(p.BasePort)) + protowire.SizeVarint(uint64(p.TotalPorts))
}
// UnmarshalPayload decodes a protobuf-encoded NebulaHandshake message.
func UnmarshalPayload(b []byte) (Payload, error) {
var p Payload
@@ -209,97 +161,6 @@ func unmarshalPayloadDetails(p *Payload, b []byte) error {
}
p.CertVersion = uint32(v)
b = b[n:]
case fieldInitiatorMultiPort:
if typ != protowire.BytesType {
return errInvalidHandshakeDetails
}
d, n := protowire.ConsumeBytes(b)
if n < 0 {
return errInvalidHandshakeMessage
}
b = b[n:]
p.InitiatorMultiPort = new(PayloadMultiPortDetails)
if err := unmarshalPayloadMultiPortDetails(p.InitiatorMultiPort, d); err != nil {
return err
}
case fieldResponderMultiPort:
if typ != protowire.BytesType {
return errInvalidHandshakeDetails
}
d, n := protowire.ConsumeBytes(b)
if n < 0 {
return errInvalidHandshakeMessage
}
b = b[n:]
p.ResponderMultiPort = new(PayloadMultiPortDetails)
if err := unmarshalPayloadMultiPortDetails(p.ResponderMultiPort, d); err != nil {
return err
}
default:
n := protowire.ConsumeFieldValue(num, typ, b)
if n < 0 {
return errInvalidHandshakeDetails
}
b = b[n:]
}
}
return nil
}
func unmarshalPayloadMultiPortDetails(p *PayloadMultiPortDetails, b []byte) error {
for len(b) > 0 {
num, typ, n := protowire.ConsumeTag(b)
if n < 0 {
return errInvalidHandshakeDetails
}
b = b[n:]
// For known field numbers, reject any non-matching wire type as a
// hard error rather than silently skipping. The caller will catch
// missing-field cases downstream, but a wire-type mismatch on a tag
// we know is a peer protocol violation worth flagging here.
// Repeated occurrences of a singular field follow proto3 last-wins.
switch num {
case fieldMultiportRxSupported:
if typ != protowire.VarintType {
return errInvalidHandshakeDetails
}
v, n := protowire.ConsumeVarint(b)
if n < 0 || v > math.MaxUint32 {
return errInvalidHandshakeDetails
}
p.RxSupported = protowire.DecodeBool(v)
b = b[n:]
case fieldMultiportTxSupported:
if typ != protowire.VarintType {
return errInvalidHandshakeDetails
}
v, n := protowire.ConsumeVarint(b)
if n < 0 || v > math.MaxUint32 {
return errInvalidHandshakeDetails
}
p.TxSupported = protowire.DecodeBool(v)
b = b[n:]
case fieldMultiportBasePort:
if typ != protowire.VarintType {
return errInvalidHandshakeDetails
}
v, n := protowire.ConsumeVarint(b)
if n < 0 || v > math.MaxUint32 {
return errInvalidHandshakeDetails
}
p.BasePort = uint32(v)
b = b[n:]
case fieldMultiportTotalPorts:
if typ != protowire.VarintType {
return errInvalidHandshakeDetails
}
v, n := protowire.ConsumeVarint(b)
if n < 0 || v > math.MaxUint32 {
return errInvalidHandshakeDetails
}
p.TotalPorts = uint32(v)
b = b[n:]
default:
n := protowire.ConsumeFieldValue(num, typ, b)
if n < 0 {
+16 -16
View File
@@ -117,24 +117,24 @@ func TestPayloadUnknownFields(t *testing.T) {
assert.Equal(t, uint32(88), got.ResponderIndex)
})
// t.Run("reserved fields 6 and 7 are skipped", func(t *testing.T) {
// // Fields 6 and 7 are reserved in the proto definition
// var details []byte
// details = protowire.AppendTag(details, fieldInitiatorIndex, protowire.VarintType)
// details = protowire.AppendVarint(details, 100)
// details = protowire.AppendTag(details, 6, protowire.VarintType)
// details = protowire.AppendVarint(details, 1)
// details = protowire.AppendTag(details, 7, protowire.VarintType)
// details = protowire.AppendVarint(details, 2)
t.Run("reserved fields 6 and 7 are skipped", func(t *testing.T) {
// Fields 6 and 7 are reserved in the proto definition
var details []byte
details = protowire.AppendTag(details, fieldInitiatorIndex, protowire.VarintType)
details = protowire.AppendVarint(details, 100)
details = protowire.AppendTag(details, 6, protowire.VarintType)
details = protowire.AppendVarint(details, 1)
details = protowire.AppendTag(details, 7, protowire.VarintType)
details = protowire.AppendVarint(details, 2)
// var data []byte
// data = protowire.AppendTag(data, 1, protowire.BytesType)
// data = protowire.AppendBytes(data, details)
var data []byte
data = protowire.AppendTag(data, 1, protowire.BytesType)
data = protowire.AppendBytes(data, details)
// got, err := UnmarshalPayload(data)
// require.NoError(t, err)
// assert.Equal(t, uint32(100), got.InitiatorIndex)
// })
got, err := UnmarshalPayload(data)
require.NoError(t, err)
assert.Equal(t, uint32(100), got.InitiatorIndex)
})
}
func TestPayloadBytesConsumed(t *testing.T) {
-55
View File
@@ -14,7 +14,6 @@ import (
"github.com/rcrowley/go-metrics"
"github.com/slackhq/nebula/cert"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/handshake"
"github.com/slackhq/nebula/header"
"github.com/slackhq/nebula/udp"
@@ -72,9 +71,6 @@ type HandshakeManager struct {
f *Interface
l *slog.Logger
multiPort config.MultiPortConfig
udpRaw *udp.RawConn
// can be used to trigger outbound handshake for the given vpnIp
trigger chan netip.Addr
}
@@ -295,7 +291,6 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered
// Send the handshake to all known ips, stage 2 takes care of assigning the hostinfo.remote based on the first to reply
var sentTo []netip.AddrPort
var sentMultiport bool
hostinfo.remotes.ForEach(hm.mainHostMap.GetPreferredRanges(), func(addr netip.AddrPort, _ bool) {
hm.messageMetrics.Tx(header.Handshake, hh.machine.Subtype(), 1)
err := hm.outside.WriteTo(stage0, addr)
@@ -316,29 +311,6 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered
} else {
sentTo = append(sentTo, addr)
}
// Attempt a multiport handshake if we are past the TxHandshakeDelay attempts
if hm.multiPort.TxHandshake && hm.udpRaw != nil && hh.counter >= hm.multiPort.TxHandshakeDelay {
sentMultiport = true
// We need to re-allocate with 8 bytes at the start of SOCK_RAW
raw := hostinfo.HandshakePacket[0x80]
if raw == nil {
raw = make([]byte, len(hostinfo.HandshakePacket[0])+udp.RawOverhead)
copy(raw[udp.RawOverhead:], hostinfo.HandshakePacket[0])
hostinfo.HandshakePacket[0x80] = raw
}
hm.messageMetrics.Tx(header.Handshake, header.MessageSubType(hostinfo.HandshakePacket[0][1]), 1)
err = hm.udpRaw.WriteTo(raw, udp.RandomSendPort.UDPSendPort(hm.multiPort.TxPorts), addr)
if err != nil {
hostinfo.logger(hm.l).Error("Failed to send handshake message",
"error", err,
"udpAddr", addr,
"initiatorIndex", hostinfo.localIndexId,
"handshake", hsFields,
)
}
}
})
// Don't be too noisy or confusing if we fail to send a handshake - if we don't get through we'll eventually log a timeout,
@@ -348,7 +320,6 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered
"udpAddrs", sentTo,
"initiatorIndex", hostinfo.localIndexId,
"handshake", hsFields,
"multiportHandshake", sentMultiport,
)
} else if hm.l.Enabled(context.Background(), slog.LevelDebug) {
hostinfo.logger(hm.l).Debug("Handshake message sent",
@@ -701,7 +672,6 @@ func (hm *HandshakeManager) buildStage0Packet(hh *HandshakeHostInfo) bool {
v, cs.GetCredential,
hm.certVerifier(), func() (uint32, error) { return hm.allocateIndex(hh) },
true, header.HandshakeIXPSK0,
hm.multiPort,
)
if err != nil {
hm.f.l.Error("Failed to create handshake machine",
@@ -743,7 +713,6 @@ func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *head
v, cs.GetCredential,
hm.certVerifier(), func() (uint32, error) { return generateIndex(f.l) },
false, header.HandshakeIXPSK0,
hm.multiPort,
)
if err != nil {
f.l.Error("Failed to create handshake machine", "from", via, "error", err)
@@ -768,12 +737,6 @@ func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *head
return
}
if !via.IsRelayed && result.MultiportTx && result.MultiportBasePort != via.UdpAddr.Port() {
// The other side sent us a handshake from a different port, make sure
// we send responses back to the BasePort
via.UdpAddr = netip.AddrPortFrom(via.UdpAddr.Addr(), result.MultiportBasePort)
}
remoteCert := result.RemoteCert
if remoteCert == nil {
f.l.Error("Handshake did not produce a peer certificate", "from", via)
@@ -798,8 +761,6 @@ func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *head
relayForByAddr: map[netip.Addr]*Relay{},
relayForByIdx: map[uint32]*Relay{},
},
multiportTx: hm.multiPort.Tx && result.MultiportRx,
multiportRx: hm.multiPort.Rx && result.MultiportTx,
}
msg := "Handshake message received"
@@ -816,8 +777,6 @@ func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *head
"initiatorIndex", result.RemoteIndex,
"responderIndex", result.LocalIndex,
"handshake", m{"stage": uint64(machine.MessageIndex()), "style": header.SubTypeName(header.Handshake, machine.Subtype())},
"multiportTx", hostinfo.multiportTx,
"multiportRx", hostinfo.multiportRx,
)
// packet aliases the listener's incoming buffer, so this copy must stay.
@@ -908,14 +867,6 @@ func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostIn
return
}
if !via.IsRelayed && result.MultiportTx && result.MultiportBasePort != via.UdpAddr.Port() {
// The other side sent us a handshake from a different port, make sure
// we send responses back to the BasePort
via.UdpAddr = netip.AddrPortFrom(via.UdpAddr.Addr(), result.MultiportBasePort)
}
hostinfo.multiportTx = hm.multiPort.Tx && result.MultiportRx
hostinfo.multiportRx = hm.multiPort.Rx && result.MultiportTx
// Handshake complete; build the ConnectionState now that we have keys and a verified peer cert.
hostinfo.ConnectionState = newConnectionStateFromResult(result)
@@ -1010,8 +961,6 @@ func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostIn
"handshake", m{"stage": uint64(machine.MessageIndex()), "style": header.SubTypeName(header.Handshake, machine.Subtype())},
"durationNs", duration,
"sentCachedPackets", len(hh.packetStore),
"multiportTx", hostinfo.multiportTx,
"multiportRx", hostinfo.multiportRx,
)
hostinfo.vpnAddrs = vpnAddrs
@@ -1153,10 +1102,6 @@ func (hm *HandshakeManager) handleCheckAndCompleteError(err error, existing, hos
switch err {
case ErrAlreadySeen:
if hostinfo.multiportRx {
// The other host is sending to us with multiport, so only grab the IP
via.UdpAddr = netip.AddrPortFrom(via.UdpAddr.Addr(), hostinfo.GetRemote().Port())
}
if existing.SetRemoteIfPreferred(f.hostMap, via) {
f.SendMessageToVpnAddr(header.Test, header.TestRequest, hostinfo.vpnAddrs[0], []byte(""), make([]byte, 12, 12), make([]byte, mtu))
}
-6
View File
@@ -254,12 +254,6 @@ type HostInfo struct {
networks *bart.Table[NetworkType]
relayState RelayState
// If true, we should send to this remote using multiport
multiportTx bool
// If true, we will receive from this remote using multiport
multiportRx bool
// HandshakePacket records the packets used to create this hostinfo
// We need these to avoid replayed handshake packets creating new hostinfos which causes churn
HandshakePacket map[uint8][]byte
+409
View File
@@ -0,0 +1,409 @@
package nebula
import (
"context"
"encoding/json"
"errors"
"fmt"
"io/fs"
"log/slog"
"net"
"net/http"
"net/netip"
"os"
"path/filepath"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/slackhq/nebula/cert"
"github.com/slackhq/nebula/config"
)
// infoAPIServer is a small http+json listener on a unix socket that lets other
// programs on this machine resolve a vpn address to its certificate identity (name, groups, networks)
// for making authorization decisions. Lifecycle works like statsServer: the constructor wires the
// reload callback, reload records config, Start runs the runtime, Stop tears it down
type infoAPIServer struct {
l *slog.Logger
ctx context.Context
hostMap *HostMap
pki *PKI
// enabled mirrors `info_api.enabled` so callers of Start don't need to know the gating rules
enabled atomic.Bool
runMu sync.Mutex
runCfg *infoAPIConfig
run *infoAPIRuntime // non-nil while a runtime is live
}
// infoAPIRuntime is the live state owned by a single Start invocation. Stop and Start's exit path
// use pointer equality to tell "my runtime" apart from one that replaced it after a reload
type infoAPIRuntime struct {
server *http.Server
listener net.Listener
}
// infoAPIConfig is a snapshot of the info_api config section, comparable with == so reload can
// detect "no change" cheaply
type infoAPIConfig struct {
enabled bool
listen string // raw config value, for error messages
addr string // unix socket path
// file mode applied to the unix socket after bind
socketMode fs.FileMode
}
// newInfoAPIServerFromConfig builds a infoAPIServer and applies the initial config. The reload
// callback is registered first so a SIGHUP can later enable, fix, or disable the listener even if
// the initial config was bad. Nothing binds until Start, so config tests are side effect free.
// A bad config is logged rather than returned: it must not stop nebula from starting, the feature
// just stays disabled until a reload provides a valid config
func newInfoAPIServerFromConfig(ctx context.Context, l *slog.Logger, pki *PKI, hostMap *HostMap, c *config.C) *infoAPIServer {
h := &infoAPIServer{
l: l,
ctx: ctx,
hostMap: hostMap,
pki: pki,
}
c.RegisterReloadCallback(func(c *config.C) {
if err := h.reload(c, false); err != nil {
h.l.Warn("Failed to reload info API from config", "error", err)
}
})
if err := h.reload(c, true); err != nil {
h.l.Warn("Failed to apply info API config; it will stay disabled until the config is fixed and reloaded", "error", err)
}
return h
}
// reload records the latest config. The initial call only records it, Control.Start launches the
// first runtime via infoAPIStart. Later calls reconcile the running listener with the new config:
// enable, disable, or restart when the listen config changed
func (h *infoAPIServer) reload(c *config.C, initial bool) error {
newCfg, err := loadInfoAPIConfig(c)
if err != nil {
return err
}
h.runMu.Lock()
sameCfg := h.runCfg != nil && *h.runCfg == newCfg
h.runCfg = &newCfg
running := h.run != nil
h.runMu.Unlock()
h.enabled.Store(newCfg.enabled)
if initial || sameCfg {
return nil
}
if running {
h.Stop()
}
if newCfg.enabled {
go h.Start()
}
return nil
}
// Start binds the listener from the latest config and serves until Stop is called or ctx fires.
// Safe to call when disabled or already running (both no-op)
func (h *infoAPIServer) Start() {
if !h.enabled.Load() {
return
}
h.runMu.Lock()
if h.ctx.Err() != nil || h.run != nil || h.runCfg == nil {
h.runMu.Unlock()
return
}
cfg := *h.runCfg
ln, err := h.listen(cfg)
if err != nil {
// drop the cached config so a SIGHUP with the same config retries the bind
h.runCfg = nil
h.runMu.Unlock()
h.l.Error("Failed to start info API listener", "listen", cfg.listen, "error", err)
return
}
mux := http.NewServeMux()
mux.HandleFunc("GET /v1/host", h.handleHost)
mux.HandleFunc("GET /v1/self", h.handleSelf)
srv := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second}
rt := &infoAPIRuntime{server: srv, listener: ln}
h.run = rt
h.runMu.Unlock()
h.l.Info("Starting info API listener", "addr", ln.Addr())
cleanExit := h.serve(srv, ln)
// A Stop that raced our bind shut the server down before Serve could adopt the listener;
// closing it again is harmless and guarantees a unix socket file gets unlinked
_ = ln.Close()
// Clear our runtime only if nothing has replaced it. Stop races through here too but leaves
// h.run == nil, so the pointer check skips
h.runMu.Lock()
if h.run == rt {
h.run = nil
// an error exit leaves runCfg cached as if it were applied, drop it so a SIGHUP with the
// same config re-triggers Start once the user fixes the underlying problem
if !cleanExit {
h.runCfg = nil
}
}
h.runMu.Unlock()
}
// serve runs srv.Serve and ensures ctx cancellation unblocks it. Returns true if the listener
// exited cleanly (Stop, ctx cancellation), false on an unexpected error
func (h *infoAPIServer) serve(srv *http.Server, ln net.Listener) bool {
// ctx cancellation triggers a server shutdown which in turn unblocks Serve, closing `done` on
// exit keeps the watcher from outliving this call
done := make(chan struct{})
go func() {
select {
case <-h.ctx.Done():
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := srv.Shutdown(shutdownCtx); err != nil {
h.l.Warn("Failed to shut down info API listener", "error", err)
}
case <-done:
}
}()
defer close(done)
err := srv.Serve(ln)
if err == nil || errors.Is(err, http.ErrServerClosed) {
return true
}
h.l.Error("Info API listener exited", "error", err)
return false
}
// Stop tears down the active runtime, if any. Idempotent
func (h *infoAPIServer) Stop() {
h.runMu.Lock()
rt := h.run
h.run = nil
h.runMu.Unlock()
if rt == nil {
return
}
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := rt.server.Shutdown(shutdownCtx); err != nil {
h.l.Warn("Failed to shut down info API listener", "error", err)
}
}
// listen binds the configured unix socket. It also clears a stale socket file left by an unclean
// exit and applies the configured file mode
func (h *infoAPIServer) listen(cfg infoAPIConfig) (net.Listener, error) {
if fi, err := os.Stat(cfg.addr); err == nil {
if fi.Mode()&os.ModeSocket == 0 {
return nil, fmt.Errorf("info_api.listen path %s exists and is not a socket, refusing to replace it", cfg.addr)
}
// a normal shutdown unlinks the socket, so a file here means a previous process exited
// uncleanly, remove it so the bind below can succeed
if err = os.Remove(cfg.addr); err != nil {
return nil, fmt.Errorf("failed to remove stale socket %s: %w", cfg.addr, err)
}
}
ln, err := net.Listen("unix", cfg.addr)
if err != nil {
return nil, err
}
// The socket is briefly live with umask-derived permissions before this chmod lands, tolerated
// because connections accepted in that window still only reach this read-only API
if err = os.Chmod(cfg.addr, cfg.socketMode); err != nil {
_ = ln.Close()
return nil, fmt.Errorf("failed to set mode on socket %s: %w", cfg.addr, err)
}
return ln, nil
}
func (h *infoAPIServer) certState() *CertState {
if h.pki == nil {
return nil
}
return h.pki.getCertState()
}
// handleHost serves GET /v1/host?addr=<vpn addr>, answering with the identity of the host that
// owns the address: a peer with an active tunnel, or this node itself. addr may include a port,
// which is ignored, so clients can pass a connection's remote address through without parsing it
func (h *infoAPIServer) handleHost(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query().Get("addr")
if q == "" {
writeJSONError(w, http.StatusBadRequest, "missing addr parameter")
return
}
ip, err := parseQueryAddrParam(q)
if err != nil {
writeJSONError(w, http.StatusBadRequest, "invalid address")
return
}
crt := findCertificateForVpnAddr(h.certState(), h.hostMap, ip)
if crt == nil {
writeJSONError(w, http.StatusNotFound, "no active tunnel for address")
return
}
h.writeHostIdentity(w, crt)
}
// handleSelf serves GET /v1/self, answering with this node's own identity
func (h *infoAPIServer) handleSelf(w http.ResponseWriter, r *http.Request) {
var crt cert.Certificate
if cs := h.certState(); cs != nil {
crt = cs.getCertificate(cs.initiatingVersion)
}
if crt == nil {
writeJSONError(w, http.StatusInternalServerError, "no certificate available")
return
}
h.writeHostIdentity(w, crt)
}
func (h *infoAPIServer) writeHostIdentity(w http.ResponseWriter, crt cert.Certificate) {
id, err := newHostIdentity(crt)
if err != nil {
writeJSONError(w, http.StatusInternalServerError, "failed to fingerprint certificate")
return
}
w.Header().Set("Content-Type", "application/json")
if err = json.NewEncoder(w).Encode(id); err != nil {
h.l.Debug("Failed to write info API response", "error", err)
}
}
// findCertificateForVpnAddr answers "who owns this vpn address": ourselves (from local cert state,
// the hostmap never carries an entry for this node) or a peer with an active tunnel. Returns nil
// when the address is unknown or the tunnel is mid-teardown
func findCertificateForVpnAddr(cs *CertState, hostMap *HostMap, ip netip.Addr) cert.Certificate {
if cs != nil && cs.myVpnAddrsTable != nil && cs.myVpnAddrsTable.Contains(ip) {
return cs.getCertificate(cs.initiatingVersion)
}
hostinfo := hostMap.QueryVpnAddr(ip)
if hostinfo == nil {
return nil
}
cc := hostinfo.GetCert()
if cc == nil {
return nil
}
return cc.Certificate
}
// hostIdentity is the json document served for both /v1/host and /v1/self, every field is derived
// from the authenticated certificate alone
type hostIdentity struct {
Name string `json:"name"`
VpnAddrs []netip.Addr `json:"vpnAddrs"`
Networks []netip.Prefix `json:"networks"`
UnsafeNetworks []netip.Prefix `json:"unsafeNetworks"`
Groups []string `json:"groups"`
Fingerprint string `json:"fingerprint"`
Issuer string `json:"issuer"`
NotBefore time.Time `json:"notBefore"`
NotAfter time.Time `json:"notAfter"`
CertVersion int `json:"certVersion"`
}
func newHostIdentity(crt cert.Certificate) (hostIdentity, error) {
fp, err := crt.Fingerprint()
if err != nil {
return hostIdentity{}, err
}
// slices are always allocated so they marshal as [] rather than null
networks := crt.Networks()
id := hostIdentity{
Name: crt.Name(),
VpnAddrs: make([]netip.Addr, 0, len(networks)),
Networks: append(make([]netip.Prefix, 0, len(networks)), networks...),
UnsafeNetworks: append(make([]netip.Prefix, 0, len(crt.UnsafeNetworks())), crt.UnsafeNetworks()...),
Groups: append(make([]string, 0, len(crt.Groups())), crt.Groups()...),
Fingerprint: fp,
Issuer: crt.Issuer(),
NotBefore: crt.NotBefore(),
NotAfter: crt.NotAfter(),
CertVersion: int(crt.Version()),
}
for _, n := range networks {
id.VpnAddrs = append(id.VpnAddrs, n.Addr())
}
return id, nil
}
func writeJSONError(w http.ResponseWriter, status int, msg string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(map[string]string{"error": msg})
}
// parseQueryAddrParam parses the addr query parameter, accepting a bare address or an address with
// a port (`192.168.100.7:54321`, `[fd00::1]:443`) so callers can pass a connection's RemoteAddr
// straight through. The result is unmapped, 4in6 addresses (::ffff:a.b.c.d) become ipv4
func parseQueryAddrParam(s string) (netip.Addr, error) {
if ip, err := netip.ParseAddr(s); err == nil {
return ip.Unmap(), nil
}
ap, err := netip.ParseAddrPort(s)
if err != nil {
return netip.Addr{}, err
}
return ap.Addr().Unmap(), nil
}
func loadInfoAPIConfig(c *config.C) (infoAPIConfig, error) {
cfg := infoAPIConfig{
enabled: c.GetBool("info_api.enabled", false),
listen: c.GetString("info_api.listen", ""),
}
if !cfg.enabled {
return cfg, nil
}
if cfg.listen == "" {
return cfg, errors.New("info_api.listen can not be empty when info_api is enabled")
}
addr, err := parseInfoAPIListen(cfg.listen)
if err != nil {
return cfg, err
}
cfg.addr = addr
// read as a string so yaml can't reinterpret the octal literal
modeStr := c.GetString("info_api.socket_mode", "0600")
mode, err := strconv.ParseUint(modeStr, 8, 32)
if err != nil || fs.FileMode(mode)&^fs.ModePerm != 0 {
return cfg, fmt.Errorf("info_api.socket_mode was not a valid octal file mode: %s", modeStr)
}
cfg.socketMode = fs.FileMode(mode)
return cfg, nil
}
// parseInfoAPIListen extracts the unix socket path from the info_api.listen config value, which
// must be a `unix://` URL with an absolute path, e.g. `unix:///var/run/nebula.sock`
func parseInfoAPIListen(listen string) (addr string, err error) {
path, ok := strings.CutPrefix(listen, "unix://")
if !ok {
return "", fmt.Errorf("info_api.listen must be a unix:// socket path: %s", listen)
} else if !filepath.IsAbs(path) {
return "", fmt.Errorf("info_api.listen unix socket path must be absolute: %s", listen)
}
return path, nil
}
+448
View File
@@ -0,0 +1,448 @@
package nebula
import (
"context"
"encoding/json"
"fmt"
"io/fs"
"log/slog"
"net"
"net/http"
"net/http/httptest"
"net/netip"
"net/url"
"os"
"path/filepath"
"runtime"
"testing"
"time"
"github.com/slackhq/nebula/cert"
"github.com/slackhq/nebula/cert_test"
"github.com/slackhq/nebula/config"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func Test_parseInfoAPIListen(t *testing.T) {
type testCase struct {
listen string
addr string
wantErr bool
}
tests := []testCase{
{listen: "", wantErr: true},
{listen: "unix://", wantErr: true},
{listen: "unix://relative/path.sock", wantErr: true},
{listen: "not an address", wantErr: true},
// tcp host:port addresses are no longer accepted
{listen: "127.0.0.1:8085", wantErr: true},
{listen: "[::1]:8085", wantErr: true},
{listen: "localhost:8085", wantErr: true},
}
// A unix socket path must be absolute for the OS that will bind it, and filepath.IsAbs is
// GOOS-specific. CI runs the suite separately on each OS, so assert the platform's own native
// absolute path is accepted while the other platform's is rejected.
posixPath := "unix:///var/run/nebula.sock"
winPath := `unix://C:\nebula\hq.sock`
if runtime.GOOS == "windows" {
tests = append(tests,
testCase{listen: winPath, addr: `C:\nebula\hq.sock`},
testCase{listen: posixPath, wantErr: true},
)
} else {
tests = append(tests,
testCase{listen: posixPath, addr: "/var/run/nebula.sock"},
testCase{listen: winPath, wantErr: true},
)
}
for _, tt := range tests {
addr, err := parseInfoAPIListen(tt.listen)
if tt.wantErr {
require.Error(t, err, "listen=%q", tt.listen)
continue
}
require.NoError(t, err, "listen=%q", tt.listen)
assert.Equal(t, tt.addr, addr, "listen=%q", tt.listen)
}
}
func Test_loadInfoAPIConfig(t *testing.T) {
c := config.NewC(nil)
// the listen path must be absolute for the OS running the test (CI is per-OS)
listen, wantAddr := "unix:///tmp/hq.sock", "/tmp/hq.sock"
if runtime.GOOS == "windows" {
listen, wantAddr = `unix://C:\tmp\hq.sock`, `C:\tmp\hq.sock`
}
// absent section means disabled, no error
cfg, err := loadInfoAPIConfig(c)
require.NoError(t, err)
assert.False(t, cfg.enabled)
// enabled without a listen address is an error
setInfoAPIConfig(c, true, "", "")
_, err = loadInfoAPIConfig(c)
require.Error(t, err)
// a unix socket gets the default mode
setInfoAPIConfig(c, true, listen, "")
cfg, err = loadInfoAPIConfig(c)
require.NoError(t, err)
assert.Equal(t, wantAddr, cfg.addr)
assert.Equal(t, fs.FileMode(0o600), cfg.socketMode)
setInfoAPIConfig(c, true, listen, "0660")
cfg, err = loadInfoAPIConfig(c)
require.NoError(t, err)
assert.Equal(t, fs.FileMode(0o660), cfg.socketMode)
setInfoAPIConfig(c, true, listen, "withers")
_, err = loadInfoAPIConfig(c)
require.Error(t, err)
// mode bits beyond the permission bits are rejected
setInfoAPIConfig(c, true, listen, "10600")
_, err = loadInfoAPIConfig(c)
require.Error(t, err)
// tcp host:port listen addresses are no longer supported
setInfoAPIConfig(c, true, "127.0.0.1:8085", "")
_, err = loadInfoAPIConfig(c)
require.Error(t, err)
}
func TestInfoAPIServer_badConfigIsNonFatal(t *testing.T) {
// an enabled-but-invalid config must not stop construction; nebula keeps starting and the
// feature simply stays disabled until a reload supplies a valid config
c := config.NewC(nil)
setInfoAPIConfig(c, true, "not-a-unix-socket", "")
h := newInfoAPIServerFromConfig(context.Background(), slog.New(slog.DiscardHandler), nil, newHostMap(slog.New(slog.DiscardHandler)), c)
require.NotNil(t, h)
assert.False(t, h.enabled.Load())
// no config was recorded, so Start has nothing to bind and is a no-op
h.runMu.Lock()
assert.Nil(t, h.runCfg)
h.runMu.Unlock()
h.Start()
h.runMu.Lock()
assert.Nil(t, h.run)
h.runMu.Unlock()
}
func setInfoAPIConfig(c *config.C, enabled bool, listen, socketMode string) {
settings := map[string]any{
"enabled": enabled,
"listen": listen,
}
if socketMode != "" {
settings["socket_mode"] = socketMode
}
c.Settings["info_api"] = settings
}
func newTestInfoAPIServer(t *testing.T) (*infoAPIServer, *config.C) {
t.Helper()
h := &infoAPIServer{
l: slog.New(slog.DiscardHandler),
ctx: context.Background(),
hostMap: newHostMap(slog.New(slog.DiscardHandler)),
}
h.hostMap.preferredRanges.Store(&[]netip.Prefix{})
return h, config.NewC(nil)
}
// addTestPeer creates a certificate for a peer owning each addr (as a /24 or /64) and inserts it
// into the hostmap as an established tunnel
func addTestPeer(t *testing.T, hm *HostMap, name string, addrs []netip.Addr, unsafeNetworks []netip.Prefix, groups []string) cert.Certificate {
t.Helper()
networks := make([]netip.Prefix, 0, len(addrs))
for _, a := range addrs {
bits := 24
if a.Is6() {
bits = 64
}
networks = append(networks, netip.PrefixFrom(a, bits))
}
ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version2, cert.Curve_CURVE25519, time.Time{}, time.Time{}, nil, nil, nil)
crt, _, _, _ := cert_test.NewTestCert(cert.Version2, cert.Curve_CURVE25519, ca, caKey, name, time.Time{}, time.Time{}, networks, unsafeNetworks, groups)
fp, err := crt.Fingerprint()
require.NoError(t, err)
hm.unlockedAddHostInfo(&HostInfo{
ConnectionState: &ConnectionState{
peerCert: &cert.CachedCertificate{Certificate: crt, Fingerprint: fp},
},
vpnAddrs: addrs,
relayState: RelayState{
relayForByAddr: map[netip.Addr]*Relay{},
relayForByIdx: map[uint32]*Relay{},
},
}, &Interface{})
return crt
}
func getHost(t *testing.T, h *infoAPIServer, addrParam string) (int, map[string]any) {
t.Helper()
r := httptest.NewRequest(http.MethodGet, "/v1/host?addr="+url.QueryEscape(addrParam), nil)
w := httptest.NewRecorder()
h.handleHost(w, r)
return decodeResponse(t, w)
}
func decodeResponse(t *testing.T, w *httptest.ResponseRecorder) (int, map[string]any) {
t.Helper()
assert.Equal(t, "application/json", w.Header().Get("Content-Type"))
var body map[string]any
require.NoError(t, json.Unmarshal(w.Body.Bytes(), &body))
return w.Code, body
}
func TestInfoAPIServer_handleHost(t *testing.T) {
h, _ := newTestInfoAPIServer(t)
h.pki = newTestPKI(t, "self", []netip.Addr{netip.MustParseAddr("10.0.0.1")})
peerV4 := netip.MustParseAddr("10.0.0.99")
peerV6 := netip.MustParseAddr("fd00::99")
addTestPeer(t, h.hostMap, "laptop-alice", []netip.Addr{peerV4, peerV6},
[]netip.Prefix{netip.MustParsePrefix("192.168.50.0/24")}, []string{"eng", "ssh"})
addTestPeer(t, h.hostMap, "groupless", []netip.Addr{netip.MustParseAddr("10.0.0.77")}, nil, nil)
// an established peer comes back with its full identity
code, body := getHost(t, h, "10.0.0.99")
require.Equal(t, http.StatusOK, code)
assert.Equal(t, "laptop-alice", body["name"])
assert.Equal(t, []any{"10.0.0.99", "fd00::99"}, body["vpnAddrs"])
assert.Equal(t, []any{"10.0.0.99/24", "fd00::99/64"}, body["networks"])
assert.Equal(t, []any{"192.168.50.0/24"}, body["unsafeNetworks"])
assert.Equal(t, []any{"eng", "ssh"}, body["groups"])
assert.NotEmpty(t, body["fingerprint"])
assert.Equal(t, "2", fmt.Sprintf("%v", body["certVersion"]))
assert.NotEmpty(t, body["notBefore"])
assert.NotEmpty(t, body["notAfter"])
// empty cert slices marshal as [] rather than null
code, body = getHost(t, h, "10.0.0.77")
require.Equal(t, http.StatusOK, code)
require.NotNil(t, body["groups"])
assert.Empty(t, body["groups"])
require.NotNil(t, body["unsafeNetworks"])
assert.Empty(t, body["unsafeNetworks"])
// a port in addr is ignored so RemoteAddr can be passed through directly, including the
// bracketed v6 and 4in6 forms
for _, q := range []string{"10.0.0.99:54321", "[fd00::99]:443", "::ffff:10.0.0.99"} {
code, body = getHost(t, h, q)
require.Equal(t, http.StatusOK, code, "addr=%q", q)
assert.Equal(t, "laptop-alice", body["name"], "addr=%q", q)
}
// our own address answers from the local cert state
code, body = getHost(t, h, "10.0.0.1")
require.Equal(t, http.StatusOK, code)
assert.Equal(t, "self", body["name"])
code, body = getHost(t, h, "10.0.0.42")
assert.Equal(t, http.StatusNotFound, code)
assert.NotEmpty(t, body["error"])
// a tunnel mid-teardown (no peer cert) is treated as unknown
h.hostMap.unlockedAddHostInfo(&HostInfo{
ConnectionState: &ConnectionState{},
vpnAddrs: []netip.Addr{netip.MustParseAddr("10.0.0.66")},
relayState: RelayState{
relayForByAddr: map[netip.Addr]*Relay{},
relayForByIdx: map[uint32]*Relay{},
},
}, &Interface{})
code, _ = getHost(t, h, "10.0.0.66")
assert.Equal(t, http.StatusNotFound, code)
code, body = getHost(t, h, "not-an-address")
assert.Equal(t, http.StatusBadRequest, code)
assert.NotEmpty(t, body["error"])
r := httptest.NewRequest(http.MethodGet, "/v1/host", nil)
w := httptest.NewRecorder()
h.handleHost(w, r)
code, body = decodeResponse(t, w)
assert.Equal(t, http.StatusBadRequest, code)
assert.NotEmpty(t, body["error"])
}
func TestInfoAPIServer_handleSelf(t *testing.T) {
h, _ := newTestInfoAPIServer(t)
h.pki = newTestPKI(t, "lighthouse", []netip.Addr{netip.MustParseAddr("10.0.0.1")})
r := httptest.NewRequest(http.MethodGet, "/v1/self", nil)
w := httptest.NewRecorder()
h.handleSelf(w, r)
code, body := decodeResponse(t, w)
require.Equal(t, http.StatusOK, code)
assert.Equal(t, "lighthouse", body["name"])
assert.Equal(t, []any{"10.0.0.1"}, body["vpnAddrs"])
// no cert state available should be an error, not a panic
h.pki = nil
w = httptest.NewRecorder()
h.handleSelf(w, r)
code, body = decodeResponse(t, w)
assert.Equal(t, http.StatusInternalServerError, code)
assert.NotEmpty(t, body["error"])
}
func unixHTTPClient(path string) *http.Client {
return &http.Client{
Timeout: time.Second,
Transport: &http.Transport{
DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
return (&net.Dialer{}).DialContext(ctx, "unix", path)
},
},
}
}
// waitForServe polls until a GET /v1/self through client succeeds
func waitForServe(t *testing.T, client *http.Client) {
t.Helper()
waitFor(t, func() bool {
resp, err := client.Get("http://hostquery/v1/self")
if err != nil {
return false
}
resp.Body.Close()
return resp.StatusCode == http.StatusOK
})
}
func skipIfNoUnixSockets(t *testing.T) {
t.Helper()
if runtime.GOOS == "windows" {
t.Skip("unix socket tests are not supported on windows CI")
}
}
func TestInfoAPIServer_unixLifecycle(t *testing.T) {
skipIfNoUnixSockets(t)
h, c := newTestInfoAPIServer(t)
h.pki = newTestPKI(t, "self", []netip.Addr{netip.MustParseAddr("10.0.0.1")})
sock := filepath.Join(t.TempDir(), "hq.sock")
setInfoAPIConfig(c, true, "unix://"+sock, "")
require.NoError(t, h.reload(c, true))
done := make(chan struct{})
go func() {
h.Start()
close(done)
}()
client := unixHTTPClient(sock)
waitForServe(t, client)
fi, err := os.Stat(sock)
require.NoError(t, err)
assert.Equal(t, fs.FileMode(0o600), fi.Mode().Perm())
resp, err := client.Get("http://hostquery/v1/host?addr=10.0.0.1")
require.NoError(t, err)
resp.Body.Close()
assert.Equal(t, http.StatusOK, resp.StatusCode)
h.Stop()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("Start did not return after Stop")
}
_, err = os.Stat(sock)
assert.True(t, os.IsNotExist(err), "socket file should be unlinked on shutdown")
}
func TestInfoAPIServer_staleSocket(t *testing.T) {
skipIfNoUnixSockets(t)
h, _ := newTestInfoAPIServer(t)
sock := filepath.Join(t.TempDir(), "hq.sock")
// simulate an unclean exit, a leftover socket file with no listener
stale, err := net.ListenUnix("unix", &net.UnixAddr{Name: sock, Net: "unix"})
require.NoError(t, err)
stale.SetUnlinkOnClose(false)
require.NoError(t, stale.Close())
_, err = os.Stat(sock)
require.NoError(t, err, "stale socket file should exist")
cfg := infoAPIConfig{addr: sock, socketMode: 0o600}
ln, err := h.listen(cfg)
require.NoError(t, err, "a stale socket should be removed and rebound")
require.NoError(t, ln.Close())
}
func TestInfoAPIServer_existingFileNotReplaced(t *testing.T) {
skipIfNoUnixSockets(t)
h, _ := newTestInfoAPIServer(t)
path := filepath.Join(t.TempDir(), "hq.sock")
require.NoError(t, os.WriteFile(path, []byte("precious"), 0o600))
cfg := infoAPIConfig{addr: path, socketMode: 0o600}
_, err := h.listen(cfg)
require.Error(t, err, "a non-socket file at the listen path must not be replaced")
content, err := os.ReadFile(path)
require.NoError(t, err)
assert.Equal(t, "precious", string(content))
}
func TestInfoAPIServer_reload(t *testing.T) {
skipIfNoUnixSockets(t)
h, c := newTestInfoAPIServer(t)
h.pki = newTestPKI(t, "self", []netip.Addr{netip.MustParseAddr("10.0.0.1")})
dir := t.TempDir()
sock1 := filepath.Join(dir, "hq1.sock")
sock2 := filepath.Join(dir, "hq2.sock")
// initial reload only records config, Control.Start is what launches the runtime
setInfoAPIConfig(c, false, "unix://"+sock1, "")
require.NoError(t, h.reload(c, true))
assert.False(t, h.enabled.Load())
h.runMu.Lock()
assert.Nil(t, h.run)
h.runMu.Unlock()
// enabling via reload spawns the listener
setInfoAPIConfig(c, true, "unix://"+sock1, "")
require.NoError(t, h.reload(c, false))
waitForServe(t, unixHTTPClient(sock1))
// changing the listen path restarts on the new address
setInfoAPIConfig(c, true, "unix://"+sock2, "")
require.NoError(t, h.reload(c, false))
waitForServe(t, unixHTTPClient(sock2))
waitFor(t, func() bool {
_, err := os.Stat(sock1)
return os.IsNotExist(err)
})
// reloading an unchanged config does not restart the runtime
h.runMu.Lock()
rt := h.run
h.runMu.Unlock()
require.NoError(t, h.reload(c, false))
h.runMu.Lock()
assert.Same(t, rt, h.run)
h.runMu.Unlock()
// disabling stops the listener
setInfoAPIConfig(c, false, "unix://"+sock2, "")
require.NoError(t, h.reload(c, false))
assert.False(t, h.enabled.Load())
waitFor(t, func() bool {
h.runMu.Lock()
defer h.runMu.Unlock()
return h.run == nil
})
}
+8 -38
View File
@@ -10,7 +10,6 @@ import (
"github.com/slackhq/nebula/iputil"
"github.com/slackhq/nebula/noiseutil"
"github.com/slackhq/nebula/routing"
"github.com/slackhq/nebula/udp"
)
func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet, nb, out []byte, q int, localCache firewall.ConntrackCache) {
@@ -74,7 +73,7 @@ func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet
dropReason := f.firewall.Drop(*fwPacket, false, hostinfo, f.pki.GetCAPool(), localCache)
if dropReason == nil {
f.sendNoMetrics(header.Message, 0, hostinfo.ConnectionState, hostinfo, netip.AddrPort{}, packet, nb, out, q, fwPacket)
f.sendNoMetrics(header.Message, 0, hostinfo.ConnectionState, hostinfo, netip.AddrPort{}, packet, nb, out, q)
} else {
f.rejectInside(packet, out, q)
@@ -123,7 +122,7 @@ func (f *Interface) rejectOutside(packet []byte, ci *ConnectionState, hostinfo *
return
}
f.sendNoMetrics(header.Message, 0, ci, hostinfo, netip.AddrPort{}, out, nb, packet, q, nil)
f.sendNoMetrics(header.Message, 0, ci, hostinfo, netip.AddrPort{}, out, nb, packet, q)
}
// Handshake will attempt to initiate a tunnel with the provided vpn address. This is a no-op if the tunnel is already established or being established
@@ -236,7 +235,7 @@ func (f *Interface) sendMessageNow(t header.MessageType, st header.MessageSubTyp
return
}
f.sendNoMetrics(header.Message, st, hostinfo.ConnectionState, hostinfo, netip.AddrPort{}, p, nb, out, 0, nil)
f.sendNoMetrics(header.Message, st, hostinfo.ConnectionState, hostinfo, netip.AddrPort{}, p, nb, out, 0)
}
// SendMessageToVpnAddr handles real addr:port lookup and sends to the current best known address for vpnAddr.
@@ -268,12 +267,12 @@ func (f *Interface) SendMessageToHostInfo(t header.MessageType, st header.Messag
func (f *Interface) send(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, p, nb, out []byte) {
f.messageMetrics.Tx(t, st, 1)
f.sendNoMetrics(t, st, ci, hostinfo, netip.AddrPort{}, p, nb, out, 0, nil)
f.sendNoMetrics(t, st, ci, hostinfo, netip.AddrPort{}, p, nb, out, 0)
}
func (f *Interface) sendTo(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote netip.AddrPort, p, nb, out []byte) {
f.messageMetrics.Tx(t, st, 1)
f.sendNoMetrics(t, st, ci, hostinfo, remote, p, nb, out, 0, nil)
f.sendNoMetrics(t, st, ci, hostinfo, remote, p, nb, out, 0)
}
// SendVia sends a payload through a Relay tunnel. No authentication or encryption is done
@@ -341,27 +340,10 @@ func (f *Interface) SendVia(via *HostInfo,
f.connectionManager.RelayUsed(relay.LocalIndex)
}
func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote netip.AddrPort, p, nb, out []byte, q int, udpPortGetter udp.SendPortGetter) {
func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote netip.AddrPort, p, nb, out []byte, q int) {
if ci.eKey == nil {
return
}
multiport := f.multiPort.Tx && hostinfo.multiportTx
rawOut := out
if multiport {
if len(out) < udp.RawOverhead {
// NOTE: This is because some spots in the code send us `out[:0]`, so
// we need to expand the slice back out to get our 8 bytes back.
out = out[:udp.RawOverhead]
}
// Preserve bytes needed for the raw socket
out = out[udp.RawOverhead:]
if udpPortGetter == nil {
udpPortGetter = udp.RandomSendPort
}
}
useRelay := !remote.IsValid() && !hostinfo.GetRemote().IsValid()
fullOut := out
@@ -414,13 +396,7 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType
}
if remote.IsValid() {
if multiport {
rawOut = rawOut[:len(out)+udp.RawOverhead]
port := udpPortGetter.UDPSendPort(f.multiPort.TxPorts)
err = f.udpRaw.WriteTo(rawOut, port, remote)
} else {
err = f.writers[q].WriteTo(out, remote)
}
err = f.writers[q].WriteTo(out, remote)
if err != nil {
hostinfo.logger(f.l).Error("Failed to write outgoing packet",
"error", err,
@@ -428,13 +404,7 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType
)
}
} else if hr := hostinfo.GetRemote(); hr.IsValid() {
if multiport {
rawOut = rawOut[:len(out)+udp.RawOverhead]
port := udpPortGetter.UDPSendPort(f.multiPort.TxPorts)
err = f.udpRaw.WriteTo(rawOut, port, hr)
} else {
err = f.writers[q].WriteTo(out, hr)
}
err = f.writers[q].WriteTo(out, hr)
if err != nil {
hostinfo.logger(f.l).Error("Failed to write outgoing packet",
"error", err,
-23
View File
@@ -99,9 +99,6 @@ type Interface struct {
// triggerShutdown is a function that will be run exactly once, when onFatal swaps something non-nil into fatalErr
triggerShutdown func()
udpRaw *udp.RawConn
multiPort config.MultiPortConfig
metricHandshakes metrics.Histogram
messageMetrics *MessageMetrics
cachedPacketMetrics *cachedPacketMetrics
@@ -109,15 +106,6 @@ type Interface struct {
l *slog.Logger
}
type MultiPortConfig struct {
Tx bool
Rx bool
TxBasePort uint16
TxPorts int
TxHandshake bool
TxHandshakeDelay int64
}
type EncWriter interface {
SendVia(via *HostInfo,
relay *Relay,
@@ -261,8 +249,6 @@ func (f *Interface) activate() error {
metrics.GetOrRegisterGauge("routines", nil).Update(int64(f.routines))
metrics.GetOrRegisterGauge("multiport.tx_ports", nil).Update(int64(f.multiPort.TxPorts))
// Prepare n tun queues
var reader io.ReadWriteCloser = f.inside
for i := 0; i < f.routines; i++ {
@@ -519,8 +505,6 @@ func (f *Interface) emitStats(ctx context.Context, i time.Duration) {
udpStats := udp.NewUDPStatsEmitter(f.writers)
var rawStats func()
certExpirationGauge := metrics.GetOrRegisterGauge("certificate.ttl_seconds", nil)
certInitiatingVersion := metrics.GetOrRegisterGauge("certificate.initiating_version", nil)
certMaxVersion := metrics.GetOrRegisterGauge("certificate.max_version", nil)
@@ -535,13 +519,6 @@ func (f *Interface) emitStats(ctx context.Context, i time.Duration) {
certExpirationGauge.Update(int64(defaultCrt.NotAfter().Sub(time.Now()) / time.Second))
certInitiatingVersion.Update(int64(defaultCrt.Version()))
if f.udpRaw != nil {
if rawStats == nil {
rawStats = udp.NewRawStatsEmitter(f.udpRaw)
}
rawStats()
}
// Report the max certificate version we are capable of using
if certState.v2Cert != nil {
certMaxVersion.Update(int64(certState.v2Cert.Version()))
+3 -33
View File
@@ -244,39 +244,6 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
ifce.writers = udpConns
lightHouse.ifce = ifce
loadMultiPortConfig := func(c *config.C) {
ifce.multiPort.Rx = c.GetBool("tun.multiport.rx_enabled", false)
tx := c.GetBool("tun.multiport.tx_enabled", false)
if tx && ifce.udpRaw == nil {
ifce.udpRaw, err = udp.NewRawConn(l, c.GetString("listen.host", "0.0.0.0"), port, uint16(port))
if err != nil {
l.Error("Failed to get raw socket for tun.multiport.tx_enabled", "error", err)
ifce.udpRaw = nil
tx = false
}
}
if tx {
ifce.multiPort.TxBasePort = uint16(port)
ifce.multiPort.TxPorts = c.GetInt("tun.multiport.tx_ports", 100)
ifce.multiPort.TxHandshake = c.GetBool("tun.multiport.tx_handshake", false)
ifce.multiPort.TxHandshakeDelay = int64(c.GetInt("tun.multiport.tx_handshake_delay", 2))
ifce.udpRaw.ReloadConfig(c)
}
ifce.multiPort.Tx = tx
// TODO: if we upstream this, make this cleaner
handshakeManager.udpRaw = ifce.udpRaw
handshakeManager.multiPort = ifce.multiPort
l.Info("Multiport configured", "multiPort", ifce.multiPort)
}
loadMultiPortConfig(c)
c.RegisterReloadCallback(loadMultiPortConfig)
ifce.RegisterConfigChangeCallbacks(c)
ifce.reloadDisconnectInvalid(c)
ifce.reloadSendRecvError(c)
@@ -293,6 +260,8 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
return nil, util.ContextualizeIfNeeded("Failed to start stats emitter", err)
}
infoAPI := newInfoAPIServerFromConfig(ctx, l, pki, hostMap, c)
if configTest {
return nil, nil
}
@@ -312,6 +281,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
sshStart: sshStart,
statsStart: stats.Start,
dnsStart: ds.Start,
infoAPIStart: infoAPI.Start,
lighthouseStart: lightHouse.StartUpdateWorker,
networkChangeStart: networkChanges.Start,
connectionManagerStart: connManager.Start,
-9
View File
@@ -264,15 +264,6 @@ func (f *Interface) sendCloseTunnel(h *HostInfo) {
func (f *Interface) handleHostRoaming(hostinfo *HostInfo, via ViaSender) {
curRemote := hostinfo.GetRemote()
if !via.IsRelayed && curRemote != via.UdpAddr {
if hostinfo.multiportRx {
// If the remote is sending with multiport, we aren't roaming unless
// the IP has changed
if curRemote.Addr().Compare(via.UdpAddr.Addr()) == 0 {
return
}
// Keep the port from the original hostinfo, because the remote is transmitting from multiport ports
via.UdpAddr = netip.AddrPortFrom(via.UdpAddr.Addr(), curRemote.Port())
}
if !f.lightHouse.GetRemoteAllowList().AllowAll(hostinfo.vpnAddrs, via.UdpAddr.Addr()) {
if f.l.Enabled(context.Background(), slog.LevelDebug) {
hostinfo.logger(f.l).Debug("lighthouse.remote_allow_list denied roaming", "newAddr", via.UdpAddr)
-16
View File
@@ -1,16 +0,0 @@
package udp
import mathrand "math/rand"
type SendPortGetter interface {
// UDPSendPort returns the port to use
UDPSendPort(maxPort int) uint16
}
type randomSendPort struct{}
func (randomSendPort) UDPSendPort(maxPort int) uint16 {
return uint16(mathrand.Intn(maxPort))
}
var RandomSendPort = randomSendPort{}
-191
View File
@@ -1,191 +0,0 @@
//go:build !android && !e2e_testing
// +build !android,!e2e_testing
package udp
import (
"encoding/binary"
"fmt"
"log/slog"
"net"
"net/netip"
"syscall"
"unsafe"
"github.com/rcrowley/go-metrics"
"github.com/slackhq/nebula/config"
"golang.org/x/net/ipv4"
"golang.org/x/sys/unix"
)
// RawOverhead is the number of bytes that need to be reserved at the start of
// the raw bytes passed to (*RawConn).WriteTo. This is used by WriteTo to prefix
// the IP and UDP headers.
const RawOverhead = 28
type RawConn struct {
sysFd int
basePort uint16
l *slog.Logger
}
func NewRawConn(l *slog.Logger, ip string, port int, basePort uint16) (*RawConn, error) {
syscall.ForkLock.RLock()
// With IPPROTO_UDP, the linux kernel tries to deliver every UDP packet
// received in the system to our socket. This constantly overflows our
// buffer and marks our socket as having dropped packets. This makes the
// stats on the socket useless.
//
// In contrast, IPPROTO_RAW is not delivered any packets and thus our read
// buffer will not fill up and mark as having dropped packets. The only
// difference is that we have to assemble the IP header as well, but this
// is fairly easy since Linux does the checksum for us.
//
// TODO: How to get this working with Inet6 correctly? I was having issues
// with the source address when testing before, probably need to `bind(2)`?
fd, err := unix.Socket(unix.AF_INET, unix.SOCK_RAW, unix.IPPROTO_RAW)
if err == nil {
unix.CloseOnExec(fd)
}
syscall.ForkLock.RUnlock()
if err != nil {
return nil, err
}
// We only want to send, not recv. This will hopefully help the kernel avoid
// wasting time on us
if err = unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_RCVBUF, 0); err != nil {
return nil, fmt.Errorf("unable to set SO_RCVBUF: %s", err)
}
var lip [16]byte
copy(lip[:], net.ParseIP(ip))
// TODO do we need to `bind(2)` so that we send from the correct address/interface?
if err = unix.Bind(fd, &unix.SockaddrInet6{Addr: lip, Port: port}); err != nil {
return nil, fmt.Errorf("unable to bind to socket: %s", err)
}
return &RawConn{
sysFd: fd,
basePort: basePort,
l: l,
}, nil
}
// WriteTo must be called with raw leaving the first `udp.RawOverhead` bytes empty,
// for the IP/UDP headers.
func (u *RawConn) WriteTo(raw []byte, fromPort uint16, ip netip.AddrPort) error {
var rsa unix.RawSockaddrInet4
rsa.Family = unix.AF_INET
rsa.Addr = ip.Addr().As4()
totalLen := len(raw)
udpLen := totalLen - ipv4.HeaderLen
// IP header
raw[0] = byte(ipv4.Version<<4 | (ipv4.HeaderLen >> 2 & 0x0f))
raw[1] = 0 // tos
binary.BigEndian.PutUint16(raw[2:4], uint16(totalLen))
binary.BigEndian.PutUint16(raw[4:6], 0) // id (linux does it for us)
binary.BigEndian.PutUint16(raw[6:8], 0) // frag options
raw[8] = byte(64) // ttl
raw[9] = byte(17) // protocol
binary.BigEndian.PutUint16(raw[10:12], 0) // checksum (linux does it for us)
binary.BigEndian.PutUint32(raw[12:16], 0) // src (linux does it for us)
copy(raw[16:20], rsa.Addr[:]) // dst
// UDP header
fromPort = u.basePort + fromPort
binary.BigEndian.PutUint16(raw[20:22], uint16(fromPort)) // src port
binary.BigEndian.PutUint16(raw[22:24], uint16(ip.Port())) // dst port
binary.BigEndian.PutUint16(raw[24:26], uint16(udpLen)) // UDP length
binary.BigEndian.PutUint16(raw[26:28], 0) // checksum (optional)
for {
_, _, err := unix.Syscall6(
unix.SYS_SENDTO,
uintptr(u.sysFd),
uintptr(unsafe.Pointer(&raw[0])),
uintptr(len(raw)),
uintptr(0),
uintptr(unsafe.Pointer(&rsa)),
uintptr(unix.SizeofSockaddrInet4),
)
if err != 0 {
return &net.OpError{Op: "sendto", Err: err}
}
//TODO: handle incomplete writes
return nil
}
}
func (u *RawConn) ReloadConfig(c *config.C) {
b := c.GetInt("listen.write_buffer", 0)
if b <= 0 {
return
}
if err := u.SetSendBuffer(b); err != nil {
u.l.Error("Failed to set listen.write_buffer", "error", err)
return
}
s, err := u.GetSendBuffer()
if err != nil {
u.l.Warn("Failed to get listen.write_buffer", "error", err)
return
}
u.l.Info("listen.write_buffer was set", "size", s)
}
func (u *RawConn) SetSendBuffer(n int) error {
return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_SNDBUFFORCE, n)
}
func (u *RawConn) GetSendBuffer() (int, error) {
return unix.GetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_SNDBUF)
}
func (u *RawConn) getMemInfo(meminfo *[unix.SK_MEMINFO_VARS]uint32) error {
var vallen uint32 = 4 * unix.SK_MEMINFO_VARS
_, _, err := unix.Syscall6(unix.SYS_GETSOCKOPT, uintptr(u.sysFd), uintptr(unix.SOL_SOCKET), uintptr(unix.SO_MEMINFO), uintptr(unsafe.Pointer(meminfo)), uintptr(unsafe.Pointer(&vallen)), 0)
if err != 0 {
return err
}
return nil
}
func NewRawStatsEmitter(rawConn *RawConn) func() {
// Check if our kernel supports SO_MEMINFO before registering the gauges
var gauges [unix.SK_MEMINFO_VARS]metrics.Gauge
var meminfo [unix.SK_MEMINFO_VARS]uint32
if err := rawConn.getMemInfo(&meminfo); err == nil {
gauges = [unix.SK_MEMINFO_VARS]metrics.Gauge{
metrics.GetOrRegisterGauge("raw.rmem_alloc", nil),
metrics.GetOrRegisterGauge("raw.rcvbuf", nil),
metrics.GetOrRegisterGauge("raw.wmem_alloc", nil),
metrics.GetOrRegisterGauge("raw.sndbuf", nil),
metrics.GetOrRegisterGauge("raw.fwd_alloc", nil),
metrics.GetOrRegisterGauge("raw.wmem_queued", nil),
metrics.GetOrRegisterGauge("raw.optmem", nil),
metrics.GetOrRegisterGauge("raw.backlog", nil),
metrics.GetOrRegisterGauge("raw.drops", nil),
}
} else {
// return no-op because we don't support SO_MEMINFO
return func() {}
}
return func() {
if err := rawConn.getMemInfo(&meminfo); err == nil {
for j := 0; j < unix.SK_MEMINFO_VARS; j++ {
gauges[j].Update(int64(meminfo[j]))
}
}
}
}
-29
View File
@@ -1,29 +0,0 @@
//go:build !linux || android || e2e_testing
// +build !linux android e2e_testing
package udp
import (
"fmt"
"log/slog"
"net/netip"
"runtime"
"github.com/slackhq/nebula/config"
)
const RawOverhead = 0
type RawConn struct{}
func NewRawConn(l *slog.Logger, ip string, port int, basePort uint16) (*RawConn, error) {
return nil, fmt.Errorf("multiport tx is not supported on %s", runtime.GOOS)
}
func (u *RawConn) WriteTo(raw []byte, fromPort uint16, addr netip.AddrPort) error {
return fmt.Errorf("multiport tx is not supported on %s", runtime.GOOS)
}
func (u *RawConn) ReloadConfig(c *config.C) {}
func NewRawStatsEmitter(rawConn *RawConn) func() { return func() {} }