mirror of
https://github.com/netbirdio/netbird.git
synced 2026-09-23 07:09:08 +02:00
Merge branch 'main' into fix/ios-debug-bundle
This commit is contained in:
@@ -11,7 +11,7 @@ import (
|
||||
"go.opentelemetry.io/otel"
|
||||
"google.golang.org/grpc"
|
||||
|
||||
"github.com/netbirdio/management-integrations/integrations"
|
||||
"github.com/netbirdio/netbird/management/server/integrations/integrated_validator/validator"
|
||||
|
||||
nbcache "github.com/netbirdio/netbird/management/server/cache"
|
||||
|
||||
@@ -109,7 +109,7 @@ func startManagement(t *testing.T, config *config.Config, testFile string) (*grp
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
iv, _ := integrations.NewIntegratedValidator(ctx, peersmanager, settingsManagerMock, eventStore, cacheStore)
|
||||
iv, _ := validator.NewIntegratedValidator(ctx, peersmanager, settingsManagerMock, eventStore, cacheStore)
|
||||
|
||||
metrics, err := telemetry.NewDefaultAppMetrics(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
@@ -12,7 +12,13 @@ var (
|
||||
Short: "Print the NetBird's client application version",
|
||||
Run: func(cmd *cobra.Command, args []string) {
|
||||
cmd.SetOut(cmd.OutOrStdout())
|
||||
cmd.Println(version.NetbirdVersion())
|
||||
out := version.NetbirdVersion()
|
||||
if version.IsDevelopmentVersion(out) {
|
||||
if commit := version.NetbirdCommit(); commit != "" {
|
||||
out += "-" + commit
|
||||
}
|
||||
}
|
||||
cmd.Println(out)
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"sync"
|
||||
|
||||
"github.com/sirupsen/logrus"
|
||||
wgdevice "golang.zx2c4.com/wireguard/device"
|
||||
wgnetstack "golang.zx2c4.com/wireguard/tun/netstack"
|
||||
|
||||
"github.com/netbirdio/netbird/client/iface"
|
||||
@@ -84,6 +85,12 @@ type Options struct {
|
||||
DisableIPv6 bool
|
||||
// BlockInbound blocks all inbound connections from peers
|
||||
BlockInbound bool
|
||||
// BlockLANAccess blocks the embedded peer from reaching the host's
|
||||
// LAN (RFC 1918, link-local, loopback) when it's used as a routing
|
||||
// peer. Mirrors profilemanager.ConfigInput.BlockLANAccess. Useful
|
||||
// when the embedded client must never act as a stepping stone into
|
||||
// the host's local network (e.g. the proxy's overlay peer).
|
||||
BlockLANAccess bool
|
||||
// WireguardPort is the port for the tunnel interface. Use 0 for a random port.
|
||||
WireguardPort *int
|
||||
// MTU is the MTU for the tunnel interface.
|
||||
@@ -94,6 +101,26 @@ type Options struct {
|
||||
MTU *uint16
|
||||
// DNSLabels defines additional DNS labels configured in the peer.
|
||||
DNSLabels []string
|
||||
// Performance configures the tunnel's buffer pool cap and batch size.
|
||||
Performance Performance
|
||||
}
|
||||
|
||||
// Performance configures the embedded client's tunnel memory/throughput knobs.
|
||||
//
|
||||
// These settings are process-global: any non-nil field also becomes the
|
||||
// default for Clients constructed by later embed.New calls in the same
|
||||
// process. Nil fields are ignored.
|
||||
type Performance struct {
|
||||
// PreallocatedBuffersPerPool caps the per-tunnel buffer pool. Zero
|
||||
// leaves the pool unbounded. Lower values trade throughput for a
|
||||
// tighter memory ceiling. May also be changed on a running Client via
|
||||
// Client.SetPerformance, provided this field was nonzero at construction.
|
||||
PreallocatedBuffersPerPool *uint32
|
||||
// MaxBatchSize overrides the number of packets the tunnel reads or
|
||||
// writes per syscall, which also bounds eager buffer allocation per
|
||||
// worker. Zero uses the platform default. Applied at construction
|
||||
// only; ignored by Client.SetPerformance.
|
||||
MaxBatchSize *uint32
|
||||
}
|
||||
|
||||
// validateCredentials checks that exactly one credential type is provided
|
||||
@@ -175,6 +202,7 @@ func New(opts Options) (*Client, error) {
|
||||
DisableClientRoutes: &opts.DisableClientRoutes,
|
||||
DisableIPv6: &opts.DisableIPv6,
|
||||
BlockInbound: &opts.BlockInbound,
|
||||
BlockLANAccess: &opts.BlockLANAccess,
|
||||
WireguardPort: opts.WireguardPort,
|
||||
MTU: opts.MTU,
|
||||
DNSLabels: parsedLabels,
|
||||
@@ -192,6 +220,13 @@ func New(opts Options) (*Client, error) {
|
||||
config.PrivateKey = opts.PrivateKey
|
||||
}
|
||||
|
||||
if opts.Performance.PreallocatedBuffersPerPool != nil {
|
||||
wgdevice.SetPreallocatedBuffersPerPool(*opts.Performance.PreallocatedBuffersPerPool)
|
||||
}
|
||||
if opts.Performance.MaxBatchSize != nil {
|
||||
wgdevice.SetMaxBatchSizeOverride(*opts.Performance.MaxBatchSize)
|
||||
}
|
||||
|
||||
return &Client{
|
||||
deviceName: opts.DeviceName,
|
||||
setupKey: opts.SetupKey,
|
||||
@@ -405,6 +440,21 @@ func (c *Client) Expose(ctx context.Context, req ExposeRequest) (*ExposeSession,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// IdentityForIP looks up a remote peer by its tunnel IP using the
|
||||
// embedded client's status recorder. Returns the peer's WireGuard public
|
||||
// key and FQDN. ok=false means the IP isn't in this client's peer
|
||||
// roster — callers should treat that as "unknown peer".
|
||||
func (c *Client) IdentityForIP(ip netip.Addr) (pubKey, fqdn string, ok bool) {
|
||||
if !ip.IsValid() || c.recorder == nil {
|
||||
return "", "", false
|
||||
}
|
||||
state, found := c.recorder.PeerStateByIP(ip.String())
|
||||
if !found {
|
||||
return "", "", false
|
||||
}
|
||||
return state.PubKey, state.FQDN, true
|
||||
}
|
||||
|
||||
// Status returns the current status of the client.
|
||||
func (c *Client) Status() (peer.FullStatus, error) {
|
||||
c.mu.Lock()
|
||||
@@ -473,6 +523,25 @@ func (c *Client) VerifySSHHostKey(peerAddress string, key []byte) error {
|
||||
return sshcommon.VerifyHostKey(storedKey, key, peerAddress)
|
||||
}
|
||||
|
||||
// SetPerformance retunes a running Client. Only PreallocatedBuffersPerPool
|
||||
// takes effect, and only when it was nonzero at construction;
|
||||
// MaxBatchSize is construction-only and returns an error if set here.
|
||||
//
|
||||
// Returns ErrClientNotStarted / ErrEngineNotStarted if the Client is not
|
||||
// running yet.
|
||||
func (c *Client) SetPerformance(t Performance) error {
|
||||
if t.MaxBatchSize != nil {
|
||||
return errors.New("MaxBatchSize is construction-only and cannot be changed at runtime")
|
||||
}
|
||||
engine, err := c.getEngine()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return engine.SetPerformance(internal.Performance{
|
||||
PreallocatedBuffersPerPool: t.PreallocatedBuffersPerPool,
|
||||
})
|
||||
}
|
||||
|
||||
// StartCapture begins capturing packets on this client's tunnel device.
|
||||
// Only one capture can be active at a time; starting a new one stops the previous.
|
||||
// Call StopCapture (or CaptureSession.Stop) to end it.
|
||||
|
||||
@@ -362,6 +362,10 @@ func (f *Forwarder) injectICMPv6Reply(id stack.TransportEndpointID, icmpPayload
|
||||
return 0
|
||||
}
|
||||
|
||||
if pc := f.endpoint.capture.Load(); pc != nil {
|
||||
(*pc).Offer(fullPacket, true)
|
||||
}
|
||||
|
||||
return len(fullPacket)
|
||||
}
|
||||
|
||||
|
||||
@@ -360,7 +360,13 @@ func isRedirectURLPortUsed(redirectURL string, excludedRanges []excludedPortRang
|
||||
return true
|
||||
}
|
||||
|
||||
addr := fmt.Sprintf(":%s", port)
|
||||
// FreeBSD 15 disables connecting to INADDR_ANY (0.0.0.0) as a localhost
|
||||
// alias by default, ensure explicit ip for localhost.
|
||||
host := parsedURL.Hostname()
|
||||
if host == "" {
|
||||
host = "127.0.0.1"
|
||||
}
|
||||
addr := net.JoinHostPort(host, port)
|
||||
conn, err := net.DialTimeout("tcp", addr, 3*time.Second)
|
||||
if err != nil {
|
||||
return false
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"fmt"
|
||||
"net"
|
||||
"net/netip"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"runtime/debug"
|
||||
"strings"
|
||||
@@ -348,6 +349,11 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
return wrapErr(err)
|
||||
}
|
||||
engineConfig.TempDir = mobileDependency.TempDir
|
||||
// Leave StateDir empty when there is no state path so a disk-backed
|
||||
// syncstore falls back to os.TempDir() instead of filepath.Dir("") == ".".
|
||||
if path != "" {
|
||||
engineConfig.StateDir = filepath.Dir(path)
|
||||
}
|
||||
|
||||
relayManager := relayClient.NewManager(engineCtx, relayURLs, myPrivateKey.PublicKey().String(), engineConfig.MTU)
|
||||
c.statusRecorder.SetRelayMgr(relayManager)
|
||||
|
||||
@@ -339,8 +339,7 @@ func (c *HandlerChain) isHandlerMatch(qname string, entry HandlerEntry) bool {
|
||||
case entry.Pattern == ".":
|
||||
return true
|
||||
case entry.IsWildcard:
|
||||
parts := strings.Split(strings.TrimSuffix(qname, entry.Pattern), ".")
|
||||
return len(parts) >= 2 && strings.HasSuffix(qname, entry.Pattern)
|
||||
return strings.HasSuffix(qname, "."+entry.Pattern)
|
||||
default:
|
||||
// For non-wildcard patterns:
|
||||
// If handler wants subdomain matching, allow suffix match
|
||||
|
||||
@@ -164,6 +164,54 @@ func TestHandlerChain_ServeDNS_DomainMatching(t *testing.T) {
|
||||
matchSubdomains: true,
|
||||
shouldMatch: true,
|
||||
},
|
||||
{
|
||||
name: "wildcard label-boundary mismatch (suffix overlap)",
|
||||
handlerDomain: "*.b.test.",
|
||||
queryDomain: "x.ab.test.",
|
||||
isWildcard: true,
|
||||
matchSubdomains: false,
|
||||
shouldMatch: false,
|
||||
},
|
||||
{
|
||||
name: "wildcard label-boundary match",
|
||||
handlerDomain: "*.b.test.",
|
||||
queryDomain: "x.b.test.",
|
||||
isWildcard: true,
|
||||
matchSubdomains: false,
|
||||
shouldMatch: true,
|
||||
},
|
||||
{
|
||||
name: "wildcard multi-label match",
|
||||
handlerDomain: "*.b.test.",
|
||||
queryDomain: "x.y.b.test.",
|
||||
isWildcard: true,
|
||||
matchSubdomains: false,
|
||||
shouldMatch: true,
|
||||
},
|
||||
{
|
||||
name: "wildcard no match on multi-label apex",
|
||||
handlerDomain: "*.b.test.",
|
||||
queryDomain: "b.test.",
|
||||
isWildcard: true,
|
||||
matchSubdomains: false,
|
||||
shouldMatch: false,
|
||||
},
|
||||
{
|
||||
name: "wildcard no match on unrelated suffix containment",
|
||||
handlerDomain: "*.example.com.",
|
||||
queryDomain: "notexample.com.",
|
||||
isWildcard: true,
|
||||
matchSubdomains: false,
|
||||
shouldMatch: false,
|
||||
},
|
||||
{
|
||||
name: "wildcard accepts pattern registered without trailing dot",
|
||||
handlerDomain: "*.b.test",
|
||||
queryDomain: "x.b.test.",
|
||||
isWildcard: true,
|
||||
matchSubdomains: false,
|
||||
shouldMatch: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
@@ -273,6 +321,19 @@ func TestHandlerChain_ServeDNS_OverlappingDomains(t *testing.T) {
|
||||
expectedCalls: 1,
|
||||
expectedHandler: 2, // highest priority matching handler should be called
|
||||
},
|
||||
{
|
||||
name: "overlapping wildcard suffixes route to correct handler",
|
||||
handlers: []struct {
|
||||
pattern string
|
||||
priority int
|
||||
}{
|
||||
{pattern: "*.b.test.", priority: nbdns.PriorityDNSRoute},
|
||||
{pattern: "*.ab.test.", priority: nbdns.PriorityDNSRoute},
|
||||
},
|
||||
queryDomain: "app.ab.test.",
|
||||
expectedCalls: 1,
|
||||
expectedHandler: 1,
|
||||
},
|
||||
{
|
||||
name: "root zone with specific domain",
|
||||
handlers: []struct {
|
||||
|
||||
@@ -26,6 +26,19 @@ type resolver interface {
|
||||
LookupNetIP(ctx context.Context, network, host string) ([]netip.Addr, error)
|
||||
}
|
||||
|
||||
// PeerConnectivity reports whether a tunnel IP belongs to a peer the
|
||||
// client knows about and whether that peer is currently connected. The
|
||||
// local resolver uses this to suppress A/AAAA answers whose RDATA points
|
||||
// at a disconnected peer (typical case: a synthesized private-service
|
||||
// record pointing at an embedded proxy peer that just went offline).
|
||||
//
|
||||
// known=false means the IP isn't in the local peerstore at all — the
|
||||
// record is left alone (it points at something outside our mesh, e.g.
|
||||
// a non-peer upstream).
|
||||
type PeerConnectivity interface {
|
||||
IsConnectedByIP(ip string) (known, connected bool)
|
||||
}
|
||||
|
||||
type Resolver struct {
|
||||
mu sync.RWMutex
|
||||
records map[dns.Question][]dns.RR
|
||||
@@ -33,6 +46,11 @@ type Resolver struct {
|
||||
// zones maps zone domain -> NonAuthoritative (true = non-authoritative, user-created zone)
|
||||
zones map[domain.Domain]bool
|
||||
resolver resolver
|
||||
// peerConn, when non-nil, is consulted on every A/AAAA answer to
|
||||
// drop records pointing at disconnected peers. nil disables the
|
||||
// filter and preserves the legacy "return whatever is registered"
|
||||
// behaviour for callers that never wire a status source.
|
||||
peerConn PeerConnectivity
|
||||
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
@@ -49,6 +67,15 @@ func NewResolver() *Resolver {
|
||||
}
|
||||
}
|
||||
|
||||
// SetPeerConnectivity wires the per-IP connectivity check used to filter
|
||||
// out A/AAAA answers pointing at disconnected peers. Pass nil to disable.
|
||||
// Safe to call multiple times; the latest value wins.
|
||||
func (d *Resolver) SetPeerConnectivity(p PeerConnectivity) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
d.peerConn = p
|
||||
}
|
||||
|
||||
func (d *Resolver) MatchSubdomains() bool {
|
||||
return true
|
||||
}
|
||||
@@ -95,6 +122,7 @@ func (d *Resolver) ServeDNS(w dns.ResponseWriter, r *dns.Msg) {
|
||||
replyMessage.RecursionAvailable = true
|
||||
|
||||
result := d.lookupRecords(logger, question)
|
||||
result.records = d.filterDisconnectedPeerAnswers(logger, question, result.records)
|
||||
replyMessage.Authoritative = !result.hasExternalData
|
||||
replyMessage.Answer = result.records
|
||||
replyMessage.Rcode = d.determineRcode(question, result)
|
||||
@@ -436,6 +464,78 @@ func (d *Resolver) logDNSError(logger *log.Entry, hostname string, qtype uint16,
|
||||
}
|
||||
}
|
||||
|
||||
// filterDisconnectedPeerAnswers drops A/AAAA records whose RDATA matches
|
||||
// a known but disconnected peer. The synthesized private-service zones
|
||||
// emit one A record per connected proxy peer in a cluster; when a peer
|
||||
// goes offline, the server-side refresh removes the record from the
|
||||
// next netmap, but the client may still hold the previous netmap for a
|
||||
// short window. This filter is the local belt to that braces — even on
|
||||
// the stale netmap, the resolver hides the offline target.
|
||||
//
|
||||
// Records pointing at unknown IPs (outside the local peerstore, e.g.
|
||||
// non-mesh upstreams) are never dropped. Non-A/AAAA records pass
|
||||
// through untouched.
|
||||
//
|
||||
// Escape hatch: if filtering would leave the answer empty AND at least
|
||||
// one record was filtered, the original list is returned. Better to
|
||||
// hand the client a record that may not respond than NXDOMAIN it
|
||||
// completely when every proxy peer is offline (the upstream may still
|
||||
// be reachable some other way, or the peerstore may be stale).
|
||||
func (d *Resolver) filterDisconnectedPeerAnswers(logger *log.Entry, question dns.Question, records []dns.RR) []dns.RR {
|
||||
if len(records) == 0 {
|
||||
return records
|
||||
}
|
||||
d.mu.RLock()
|
||||
checker := d.peerConn
|
||||
d.mu.RUnlock()
|
||||
if checker == nil {
|
||||
return records
|
||||
}
|
||||
|
||||
kept := make([]dns.RR, 0, len(records))
|
||||
var dropped int
|
||||
for _, rr := range records {
|
||||
ip := extractRecordIP(rr)
|
||||
if ip == "" {
|
||||
kept = append(kept, rr)
|
||||
continue
|
||||
}
|
||||
known, connected := checker.IsConnectedByIP(ip)
|
||||
if known && !connected {
|
||||
dropped++
|
||||
continue
|
||||
}
|
||||
kept = append(kept, rr)
|
||||
}
|
||||
if dropped == 0 {
|
||||
return records
|
||||
}
|
||||
if len(kept) == 0 {
|
||||
logger.Debugf("all %d answers for %s point at disconnected peers; returning the original list", dropped, question.Name)
|
||||
return records
|
||||
}
|
||||
logger.Tracef("dropped %d disconnected-peer answer(s) for %s, returning %d", dropped, question.Name, len(kept))
|
||||
return kept
|
||||
}
|
||||
|
||||
// extractRecordIP returns the dotted-decimal / colon-hex IP carried by
|
||||
// an A or AAAA record, or "" for any other record type.
|
||||
func extractRecordIP(rr dns.RR) string {
|
||||
switch r := rr.(type) {
|
||||
case *dns.A:
|
||||
if r.A == nil {
|
||||
return ""
|
||||
}
|
||||
return r.A.String()
|
||||
case *dns.AAAA:
|
||||
if r.AAAA == nil {
|
||||
return ""
|
||||
}
|
||||
return r.AAAA.String()
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// Update replaces all zones and their records
|
||||
func (d *Resolver) Update(customZones []nbdns.CustomZone) {
|
||||
d.mu.Lock()
|
||||
|
||||
@@ -30,6 +30,21 @@ func (m *mockResolver) LookupNetIP(ctx context.Context, network, host string) ([
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// mockPeerConnectivity returns canned (known, connected) results per IP.
|
||||
// Used by the disconnected-peer filter tests below. IPs not in the map
|
||||
// are reported as unknown so the filter leaves them alone.
|
||||
type mockPeerConnectivity struct {
|
||||
byIP map[string]struct{ known, connected bool }
|
||||
}
|
||||
|
||||
func (m mockPeerConnectivity) IsConnectedByIP(ip string) (known, connected bool) {
|
||||
v, ok := m.byIP[ip]
|
||||
if !ok {
|
||||
return false, false
|
||||
}
|
||||
return v.known, v.connected
|
||||
}
|
||||
|
||||
func TestLocalResolver_ServeDNS(t *testing.T) {
|
||||
recordA := nbdns.SimpleRecord{
|
||||
Name: "peera.netbird.cloud.",
|
||||
@@ -2652,3 +2667,114 @@ func BenchmarkIsInManagedZone_ManyZones(b *testing.B) {
|
||||
resolver.isInManagedZone(qname)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLocalResolver_FilterDisconnectedPeerAnswers verifies the
|
||||
// connectivity-aware filtering layered on top of lookupRecords:
|
||||
// when an A record's IP belongs to a known peer that's disconnected,
|
||||
// the record is dropped from the answer. Records for unknown IPs pass
|
||||
// through. If filtering would empty the answer entirely and at least
|
||||
// one record was dropped, the original list is restored (escape hatch
|
||||
// for the "all proxies offline" case).
|
||||
func TestLocalResolver_FilterDisconnectedPeerAnswers(t *testing.T) {
|
||||
zone := "svc.cluster.netbird."
|
||||
connectedRec := nbdns.SimpleRecord{
|
||||
Name: zone,
|
||||
Type: int(dns.TypeA),
|
||||
Class: nbdns.DefaultClass,
|
||||
TTL: 5,
|
||||
RData: "100.64.0.10",
|
||||
}
|
||||
disconnectedRec := nbdns.SimpleRecord{
|
||||
Name: zone,
|
||||
Type: int(dns.TypeA),
|
||||
Class: nbdns.DefaultClass,
|
||||
TTL: 5,
|
||||
RData: "100.64.0.11",
|
||||
}
|
||||
unknownRec := nbdns.SimpleRecord{
|
||||
Name: zone,
|
||||
Type: int(dns.TypeA),
|
||||
Class: nbdns.DefaultClass,
|
||||
TTL: 5,
|
||||
RData: "203.0.113.5",
|
||||
}
|
||||
|
||||
type ipState struct{ known, connected bool }
|
||||
tests := []struct {
|
||||
name string
|
||||
records []nbdns.SimpleRecord
|
||||
connByIP map[string]ipState
|
||||
wantInOrder []string
|
||||
}{
|
||||
{
|
||||
name: "drops disconnected peer, keeps connected",
|
||||
records: []nbdns.SimpleRecord{connectedRec, disconnectedRec},
|
||||
connByIP: map[string]ipState{
|
||||
"100.64.0.10": {known: true, connected: true},
|
||||
"100.64.0.11": {known: true, connected: false},
|
||||
},
|
||||
wantInOrder: []string{"100.64.0.10"},
|
||||
},
|
||||
{
|
||||
name: "unknown IPs pass through untouched",
|
||||
records: []nbdns.SimpleRecord{unknownRec, disconnectedRec},
|
||||
connByIP: map[string]ipState{
|
||||
"100.64.0.11": {known: true, connected: false},
|
||||
},
|
||||
wantInOrder: []string{"203.0.113.5"},
|
||||
},
|
||||
{
|
||||
name: "all disconnected falls back to original list",
|
||||
records: []nbdns.SimpleRecord{disconnectedRec, connectedRec},
|
||||
connByIP: map[string]ipState{
|
||||
"100.64.0.10": {known: true, connected: false},
|
||||
"100.64.0.11": {known: true, connected: false},
|
||||
},
|
||||
wantInOrder: []string{"100.64.0.11", "100.64.0.10"},
|
||||
},
|
||||
{
|
||||
name: "no checker wired returns all records",
|
||||
records: []nbdns.SimpleRecord{connectedRec, disconnectedRec},
|
||||
connByIP: nil,
|
||||
wantInOrder: []string{"100.64.0.10", "100.64.0.11"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
resolver := NewResolver()
|
||||
if tc.connByIP != nil {
|
||||
cm := mockPeerConnectivity{byIP: make(map[string]struct{ known, connected bool }, len(tc.connByIP))}
|
||||
for ip, st := range tc.connByIP {
|
||||
cm.byIP[ip] = struct{ known, connected bool }{st.known, st.connected}
|
||||
}
|
||||
resolver.SetPeerConnectivity(cm)
|
||||
}
|
||||
resolver.Update([]nbdns.CustomZone{{
|
||||
Domain: strings.TrimSuffix(zone, "."),
|
||||
Records: tc.records,
|
||||
NonAuthoritative: true,
|
||||
}})
|
||||
|
||||
var got *dns.Msg
|
||||
writer := &test.MockResponseWriter{
|
||||
WriteMsgFunc: func(m *dns.Msg) error {
|
||||
got = m
|
||||
return nil
|
||||
},
|
||||
}
|
||||
req := new(dns.Msg).SetQuestion(zone, dns.TypeA)
|
||||
resolver.ServeDNS(writer, req)
|
||||
|
||||
require.NotNil(t, got, "resolver must produce a response")
|
||||
require.Len(t, got.Answer, len(tc.wantInOrder),
|
||||
"answer count must match expected: %v", tc.wantInOrder)
|
||||
for i, want := range tc.wantInOrder {
|
||||
a, ok := got.Answer[i].(*dns.A)
|
||||
require.True(t, ok, "answer[%d] must be an A record", i)
|
||||
assert.Equal(t, want, a.A.String(),
|
||||
"answer[%d] expected %s got %s", i, want, a.A.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -301,6 +301,11 @@ func newDefaultServer(
|
||||
warningDelayBase: defaultWarningDelayBase,
|
||||
healthRefresh: make(chan struct{}, 1),
|
||||
}
|
||||
// Wire the local resolver against the peer status recorder so it can
|
||||
// suppress A/AAAA answers that point at disconnected peers (typical
|
||||
// case: synthesised private-service records pointing at an embedded
|
||||
// proxy peer that just went offline).
|
||||
defaultServer.localResolver.SetPeerConnectivity(localPeerConnectivity{statusRecorder})
|
||||
|
||||
// register with root zone, handler chain takes care of the routing
|
||||
dnsService.RegisterMux(".", handlerChain)
|
||||
@@ -1386,3 +1391,25 @@ func (s *DefaultServer) PopulateManagementDomain(mgmtURL *url.URL) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// localPeerConnectivity adapts *peer.Status to local.PeerConnectivity so
|
||||
// the local resolver can ask "is this IP a known peer and is it
|
||||
// connected?" without taking on the peer package as a dependency.
|
||||
// A nil status recorder always reports known=false so the resolver
|
||||
// short-circuits to the legacy "return everything" path.
|
||||
type localPeerConnectivity struct {
|
||||
status *peer.Status
|
||||
}
|
||||
|
||||
// IsConnectedByIP looks the IP up in the peerstore and surfaces both
|
||||
// the known and connected bits. Used by Resolver.filterDisconnectedPeerAnswers.
|
||||
func (l localPeerConnectivity) IsConnectedByIP(ip string) (known, connected bool) {
|
||||
if l.status == nil {
|
||||
return false, false
|
||||
}
|
||||
state, ok := l.status.PeerStateByIP(ip)
|
||||
if !ok {
|
||||
return false, false
|
||||
}
|
||||
return true, state.ConnStatus == peer.StatusConnected
|
||||
}
|
||||
|
||||
+78
-37
@@ -22,7 +22,6 @@ import (
|
||||
log "github.com/sirupsen/logrus"
|
||||
"golang.zx2c4.com/wireguard/tun/netstack"
|
||||
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
|
||||
"google.golang.org/protobuf/proto"
|
||||
|
||||
nberrors "github.com/netbirdio/netbird/client/errors"
|
||||
"github.com/netbirdio/netbird/client/firewall"
|
||||
@@ -56,6 +55,7 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/routemanager"
|
||||
"github.com/netbirdio/netbird/client/internal/routemanager/systemops"
|
||||
"github.com/netbirdio/netbird/client/internal/statemanager"
|
||||
"github.com/netbirdio/netbird/client/internal/syncstore"
|
||||
"github.com/netbirdio/netbird/client/internal/updater"
|
||||
"github.com/netbirdio/netbird/client/jobexec"
|
||||
cProto "github.com/netbirdio/netbird/client/proto"
|
||||
@@ -148,6 +148,10 @@ type EngineConfig struct {
|
||||
|
||||
LogPath string
|
||||
TempDir string
|
||||
|
||||
// StateDir is the directory holding the state file. The sync response
|
||||
// (network map) is serialized here on platforms that persist it to disk.
|
||||
StateDir string
|
||||
}
|
||||
|
||||
// EngineServices holds the external service dependencies required by the Engine.
|
||||
@@ -226,10 +230,15 @@ type Engine struct {
|
||||
|
||||
afpacketCapture *capture.AFPacketCapture
|
||||
|
||||
// Sync response persistence (protected by syncRespMux)
|
||||
syncRespMux sync.RWMutex
|
||||
persistSyncResponse bool
|
||||
latestSyncResponse *mgmProto.SyncResponse
|
||||
// Sync response persistence (protected by syncRespMux).
|
||||
// syncStore is nil unless persistence has been enabled; its presence is
|
||||
// what marks persistence as active. The backend (disk or memory) is
|
||||
// selected per-platform; see the syncstore package. syncStoreDir is where
|
||||
// a disk-backed store serializes to.
|
||||
syncRespMux sync.RWMutex
|
||||
syncStore syncstore.Store
|
||||
syncStoreDir string
|
||||
|
||||
flowManager nftypes.FlowManager
|
||||
|
||||
// auto-update
|
||||
@@ -292,6 +301,7 @@ func NewEngine(
|
||||
jobExecutor: jobexec.NewExecutor(),
|
||||
clientMetrics: services.ClientMetrics,
|
||||
updateManager: services.UpdateManager,
|
||||
syncStoreDir: config.StateDir,
|
||||
}
|
||||
|
||||
log.Infof("I am: %s", config.WgPrivateKey.PublicKey().String())
|
||||
@@ -913,19 +923,18 @@ func (e *Engine) handleSync(update *mgmProto.SyncResponse) error {
|
||||
}
|
||||
|
||||
// Persist sync response under the dedicated lock (syncRespMux), not under syncMsgMux.
|
||||
// Read the storage-enabled flag under the syncRespMux too.
|
||||
// A non-nil syncStore is what marks persistence as enabled. Hold the lock for
|
||||
// the whole Set so the store cannot be cleared (disabled / engine close)
|
||||
// mid-call and have this write resurrect a file that was just removed.
|
||||
e.syncRespMux.RLock()
|
||||
enabled := e.persistSyncResponse
|
||||
e.syncRespMux.RUnlock()
|
||||
|
||||
// Store sync response if persistence is enabled
|
||||
if enabled {
|
||||
e.syncRespMux.Lock()
|
||||
e.latestSyncResponse = update
|
||||
e.syncRespMux.Unlock()
|
||||
|
||||
log.Debugf("sync response persisted with serial %d", nm.GetSerial())
|
||||
if e.syncStore != nil {
|
||||
if err := e.syncStore.Set(update); err != nil {
|
||||
log.Errorf("failed to persist sync response: %v", err)
|
||||
} else {
|
||||
log.Debugf("sync response persisted with serial %d", nm.GetSerial())
|
||||
}
|
||||
}
|
||||
e.syncRespMux.RUnlock()
|
||||
|
||||
// only apply new changes and ignore old ones
|
||||
if err := e.updateNetworkMap(nm); err != nil {
|
||||
@@ -1813,6 +1822,18 @@ func (e *Engine) close() {
|
||||
if err := e.portForwardManager.GracefullyStop(ctx); err != nil {
|
||||
log.Warnf("failed to gracefully stop port forwarding manager: %s", err)
|
||||
}
|
||||
|
||||
// Drop any persisted sync response so its network map does not linger on
|
||||
// disk after the engine stops (and cannot leak into a later run).
|
||||
e.syncRespMux.Lock()
|
||||
store := e.syncStore
|
||||
e.syncStore = nil
|
||||
e.syncRespMux.Unlock()
|
||||
if store != nil {
|
||||
if err := store.Clear(); err != nil {
|
||||
log.Warnf("failed to clear persisted sync response on close: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (e *Engine) readInitialSettings() ([]*route.Route, *nbdns.Config, bool, error) {
|
||||
@@ -1967,6 +1988,29 @@ func (e *Engine) GetClientMetrics() *metrics.ClientMetrics {
|
||||
return e.clientMetrics
|
||||
}
|
||||
|
||||
// Performance bundles runtime-adjustable tunnel pool knobs.
|
||||
// See Engine.SetPerformance. Nil fields are ignored.
|
||||
type Performance struct {
|
||||
PreallocatedBuffersPerPool *uint32
|
||||
}
|
||||
|
||||
// SetPerformance applies the given tuning to this engine's live Device.
|
||||
func (e *Engine) SetPerformance(t Performance) error {
|
||||
e.syncMsgMux.Lock()
|
||||
defer e.syncMsgMux.Unlock()
|
||||
if e.wgInterface == nil {
|
||||
return fmt.Errorf("wg interface not initialized")
|
||||
}
|
||||
dev := e.wgInterface.GetWGDevice()
|
||||
if dev == nil {
|
||||
return fmt.Errorf("wg device not initialized")
|
||||
}
|
||||
if t.PreallocatedBuffersPerPool != nil {
|
||||
dev.SetPreallocatedBuffersPerPool(*t.PreallocatedBuffersPerPool)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func findIPFromInterfaceName(ifaceName string) (net.IP, error) {
|
||||
iface, err := net.InterfaceByName(ifaceName)
|
||||
if err != nil {
|
||||
@@ -2119,45 +2163,42 @@ func (e *Engine) stopDNSServer() {
|
||||
e.statusRecorder.UpdateDNSStates(nsGroupStates)
|
||||
}
|
||||
|
||||
// SetSyncResponsePersistence enables or disables sync response persistence
|
||||
// SetSyncResponsePersistence enables or disables sync response persistence.
|
||||
// The store is only instantiated while persistence is enabled; construction
|
||||
// itself drops any stale data left over from an earlier run (see syncstore).
|
||||
func (e *Engine) SetSyncResponsePersistence(enabled bool) {
|
||||
e.syncRespMux.Lock()
|
||||
defer e.syncRespMux.Unlock()
|
||||
|
||||
if enabled == e.persistSyncResponse {
|
||||
if enabled == (e.syncStore != nil) {
|
||||
return
|
||||
}
|
||||
e.persistSyncResponse = enabled
|
||||
log.Debugf("Sync response persistence is set to %t", enabled)
|
||||
|
||||
if !enabled {
|
||||
e.latestSyncResponse = nil
|
||||
if err := e.syncStore.Clear(); err != nil {
|
||||
log.Warnf("failed to clear persisted sync response: %v", err)
|
||||
}
|
||||
e.syncStore = nil
|
||||
return
|
||||
}
|
||||
|
||||
e.syncStore = syncstore.New(e.syncStoreDir)
|
||||
}
|
||||
|
||||
// GetLatestSyncResponse returns the stored sync response if persistence is enabled
|
||||
func (e *Engine) GetLatestSyncResponse() (*mgmProto.SyncResponse, error) {
|
||||
// Hold the lock for the whole Get so the store cannot be cleared
|
||||
// (disabled / engine close) mid-call.
|
||||
e.syncRespMux.RLock()
|
||||
enabled := e.persistSyncResponse
|
||||
latest := e.latestSyncResponse
|
||||
e.syncRespMux.RUnlock()
|
||||
defer e.syncRespMux.RUnlock()
|
||||
|
||||
if !enabled {
|
||||
if e.syncStore == nil {
|
||||
return nil, errors.New("sync response persistence is disabled")
|
||||
}
|
||||
|
||||
if latest == nil {
|
||||
//nolint:nilnil
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
log.Debugf("Retrieving latest sync response with size %d bytes", proto.Size(latest))
|
||||
sr, ok := proto.Clone(latest).(*mgmProto.SyncResponse)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("failed to clone sync response")
|
||||
}
|
||||
|
||||
return sr, nil
|
||||
//nolint:nilnil
|
||||
return e.syncStore.Get()
|
||||
}
|
||||
|
||||
// GetWgAddr returns the wireguard address
|
||||
@@ -2193,7 +2234,7 @@ func (e *Engine) updateDNSForwarder(
|
||||
enabled bool,
|
||||
fwdEntries []*dnsfwd.ForwarderEntry,
|
||||
) {
|
||||
if e.config.DisableServerRoutes {
|
||||
if e.config.DisableServerRoutes || e.config.BlockInbound {
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -27,7 +27,7 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/stdnet"
|
||||
"github.com/netbirdio/netbird/management/server/job"
|
||||
|
||||
"github.com/netbirdio/management-integrations/integrations"
|
||||
"github.com/netbirdio/netbird/management/server/integrations/integrated_validator/validator"
|
||||
|
||||
"github.com/netbirdio/netbird/management/internals/controllers/network_map/controller"
|
||||
"github.com/netbirdio/netbird/management/internals/controllers/network_map/update_channel"
|
||||
@@ -66,8 +66,8 @@ import (
|
||||
"github.com/netbirdio/netbird/route"
|
||||
mgmt "github.com/netbirdio/netbird/shared/management/client"
|
||||
mgmtProto "github.com/netbirdio/netbird/shared/management/proto"
|
||||
relayClient "github.com/netbirdio/netbird/shared/relay/client"
|
||||
"github.com/netbirdio/netbird/shared/netiputil"
|
||||
relayClient "github.com/netbirdio/netbird/shared/relay/client"
|
||||
signal "github.com/netbirdio/netbird/shared/signal/client"
|
||||
"github.com/netbirdio/netbird/shared/signal/proto"
|
||||
signalServer "github.com/netbirdio/netbird/signal/server"
|
||||
@@ -1641,7 +1641,7 @@ func startManagement(t *testing.T, dataDir, testFile string) (*grpc.Server, stri
|
||||
return nil, "", err
|
||||
}
|
||||
|
||||
ia, _ := integrations.NewIntegratedValidator(context.Background(), peersManager, nil, eventStore, cacheStore)
|
||||
ia, _ := validator.NewIntegratedValidator(context.Background(), peersManager, nil, eventStore, cacheStore)
|
||||
|
||||
metrics, err := telemetry.NewDefaultAppMetrics(context.Background())
|
||||
require.NoError(t, err)
|
||||
|
||||
@@ -4,6 +4,8 @@ import (
|
||||
"strings"
|
||||
|
||||
"github.com/hashicorp/go-version"
|
||||
|
||||
nbversion "github.com/netbirdio/netbird/version"
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -11,7 +13,7 @@ var (
|
||||
)
|
||||
|
||||
func IsSupported(agentVersion string) bool {
|
||||
if agentVersion == "development" {
|
||||
if nbversion.IsDevelopmentVersion(agentVersion) {
|
||||
return true
|
||||
}
|
||||
|
||||
|
||||
@@ -50,7 +50,7 @@ func routeCheck(ctx context.Context, fd int, nexthopv4, nexthopv6 systemops.Next
|
||||
switch msg.Type {
|
||||
// handle route changes
|
||||
case unix.RTM_ADD, syscall.RTM_DELETE:
|
||||
route, err := parseRouteMessage(buf[:n])
|
||||
route, flags, err := parseRouteMessage(buf[:n])
|
||||
if err != nil {
|
||||
log.Debugf("Network monitor: error parsing routing message: %v", err)
|
||||
continue
|
||||
@@ -66,6 +66,10 @@ func routeCheck(ctx context.Context, fd int, nexthopv4, nexthopv6 systemops.Next
|
||||
}
|
||||
switch msg.Type {
|
||||
case unix.RTM_ADD:
|
||||
if systemops.IgnoreAddedDefaultRoute(flags) {
|
||||
log.Debugf("Network monitor: ignoring added default route via %s, interface %s, flags %#x", route.Gw, intf, flags)
|
||||
continue
|
||||
}
|
||||
log.Infof("Network monitor: default route changed: via %s, interface %s", route.Gw, intf)
|
||||
return nil
|
||||
case unix.RTM_DELETE:
|
||||
@@ -78,22 +82,26 @@ func routeCheck(ctx context.Context, fd int, nexthopv4, nexthopv6 systemops.Next
|
||||
}
|
||||
}
|
||||
|
||||
func parseRouteMessage(buf []byte) (*systemops.Route, error) {
|
||||
func parseRouteMessage(buf []byte) (*systemops.Route, int, error) {
|
||||
msgs, err := route.ParseRIB(route.RIBTypeRoute, buf)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse RIB: %v", err)
|
||||
return nil, 0, fmt.Errorf("parse RIB: %v", err)
|
||||
}
|
||||
|
||||
if len(msgs) != 1 {
|
||||
return nil, fmt.Errorf("unexpected RIB message msgs: %v", msgs)
|
||||
return nil, 0, fmt.Errorf("unexpected RIB message msgs: %v", msgs)
|
||||
}
|
||||
|
||||
msg, ok := msgs[0].(*route.RouteMessage)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("unexpected RIB message type: %T", msgs[0])
|
||||
return nil, 0, fmt.Errorf("unexpected RIB message type: %T", msgs[0])
|
||||
}
|
||||
|
||||
return systemops.MsgToRoute(msg)
|
||||
r, err := systemops.MsgToRoute(msg)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return r, msg.Flags, nil
|
||||
}
|
||||
|
||||
// waitReadable blocks until fd has data to read, or ctx is cancelled.
|
||||
|
||||
@@ -23,6 +23,7 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/peer/id"
|
||||
"github.com/netbirdio/netbird/client/internal/peer/worker"
|
||||
"github.com/netbirdio/netbird/client/internal/portforward"
|
||||
"github.com/netbirdio/netbird/client/internal/rosenpass"
|
||||
"github.com/netbirdio/netbird/client/internal/stdnet"
|
||||
"github.com/netbirdio/netbird/route"
|
||||
relayClient "github.com/netbirdio/netbird/shared/relay/client"
|
||||
@@ -899,7 +900,7 @@ func (conn *Conn) presharedKey(remoteRosenpassKey []byte) *wgtypes.Key {
|
||||
}
|
||||
|
||||
// Fallback to deterministic key if no NetBird PSK is configured
|
||||
determKey, err := conn.rosenpassDetermKey()
|
||||
determKey, err := rosenpass.DeterministicSeedKey(conn.config.LocalKey, conn.config.Key)
|
||||
if err != nil {
|
||||
conn.Log.Errorf("failed to generate Rosenpass initial key: %v", err)
|
||||
return nil
|
||||
@@ -908,26 +909,6 @@ func (conn *Conn) presharedKey(remoteRosenpassKey []byte) *wgtypes.Key {
|
||||
return determKey
|
||||
}
|
||||
|
||||
// todo: move this logic into Rosenpass package
|
||||
func (conn *Conn) rosenpassDetermKey() (*wgtypes.Key, error) {
|
||||
lk := []byte(conn.config.LocalKey)
|
||||
rk := []byte(conn.config.Key) // remote key
|
||||
var keyInput []byte
|
||||
if string(lk) > string(rk) {
|
||||
//nolint:gocritic
|
||||
keyInput = append(lk[:16], rk[:16]...)
|
||||
} else {
|
||||
//nolint:gocritic
|
||||
keyInput = append(rk[:16], lk[:16]...)
|
||||
}
|
||||
|
||||
key, err := wgtypes.NewKey(keyInput)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &key, nil
|
||||
}
|
||||
|
||||
func isController(config ConnConfig) bool {
|
||||
return config.LocalKey > config.Key
|
||||
}
|
||||
|
||||
@@ -185,9 +185,12 @@ func (s *StatusChangeSubscription) Events() chan map[string]RouterState {
|
||||
return s.eventsChan
|
||||
}
|
||||
|
||||
// Status holds a state of peers, signal, management connections and relays
|
||||
// Status holds a state of peers, signal, management connections and relays.
|
||||
// mux is an RWMutex so hot read paths (notably PeerStateByIP, called for
|
||||
// every private-service request) don't contend against each other.
|
||||
// Pure read methods take RLock; anything that mutates state takes Lock.
|
||||
type Status struct {
|
||||
mux sync.Mutex
|
||||
mux sync.RWMutex
|
||||
peers map[string]State
|
||||
changeNotify map[string]map[string]*StatusChangeSubscription // map[peerID]map[subscriptionID]*StatusChangeSubscription
|
||||
signalState bool
|
||||
@@ -283,8 +286,8 @@ func (d *Status) AddPeer(peerPubKey string, fqdn string, ip string, ipv6 string)
|
||||
|
||||
// GetPeer adds peer to Daemon status map
|
||||
func (d *Status) GetPeer(peerPubKey string) (State, error) {
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
|
||||
state, ok := d.peers[peerPubKey]
|
||||
if !ok {
|
||||
@@ -294,8 +297,8 @@ func (d *Status) GetPeer(peerPubKey string) (State, error) {
|
||||
}
|
||||
|
||||
func (d *Status) PeerByIP(ip string) (string, bool) {
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
|
||||
for _, state := range d.peers {
|
||||
if state.IP == ip {
|
||||
@@ -305,6 +308,34 @@ func (d *Status) PeerByIP(ip string) (string, bool) {
|
||||
return "", false
|
||||
}
|
||||
|
||||
// PeerStateByIP returns the full peer State for the given tunnel IP.
|
||||
// Matches against either the IPv4 (State.IP) or IPv6 (State.IPv6) tunnel
|
||||
// address so dual-stack peers are reachable on either family. Searches
|
||||
// both d.peers and d.offlinePeers — peers that have been moved into
|
||||
// the offline slice by ReplaceOfflinePeers are still part of the
|
||||
// account's roster and callers (DNS filter, embed.Client.IdentityForIP)
|
||||
// need to recognise them rather than treating them as unknown. Returns
|
||||
// the zero State and false when no peer matches or the input is empty.
|
||||
func (d *Status) PeerStateByIP(ip string) (State, bool) {
|
||||
if ip == "" {
|
||||
return State{}, false
|
||||
}
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
|
||||
for _, state := range d.peers {
|
||||
if (state.IP != "" && state.IP == ip) || (state.IPv6 != "" && state.IPv6 == ip) {
|
||||
return state, true
|
||||
}
|
||||
}
|
||||
for _, state := range d.offlinePeers {
|
||||
if (state.IP != "" && state.IP == ip) || (state.IPv6 != "" && state.IPv6 == ip) {
|
||||
return state, true
|
||||
}
|
||||
}
|
||||
return State{}, false
|
||||
}
|
||||
|
||||
// RemovePeer removes peer from Daemon status map
|
||||
func (d *Status) RemovePeer(peerPubKey string) error {
|
||||
d.mux.Lock()
|
||||
@@ -702,8 +733,8 @@ func (d *Status) UnsubscribePeerStateChanges(subscription *StatusChangeSubscript
|
||||
|
||||
// GetLocalPeerState returns the local peer state
|
||||
func (d *Status) GetLocalPeerState() LocalPeerState {
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
return d.localPeer.Clone()
|
||||
}
|
||||
|
||||
@@ -909,8 +940,8 @@ func (d *Status) DeleteResolvedDomainsStates(domain domain.Domain) {
|
||||
}
|
||||
|
||||
func (d *Status) GetRosenpassState() RosenpassState {
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
return RosenpassState{
|
||||
d.rosenpassEnabled,
|
||||
d.rosenpassPermissive,
|
||||
@@ -918,14 +949,14 @@ func (d *Status) GetRosenpassState() RosenpassState {
|
||||
}
|
||||
|
||||
func (d *Status) GetLazyConnection() bool {
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
return d.lazyConnectionEnabled
|
||||
}
|
||||
|
||||
func (d *Status) GetManagementState() ManagementState {
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
return ManagementState{
|
||||
d.mgmAddress,
|
||||
d.managementState,
|
||||
@@ -951,8 +982,8 @@ func (d *Status) UpdateLatency(pubKey string, latency time.Duration) error {
|
||||
|
||||
// IsLoginRequired determines if a peer's login has expired.
|
||||
func (d *Status) IsLoginRequired() bool {
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
|
||||
// if peer is connected to the management then login is not expired
|
||||
if d.managementState {
|
||||
@@ -967,8 +998,8 @@ func (d *Status) IsLoginRequired() bool {
|
||||
}
|
||||
|
||||
func (d *Status) GetSignalState() SignalState {
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
return SignalState{
|
||||
d.signalAddress,
|
||||
d.signalState,
|
||||
@@ -978,8 +1009,8 @@ func (d *Status) GetSignalState() SignalState {
|
||||
|
||||
// GetRelayStates returns the stun/turn/permanent relay states
|
||||
func (d *Status) GetRelayStates() []relay.ProbeResult {
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
if d.relayMgr == nil {
|
||||
return d.relayStates
|
||||
}
|
||||
@@ -1008,8 +1039,8 @@ func (d *Status) GetRelayStates() []relay.ProbeResult {
|
||||
}
|
||||
|
||||
func (d *Status) ForwardingRules() []firewall.ForwardRule {
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
if d.ingressGwMgr == nil {
|
||||
return nil
|
||||
}
|
||||
@@ -1018,16 +1049,16 @@ func (d *Status) ForwardingRules() []firewall.ForwardRule {
|
||||
}
|
||||
|
||||
func (d *Status) GetDNSStates() []NSGroupState {
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
|
||||
// shallow copy is good enough, as slices fields are currently not updated
|
||||
return slices.Clone(d.nsGroupStates)
|
||||
}
|
||||
|
||||
func (d *Status) GetResolvedDomainsStates() map[domain.Domain]ResolvedDomainInfo {
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
return maps.Clone(d.resolvedDomainsStates)
|
||||
}
|
||||
|
||||
@@ -1043,8 +1074,8 @@ func (d *Status) GetFullStatus() FullStatus {
|
||||
LazyConnectionEnabled: d.GetLazyConnection(),
|
||||
}
|
||||
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
|
||||
fullStatus.LocalPeerState = d.localPeer
|
||||
|
||||
@@ -1219,8 +1250,8 @@ func (d *Status) SetWgIface(wgInterface WGIfaceStatus) {
|
||||
}
|
||||
|
||||
func (d *Status) PeersStatus() (*configurer.Stats, error) {
|
||||
d.mux.Lock()
|
||||
defer d.mux.Unlock()
|
||||
d.mux.RLock()
|
||||
defer d.mux.RUnlock()
|
||||
if d.wgIface == nil {
|
||||
return nil, fmt.Errorf("wgInterface is nil, cannot retrieve peers status")
|
||||
}
|
||||
|
||||
@@ -63,6 +63,55 @@ func TestUpdatePeerState(t *testing.T) {
|
||||
assert.Equal(t, ip, state.IP, "ip should be equal")
|
||||
}
|
||||
|
||||
func TestStatus_PeerStateByIP(t *testing.T) {
|
||||
status := NewRecorder("https://mgm")
|
||||
req := require.New(t)
|
||||
|
||||
req.NoError(status.AddPeer("pk-1", "peer-1.netbird", "100.64.0.10", ""))
|
||||
req.NoError(status.AddPeer("pk-2", "peer-2.netbird", "100.64.0.11", ""))
|
||||
|
||||
state, ok := status.PeerStateByIP("100.64.0.10")
|
||||
req.True(ok, "known tunnel IP should resolve to a peer state")
|
||||
req.Equal("pk-1", state.PubKey, "matching state must carry the right pub key")
|
||||
req.Equal("peer-1.netbird", state.FQDN, "matching state must carry the right FQDN")
|
||||
|
||||
_, ok = status.PeerStateByIP("100.64.0.99")
|
||||
req.False(ok, "unknown IP must report ok=false")
|
||||
}
|
||||
|
||||
func TestStatus_PeerStateByIP_MatchesIPv6(t *testing.T) {
|
||||
status := NewRecorder("https://mgm")
|
||||
req := require.New(t)
|
||||
|
||||
req.NoError(status.AddPeer("pk-1", "peer-1.netbird", "100.64.0.10", "fd00::1"))
|
||||
|
||||
state, ok := status.PeerStateByIP("fd00::1")
|
||||
req.True(ok, "IPv6-only match must resolve to the peer state")
|
||||
req.Equal("pk-1", state.PubKey, "matching state must carry the right pub key")
|
||||
}
|
||||
|
||||
// TestStatus_PeerStateByIP_MatchesOfflinePeers covers peers that have
|
||||
// been moved into the offline slice via ReplaceOfflinePeers. Callers
|
||||
// (DNS filter, embed.Client.IdentityForIP) need to treat them as known
|
||||
// rather than unknown — otherwise authentication / DNS filtering treats
|
||||
// known-but-offline peers as foreign IPs.
|
||||
func TestStatus_PeerStateByIP_MatchesOfflinePeers(t *testing.T) {
|
||||
status := NewRecorder("https://mgm")
|
||||
req := require.New(t)
|
||||
|
||||
status.ReplaceOfflinePeers([]State{
|
||||
{PubKey: "pk-offline", FQDN: "offline.netbird", IP: "100.64.0.20", IPv6: "fd00::20"},
|
||||
})
|
||||
|
||||
state, ok := status.PeerStateByIP("100.64.0.20")
|
||||
req.True(ok, "offline peer must resolve by IPv4 tunnel address")
|
||||
req.Equal("pk-offline", state.PubKey, "matching state must carry the offline peer's pub key")
|
||||
|
||||
state, ok = status.PeerStateByIP("fd00::20")
|
||||
req.True(ok, "offline peer must resolve by IPv6 tunnel address")
|
||||
req.Equal("pk-offline", state.PubKey, "IPv6 match must carry the offline peer's pub key")
|
||||
}
|
||||
|
||||
func TestStatus_UpdatePeerFQDN(t *testing.T) {
|
||||
key := "abc"
|
||||
fqdn := "peer-a.netbird.local"
|
||||
|
||||
@@ -179,8 +179,10 @@ func getDefaultGateway() (gateway net.IP, localIP net.IP, err error) {
|
||||
}
|
||||
|
||||
dst := net.IPv4zero
|
||||
if runtime.GOOS == "linux" {
|
||||
// go-netroute v0.4.0 rejects unspecified destinations client-side on Linux.
|
||||
if runtime.GOOS == "linux" || runtime.GOOS == "android" {
|
||||
// go-netroute v0.4.0 rejects unspecified destinations client-side on Linux/Android.
|
||||
// TODO: on android/ios, use platform APIs (ConnectivityManager.getLinkProperties /
|
||||
// NWPathMonitor) when netlink-based lookup is restricted or unavailable.
|
||||
dst = net.IPv4(0, 0, 0, 1)
|
||||
}
|
||||
_, gateway, localIP, err = router.Route(dst)
|
||||
@@ -203,7 +205,7 @@ func getDefaultGateway6() (gateway net.IP, localIP net.IP, err error) {
|
||||
}
|
||||
|
||||
dst := net.IPv6zero
|
||||
if runtime.GOOS == "linux" {
|
||||
if runtime.GOOS == "linux" || runtime.GOOS == "android" {
|
||||
// ::2
|
||||
dst = net.IP{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2}
|
||||
}
|
||||
|
||||
@@ -28,6 +28,15 @@ func hashRosenpassKey(key []byte) string {
|
||||
return hex.EncodeToString(hasher.Sum(nil))
|
||||
}
|
||||
|
||||
// rpServer is the subset of rp.Server used by Manager. Defined as an interface
|
||||
// so tests can substitute a mock without spinning up a real UDP server.
|
||||
type rpServer interface {
|
||||
AddPeer(rp.PeerConfig) (rp.PeerID, error)
|
||||
RemovePeer(rp.PeerID) error
|
||||
Run() error
|
||||
Close() error
|
||||
}
|
||||
|
||||
type Manager struct {
|
||||
ifaceName string
|
||||
spk []byte
|
||||
@@ -36,7 +45,7 @@ type Manager struct {
|
||||
preSharedKey *[32]byte
|
||||
rpPeerIDs map[string]*rp.PeerID
|
||||
rpWgHandler *NetbirdHandler
|
||||
server *rp.Server
|
||||
server rpServer
|
||||
lock sync.Mutex
|
||||
port int
|
||||
wgIface PresharedKeySetter
|
||||
@@ -51,7 +60,22 @@ func NewManager(preSharedKey *wgtypes.Key, wgIfaceName string) (*Manager, error)
|
||||
|
||||
rpKeyHash := hashRosenpassKey(public)
|
||||
log.Tracef("generated new rosenpass key pair with public key %s", rpKeyHash)
|
||||
return &Manager{ifaceName: wgIfaceName, rpKeyHash: rpKeyHash, spk: public, ssk: secret, preSharedKey: (*[32]byte)(preSharedKey), rpPeerIDs: make(map[string]*rp.PeerID), lock: sync.Mutex{}}, nil
|
||||
return &Manager{
|
||||
ifaceName: wgIfaceName,
|
||||
rpKeyHash: rpKeyHash,
|
||||
spk: public,
|
||||
ssk: secret,
|
||||
preSharedKey: (*[32]byte)(preSharedKey),
|
||||
rpPeerIDs: make(map[string]*rp.PeerID),
|
||||
// rpWgHandler is created here (instead of only in generateConfig) so it
|
||||
// is never nil between NewManager and Run(). Otherwise an early
|
||||
// OnConnected call (race observed on Android, issue #4341) panics on
|
||||
// nil receiver in addPeer -> m.rpWgHandler.AddPeer. generateConfig will
|
||||
// replace it with a fresh handler on each Run() to clear stale peer
|
||||
// state from previous engine sessions.
|
||||
rpWgHandler: NewNetbirdHandler(),
|
||||
lock: sync.Mutex{},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (m *Manager) GetPubKey() []byte {
|
||||
@@ -65,6 +89,16 @@ func (m *Manager) GetAddress() *net.UDPAddr {
|
||||
|
||||
// addPeer adds a new peer to the Rosenpass server
|
||||
func (m *Manager) addPeer(rosenpassPubKey []byte, rosenpassAddr string, wireGuardIP string, wireGuardPubKey string) error {
|
||||
// Defense in depth against issue #4341 (Android crash): if Run() has not
|
||||
// completed yet, m.server / m.rpWgHandler may be nil. Return an explicit
|
||||
// error instead of panicking on nil-receiver dereference.
|
||||
if m.server == nil {
|
||||
return fmt.Errorf("rosenpass server not initialized")
|
||||
}
|
||||
if m.rpWgHandler == nil {
|
||||
return fmt.Errorf("rosenpass wg handler not initialized")
|
||||
}
|
||||
|
||||
var err error
|
||||
pcfg := rp.PeerConfig{PublicKey: rosenpassPubKey}
|
||||
if m.preSharedKey != nil {
|
||||
@@ -79,6 +113,16 @@ func (m *Manager) addPeer(rosenpassPubKey []byte, rosenpassAddr string, wireGuar
|
||||
if pcfg.Endpoint, err = net.ResolveUDPAddr("udp", peerAddr); err != nil {
|
||||
return fmt.Errorf("failed to resolve peer endpoint address: %w", err)
|
||||
}
|
||||
// Our local Rosenpass UDP server binds on the IPv6 wildcard ([::]) — see
|
||||
// GetAddress(). The remote peer's endpoint (pcfg.Endpoint) is the destination
|
||||
// our server will sendto when initiating handshakes. ResolveUDPAddr returns a
|
||||
// 4-byte IPv4 for IPv4 hosts, which the kernel rejects (EDESTADDRREQ) when
|
||||
// sent from an AF_INET6 socket. Normalize the remote endpoint to IPv4-mapped
|
||||
// IPv6 so its address family matches our listening socket.
|
||||
// TODO: maybe bind the Rosenpass UDP server to the peer wg IP addr
|
||||
if v4 := pcfg.Endpoint.IP.To4(); v4 != nil {
|
||||
pcfg.Endpoint.IP = v4.To16()
|
||||
}
|
||||
}
|
||||
peerID, err := m.server.AddPeer(pcfg)
|
||||
if err != nil {
|
||||
@@ -182,24 +226,31 @@ func (m *Manager) Run() error {
|
||||
return err
|
||||
}
|
||||
|
||||
m.server, err = rp.NewUDPServer(conf)
|
||||
server, err := rp.NewUDPServer(conf)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
m.lock.Lock()
|
||||
m.server = server
|
||||
m.lock.Unlock()
|
||||
|
||||
log.Infof("starting rosenpass server on port %d", m.port)
|
||||
|
||||
return m.server.Run()
|
||||
return server.Run()
|
||||
}
|
||||
|
||||
// Close closes the Rosenpass server
|
||||
func (m *Manager) Close() error {
|
||||
if m.server != nil {
|
||||
err := m.server.Close()
|
||||
if err != nil {
|
||||
log.Errorf("failed closing local rosenpass server")
|
||||
}
|
||||
m.server = nil
|
||||
m.lock.Lock()
|
||||
server := m.server
|
||||
m.server = nil
|
||||
m.lock.Unlock()
|
||||
if server == nil {
|
||||
return nil
|
||||
}
|
||||
if err := server.Close(); err != nil {
|
||||
log.Errorf("failed closing local rosenpass server: %v", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -1,14 +1,412 @@
|
||||
package rosenpass
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
rp "cunicu.li/go-rosenpass"
|
||||
"github.com/stretchr/testify/require"
|
||||
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
|
||||
)
|
||||
|
||||
// --- test doubles -----------------------------------------------------------
|
||||
|
||||
type addPeerCall struct {
|
||||
cfg rp.PeerConfig
|
||||
}
|
||||
|
||||
type removePeerCall struct {
|
||||
id rp.PeerID
|
||||
}
|
||||
|
||||
type mockServer struct {
|
||||
mu sync.Mutex
|
||||
addCalls []addPeerCall
|
||||
removed []removePeerCall
|
||||
nextID rp.PeerID
|
||||
addErr error
|
||||
removeErr error
|
||||
closed bool
|
||||
ran bool
|
||||
}
|
||||
|
||||
func (m *mockServer) AddPeer(cfg rp.PeerConfig) (rp.PeerID, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.addCalls = append(m.addCalls, addPeerCall{cfg: cfg})
|
||||
if m.addErr != nil {
|
||||
return rp.PeerID{}, m.addErr
|
||||
}
|
||||
// Increment a byte in nextID so distinct peers get distinct IDs.
|
||||
m.nextID[0]++
|
||||
return m.nextID, nil
|
||||
}
|
||||
|
||||
func (m *mockServer) RemovePeer(id rp.PeerID) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.removed = append(m.removed, removePeerCall{id: id})
|
||||
return m.removeErr
|
||||
}
|
||||
|
||||
func (m *mockServer) Run() error { m.ran = true; return nil }
|
||||
func (m *mockServer) Close() error { m.closed = true; return nil }
|
||||
|
||||
type setPSKCall struct {
|
||||
peerKey string
|
||||
psk wgtypes.Key
|
||||
updateOnly bool
|
||||
}
|
||||
|
||||
type mockIface struct {
|
||||
mu sync.Mutex
|
||||
calls []setPSKCall
|
||||
err error
|
||||
}
|
||||
|
||||
func (m *mockIface) SetPresharedKey(peerKey string, psk wgtypes.Key, updateOnly bool) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.calls = append(m.calls, setPSKCall{peerKey: peerKey, psk: psk, updateOnly: updateOnly})
|
||||
return m.err
|
||||
}
|
||||
|
||||
// newTestManager builds a Manager with deterministic spk so tie-break
|
||||
// against a peer pubkey is controllable from tests. The provided spk byte
|
||||
// becomes the first byte; remaining bytes are zero.
|
||||
func newTestManager(spkFirstByte byte, mock *mockServer) *Manager {
|
||||
spk := make([]byte, 32)
|
||||
spk[0] = spkFirstByte
|
||||
return &Manager{
|
||||
ifaceName: "wt0",
|
||||
spk: spk,
|
||||
ssk: make([]byte, 32),
|
||||
rpKeyHash: "test-hash",
|
||||
rpPeerIDs: make(map[string]*rp.PeerID),
|
||||
rpWgHandler: NewNetbirdHandler(),
|
||||
server: mock,
|
||||
}
|
||||
}
|
||||
|
||||
// validWGKey returns a deterministic 32-byte wireguard public key (base64).
|
||||
func validWGKey(t *testing.T, lastByte byte) string {
|
||||
t.Helper()
|
||||
var k wgtypes.Key
|
||||
k[31] = lastByte
|
||||
return k.String()
|
||||
}
|
||||
|
||||
// --- pure helpers ----------------------------------------------------------
|
||||
|
||||
func TestHashRosenpassKey_Deterministic(t *testing.T) {
|
||||
key := []byte("hello-rosenpass")
|
||||
require.Equal(t, hashRosenpassKey(key), hashRosenpassKey(key))
|
||||
require.Len(t, hashRosenpassKey(key), 64) // sha256 hex
|
||||
}
|
||||
|
||||
func TestHashRosenpassKey_DifferentInputsDifferOutputs(t *testing.T) {
|
||||
require.NotEqual(t, hashRosenpassKey([]byte("a")), hashRosenpassKey([]byte("b")))
|
||||
}
|
||||
|
||||
func TestGetLogLevel_DefaultWhenUnset(t *testing.T) {
|
||||
// Snapshot + unset to exercise the LookupEnv ok=false branch. t.Setenv
|
||||
// can only set, not delete, so do it manually with restore via t.Cleanup.
|
||||
prev, hadPrev := os.LookupEnv(defaultLogLevelVar)
|
||||
require.NoError(t, os.Unsetenv(defaultLogLevelVar))
|
||||
t.Cleanup(func() {
|
||||
if hadPrev {
|
||||
_ = os.Setenv(defaultLogLevelVar, prev)
|
||||
} else {
|
||||
_ = os.Unsetenv(defaultLogLevelVar)
|
||||
}
|
||||
})
|
||||
require.Equal(t, defaultLog.String(), getLogLevel().String())
|
||||
}
|
||||
|
||||
func TestGetLogLevel_Cases(t *testing.T) {
|
||||
cases := map[string]string{
|
||||
"debug": "DEBUG",
|
||||
"info": "INFO",
|
||||
"warn": "WARN",
|
||||
"error": "ERROR",
|
||||
"unknown": "INFO", // default fallback
|
||||
}
|
||||
for input, wantStr := range cases {
|
||||
input, wantStr := input, wantStr
|
||||
t.Run(input, func(t *testing.T) {
|
||||
t.Setenv(defaultLogLevelVar, input)
|
||||
require.Equal(t, wantStr, getLogLevel().String())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestFindRandomAvailableUDPPort(t *testing.T) {
|
||||
port, err := findRandomAvailableUDPPort()
|
||||
require.NoError(t, err)
|
||||
require.Greater(t, port, 0)
|
||||
require.LessOrEqual(t, port, 65535)
|
||||
}
|
||||
|
||||
// --- addPeer ---------------------------------------------------------------
|
||||
|
||||
func TestAddPeer_HigherLocalPubkey_SetsEndpoint(t *testing.T) {
|
||||
srv := &mockServer{}
|
||||
m := newTestManager(0xFF, srv) // local spk lexicographically larger
|
||||
|
||||
remotePubKey := make([]byte, 32) // remote spk = all zeros (smaller)
|
||||
err := m.addPeer(remotePubKey, "rosenpass-host:7000", "100.1.1.1", validWGKey(t, 1))
|
||||
require.NoError(t, err)
|
||||
require.Len(t, srv.addCalls, 1)
|
||||
|
||||
ep := srv.addCalls[0].cfg.Endpoint
|
||||
require.NotNil(t, ep, "initiator side must set Endpoint")
|
||||
require.Equal(t, 7000, ep.Port)
|
||||
require.Equal(t, "100.1.1.1", ep.IP.String())
|
||||
}
|
||||
|
||||
func TestAddPeer_HigherLocalPubkey_EndpointIPIsIPv4Mapped(t *testing.T) {
|
||||
// Regression guard for the EDESTADDRREQ fix: Endpoint.IP must be 16-byte
|
||||
// (IPv4-mapped IPv6) so it matches the AF_INET6 listening socket family.
|
||||
srv := &mockServer{}
|
||||
m := newTestManager(0xFF, srv)
|
||||
|
||||
err := m.addPeer(make([]byte, 32), "rp:5000", "100.1.1.1", validWGKey(t, 1))
|
||||
require.NoError(t, err)
|
||||
|
||||
ep := srv.addCalls[0].cfg.Endpoint
|
||||
require.NotNil(t, ep)
|
||||
require.Len(t, ep.IP, 16, "IPv4 endpoint must be normalized to 16-byte v4-mapped form")
|
||||
require.True(t, ep.IP.To4() != nil, "Endpoint must still be detected as IPv4")
|
||||
}
|
||||
|
||||
func TestAddPeer_LowerLocalPubkey_LeavesEndpointNil(t *testing.T) {
|
||||
srv := &mockServer{}
|
||||
m := newTestManager(0x00, srv) // local spk smaller
|
||||
|
||||
remotePubKey := make([]byte, 32)
|
||||
remotePubKey[0] = 0xFF
|
||||
err := m.addPeer(remotePubKey, "rp:5000", "100.1.1.1", validWGKey(t, 2))
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Nil(t, srv.addCalls[0].cfg.Endpoint, "responder side must NOT set Endpoint")
|
||||
}
|
||||
|
||||
func TestAddPeer_PresharedKeyPropagated(t *testing.T) {
|
||||
srv := &mockServer{}
|
||||
psk := &wgtypes.Key{0x42}
|
||||
m := newTestManager(0xFF, srv)
|
||||
m.preSharedKey = (*[32]byte)(psk)
|
||||
|
||||
err := m.addPeer(make([]byte, 32), "rp:5000", "100.1.1.1", validWGKey(t, 3))
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, [32]byte(*psk), [32]byte(srv.addCalls[0].cfg.PresharedKey))
|
||||
}
|
||||
|
||||
func TestAddPeer_InvalidRosenpassAddr_ReturnsError(t *testing.T) {
|
||||
srv := &mockServer{}
|
||||
m := newTestManager(0xFF, srv) // initiator path → parses rosenpassAddr
|
||||
|
||||
err := m.addPeer(make([]byte, 32), "not-a-host-port", "100.1.1.1", validWGKey(t, 1))
|
||||
require.Error(t, err)
|
||||
require.Empty(t, srv.addCalls, "server.AddPeer must not run when address parse fails")
|
||||
}
|
||||
|
||||
func TestAddPeer_InvalidWireGuardPubKey_ReturnsError(t *testing.T) {
|
||||
srv := &mockServer{}
|
||||
m := newTestManager(0xFF, srv)
|
||||
|
||||
err := m.addPeer(make([]byte, 32), "rp:5000", "100.1.1.1", "not-a-valid-key")
|
||||
require.Error(t, err)
|
||||
}
|
||||
|
||||
func TestAddPeer_ServerError_Propagates(t *testing.T) {
|
||||
srv := &mockServer{addErr: errors.New("boom")}
|
||||
m := newTestManager(0xFF, srv)
|
||||
|
||||
err := m.addPeer(make([]byte, 32), "rp:5000", "100.1.1.1", validWGKey(t, 1))
|
||||
require.Error(t, err)
|
||||
}
|
||||
|
||||
// Regression guard for issue #4341 (Android crash). If Run() has not completed
|
||||
// before OnConnected fires, m.rpWgHandler or m.server may be nil. Without the
|
||||
// nil guards, m.rpWgHandler.AddPeer panics on nil receiver.
|
||||
func TestAddPeer_NilHandler_ReturnsErrorNoCrash(t *testing.T) {
|
||||
srv := &mockServer{}
|
||||
m := newTestManager(0xFF, srv)
|
||||
m.rpWgHandler = nil // simulate Run() not yet completed
|
||||
|
||||
err := m.addPeer(make([]byte, 32), "rp:5000", "100.1.1.1", validWGKey(t, 1))
|
||||
require.Error(t, err)
|
||||
require.Contains(t, err.Error(), "wg handler not initialized")
|
||||
}
|
||||
|
||||
func TestAddPeer_NilServer_ReturnsErrorNoCrash(t *testing.T) {
|
||||
m := newTestManager(0xFF, nil)
|
||||
m.server = nil // simulate Run() not yet completed
|
||||
|
||||
err := m.addPeer(make([]byte, 32), "rp:5000", "100.1.1.1", validWGKey(t, 1))
|
||||
require.Error(t, err)
|
||||
require.Contains(t, err.Error(), "server not initialized")
|
||||
}
|
||||
|
||||
// NewManager must pre-initialize rpWgHandler so the nil-receiver crash from
|
||||
// issue #4341 cannot occur in the window between NewManager and Run().
|
||||
func TestNewManager_PreInitializesHandler(t *testing.T) {
|
||||
psk := wgtypes.Key{}
|
||||
m, err := NewManager(&psk, "wt0")
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, m.rpWgHandler, "rpWgHandler must be initialized in NewManager")
|
||||
}
|
||||
|
||||
func TestAddPeer_RecordsPeerID(t *testing.T) {
|
||||
srv := &mockServer{}
|
||||
m := newTestManager(0xFF, srv)
|
||||
|
||||
wgKey := validWGKey(t, 5)
|
||||
err := m.addPeer(make([]byte, 32), "rp:5000", "100.1.1.1", wgKey)
|
||||
require.NoError(t, err)
|
||||
require.Contains(t, m.rpPeerIDs, wgKey)
|
||||
}
|
||||
|
||||
// --- OnConnected / OnDisconnected ------------------------------------------
|
||||
|
||||
func TestOnConnected_NilRemotePubKey_NoAddPeer(t *testing.T) {
|
||||
srv := &mockServer{}
|
||||
m := newTestManager(0xFF, srv)
|
||||
|
||||
m.OnConnected(validWGKey(t, 1), nil, "100.1.1.1", "rp:5000")
|
||||
require.Empty(t, srv.addCalls, "nil remote rosenpass pubkey must skip AddPeer")
|
||||
require.Empty(t, m.rpPeerIDs)
|
||||
}
|
||||
|
||||
func TestOnConnected_ValidPubKey_CallsAddPeer(t *testing.T) {
|
||||
srv := &mockServer{}
|
||||
m := newTestManager(0xFF, srv)
|
||||
|
||||
wgKey := validWGKey(t, 1)
|
||||
m.OnConnected(wgKey, make([]byte, 32), "100.1.1.1", "rp:5000")
|
||||
require.Len(t, srv.addCalls, 1)
|
||||
require.Contains(t, m.rpPeerIDs, wgKey)
|
||||
}
|
||||
|
||||
func TestOnDisconnected_UnknownPeer_NoOp(t *testing.T) {
|
||||
srv := &mockServer{}
|
||||
m := newTestManager(0xFF, srv)
|
||||
|
||||
m.OnDisconnected(validWGKey(t, 99))
|
||||
require.Empty(t, srv.removed, "unknown peer key must not call RemovePeer")
|
||||
}
|
||||
|
||||
func TestOnDisconnected_KnownPeer_CallsRemoveAndForgets(t *testing.T) {
|
||||
srv := &mockServer{}
|
||||
m := newTestManager(0xFF, srv)
|
||||
|
||||
wgKey := validWGKey(t, 1)
|
||||
require.NoError(t, m.addPeer(make([]byte, 32), "rp:5000", "100.1.1.1", wgKey))
|
||||
require.Contains(t, m.rpPeerIDs, wgKey)
|
||||
|
||||
m.OnDisconnected(wgKey)
|
||||
require.Len(t, srv.removed, 1)
|
||||
require.NotContains(t, m.rpPeerIDs, wgKey, "peer must be forgotten after disconnect")
|
||||
}
|
||||
|
||||
// --- IsPresharedKeyInitialized ---------------------------------------------
|
||||
|
||||
func TestIsPresharedKeyInitialized_UnknownPeer_ReturnsFalse(t *testing.T) {
|
||||
srv := &mockServer{}
|
||||
m := newTestManager(0xFF, srv)
|
||||
require.False(t, m.IsPresharedKeyInitialized(validWGKey(t, 1)))
|
||||
}
|
||||
|
||||
func TestIsPresharedKeyInitialized_AddedButNotHandshaken_ReturnsFalse(t *testing.T) {
|
||||
srv := &mockServer{}
|
||||
m := newTestManager(0xFF, srv)
|
||||
|
||||
wgKey := validWGKey(t, 2)
|
||||
require.NoError(t, m.addPeer(make([]byte, 32), "rp:5000", "100.1.1.1", wgKey))
|
||||
require.False(t, m.IsPresharedKeyInitialized(wgKey))
|
||||
}
|
||||
|
||||
// --- NetbirdHandler.outputKey ----------------------------------------------
|
||||
|
||||
func TestHandler_OutputKey_FirstCallUsesUpdateOnlyFalse(t *testing.T) {
|
||||
h := NewNetbirdHandler()
|
||||
iface := &mockIface{}
|
||||
h.SetInterface(iface)
|
||||
|
||||
pid := rp.PeerID{0x01}
|
||||
wgKey := wgtypes.Key{0xAA}
|
||||
h.AddPeer(pid, "wt0", rp.Key(wgKey))
|
||||
|
||||
psk := rp.Key{0xBB}
|
||||
h.HandshakeCompleted(pid, psk)
|
||||
|
||||
require.Len(t, iface.calls, 1)
|
||||
require.False(t, iface.calls[0].updateOnly, "first PSK rotation must use updateOnly=false")
|
||||
require.Equal(t, wgKey.String(), iface.calls[0].peerKey)
|
||||
}
|
||||
|
||||
func TestHandler_OutputKey_SubsequentCallsUseUpdateOnlyTrue(t *testing.T) {
|
||||
h := NewNetbirdHandler()
|
||||
iface := &mockIface{}
|
||||
h.SetInterface(iface)
|
||||
|
||||
pid := rp.PeerID{0x02}
|
||||
h.AddPeer(pid, "wt0", rp.Key(wgtypes.Key{0xCC}))
|
||||
|
||||
h.HandshakeCompleted(pid, rp.Key{0x01}) // first
|
||||
h.HandshakeCompleted(pid, rp.Key{0x02}) // second
|
||||
|
||||
require.Len(t, iface.calls, 2)
|
||||
require.False(t, iface.calls[0].updateOnly)
|
||||
require.True(t, iface.calls[1].updateOnly, "subsequent rotations must use updateOnly=true")
|
||||
}
|
||||
|
||||
func TestHandler_OutputKey_NilInterface_NoCrashNoCall(t *testing.T) {
|
||||
h := NewNetbirdHandler()
|
||||
// no SetInterface — iface remains nil
|
||||
pid := rp.PeerID{0x03}
|
||||
h.AddPeer(pid, "wt0", rp.Key(wgtypes.Key{}))
|
||||
|
||||
// Must not panic.
|
||||
h.HandshakeCompleted(pid, rp.Key{})
|
||||
}
|
||||
|
||||
func TestHandler_OutputKey_UnknownPeer_NoCall(t *testing.T) {
|
||||
h := NewNetbirdHandler()
|
||||
iface := &mockIface{}
|
||||
h.SetInterface(iface)
|
||||
|
||||
h.HandshakeCompleted(rp.PeerID{0xFF}, rp.Key{})
|
||||
require.Empty(t, iface.calls, "unknown peer id must not trigger SetPresharedKey")
|
||||
}
|
||||
|
||||
func TestHandler_RemovePeer_ClearsInitializedState(t *testing.T) {
|
||||
h := NewNetbirdHandler()
|
||||
iface := &mockIface{}
|
||||
h.SetInterface(iface)
|
||||
|
||||
pid := rp.PeerID{0x04}
|
||||
h.AddPeer(pid, "wt0", rp.Key(wgtypes.Key{0xDD}))
|
||||
h.HandshakeCompleted(pid, rp.Key{0x01})
|
||||
require.True(t, h.IsPeerInitialized(pid))
|
||||
|
||||
h.RemovePeer(pid)
|
||||
require.False(t, h.IsPeerInitialized(pid), "RemovePeer must clear initialized flag")
|
||||
}
|
||||
|
||||
func TestHandler_SetInterfaceAfterAddPeer_StillReceivesKey(t *testing.T) {
|
||||
h := NewNetbirdHandler()
|
||||
pid := rp.PeerID{0x05}
|
||||
wgKey := wgtypes.Key{0xEE}
|
||||
h.AddPeer(pid, "wt0", rp.Key(wgKey))
|
||||
|
||||
iface := &mockIface{}
|
||||
h.SetInterface(iface) // set after AddPeer
|
||||
|
||||
h.HandshakeCompleted(pid, rp.Key{0x42})
|
||||
require.Len(t, iface.calls, 1)
|
||||
require.Equal(t, wgKey.String(), iface.calls[0].peerKey)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
package rosenpass
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
|
||||
)
|
||||
|
||||
// DeterministicSeedKey derives a 32-byte WireGuard preshared key from a pair
|
||||
// of peer public keys. Both peers, given the same key pair, produce the same
|
||||
// output regardless of which side runs the function: the inputs are ordered
|
||||
// lexicographically before concatenation.
|
||||
//
|
||||
// NetBird uses this value as the initial Rosenpass-side preshared key when no
|
||||
// explicit account-level PSK is configured, so both peers converge on the same
|
||||
// PSK before the first post-quantum handshake completes.
|
||||
//
|
||||
// The resulting key MUST NOT be treated as quantum-safe: it is deterministic
|
||||
// from public keys and exists only to seed WireGuard until Rosenpass rotates
|
||||
// in a real post-quantum PSK.
|
||||
func DeterministicSeedKey(localKey, remoteKey string) (*wgtypes.Key, error) {
|
||||
lk := []byte(localKey)
|
||||
rk := []byte(remoteKey)
|
||||
if len(lk) < 16 || len(rk) < 16 {
|
||||
return nil, fmt.Errorf("rosenpass: peer keys must be at least 16 bytes (got local=%d, remote=%d)", len(lk), len(rk))
|
||||
}
|
||||
|
||||
var keyInput []byte
|
||||
if localKey > remoteKey {
|
||||
keyInput = append(keyInput, lk[:16]...)
|
||||
keyInput = append(keyInput, rk[:16]...)
|
||||
} else {
|
||||
keyInput = append(keyInput, rk[:16]...)
|
||||
keyInput = append(keyInput, lk[:16]...)
|
||||
}
|
||||
|
||||
key, err := wgtypes.NewKey(keyInput)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("rosenpass: deterministic seed key: %w", err)
|
||||
}
|
||||
return &key, nil
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
package rosenpass
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestDeterministicSeedKey_SameForBothSides(t *testing.T) {
|
||||
// Peer A and peer B must derive the same PSK regardless of which side
|
||||
// computes it: the function orders inputs internally.
|
||||
a := strings.Repeat("a", 32)
|
||||
b := strings.Repeat("b", 32)
|
||||
|
||||
keyAB, err := DeterministicSeedKey(a, b)
|
||||
require.NoError(t, err)
|
||||
keyBA, err := DeterministicSeedKey(b, a)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, keyAB.String(), keyBA.String(), "swapping arguments must yield identical key")
|
||||
}
|
||||
|
||||
func TestDeterministicSeedKey_ChangesWithKeys(t *testing.T) {
|
||||
a := strings.Repeat("a", 32)
|
||||
b := strings.Repeat("b", 32)
|
||||
c := strings.Repeat("c", 32)
|
||||
|
||||
keyAB, err := DeterministicSeedKey(a, b)
|
||||
require.NoError(t, err)
|
||||
keyAC, err := DeterministicSeedKey(a, c)
|
||||
require.NoError(t, err)
|
||||
require.NotEqual(t, keyAB.String(), keyAC.String(), "different peer pair must yield different key")
|
||||
}
|
||||
|
||||
func TestDeterministicSeedKey_TooShortKey_ReturnsError(t *testing.T) {
|
||||
short := "short" // < 16 bytes
|
||||
long := strings.Repeat("x", 32)
|
||||
|
||||
_, err := DeterministicSeedKey(short, long)
|
||||
require.Error(t, err)
|
||||
_, err = DeterministicSeedKey(long, short)
|
||||
require.Error(t, err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
//go:build dragonfly || freebsd || netbsd || openbsd
|
||||
|
||||
package systemops
|
||||
|
||||
// IgnoreAddedDefaultRoute reports whether an RTM_ADD default route with the
|
||||
// given flags should be ignored by the network monitor.
|
||||
func IgnoreAddedDefaultRoute(flags int) bool {
|
||||
return filterRoutesByFlags(flags)
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
//go:build darwin
|
||||
|
||||
package systemops
|
||||
|
||||
import "golang.org/x/sys/unix"
|
||||
|
||||
// IgnoreAddedDefaultRoute reports whether an RTM_ADD default route with the
|
||||
// given flags should be ignored by the network monitor. Scoped routes
|
||||
// (RTF_IFSCOPE) are tied to a specific interface index and cannot replace the
|
||||
// unscoped default the kernel uses for general egress, so flapping ones (e.g.
|
||||
// Wi-Fi calling IMS tunnels on ipsec0, Docker bridges, scoped utun defaults)
|
||||
// must not trigger an engine restart.
|
||||
func IgnoreAddedDefaultRoute(flags int) bool {
|
||||
if filterRoutesByFlags(flags) {
|
||||
return true
|
||||
}
|
||||
if flags&unix.RTF_IFSCOPE != 0 {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -96,17 +96,19 @@ func (m *Manager) Stop(ctx context.Context) error {
|
||||
}
|
||||
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
cancel := m.cancel
|
||||
done := m.done
|
||||
m.mu.Unlock()
|
||||
|
||||
if m.cancel == nil {
|
||||
if cancel == nil {
|
||||
return nil
|
||||
}
|
||||
m.cancel()
|
||||
cancel()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-m.done:
|
||||
case <-done:
|
||||
}
|
||||
|
||||
return nil
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
package syncstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
"google.golang.org/protobuf/proto"
|
||||
|
||||
mgmProto "github.com/netbirdio/netbird/shared/management/proto"
|
||||
"github.com/netbirdio/netbird/util"
|
||||
)
|
||||
|
||||
// syncResponseFileName is the name of the file the sync response is serialized
|
||||
// to, placed inside the configured directory (the state directory).
|
||||
const syncResponseFileName = "networkmap.pb"
|
||||
|
||||
// diskStore serializes the latest sync response to a file on disk instead of
|
||||
// keeping it in memory. This trades disk I/O for a much smaller memory
|
||||
// footprint, which matters on memory-constrained platforms (iOS).
|
||||
type diskStore struct {
|
||||
mu sync.Mutex
|
||||
path string
|
||||
}
|
||||
|
||||
// NewDiskStore returns a Store that serializes the sync response to a file in
|
||||
// the given directory. If dir is empty it falls back to the OS temp directory.
|
||||
//
|
||||
// Any file left over from a previous run is removed on construction so a fresh
|
||||
// store never reads stale data (e.g. another profile's network map).
|
||||
func NewDiskStore(dir string) Store {
|
||||
if dir == "" {
|
||||
dir = os.TempDir()
|
||||
}
|
||||
s := &diskStore{
|
||||
path: filepath.Join(dir, syncResponseFileName),
|
||||
}
|
||||
if err := s.Clear(); err != nil {
|
||||
log.Warnf("failed to clear stale sync response file: %v", err)
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *diskStore) Set(resp *mgmProto.SyncResponse) error {
|
||||
if resp == nil {
|
||||
return s.Clear()
|
||||
}
|
||||
|
||||
bs, err := proto.Marshal(resp)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal sync response: %w", err)
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
if err := util.WriteBytesWithRestrictedPermission(context.Background(), s.path, bs); err != nil {
|
||||
return fmt.Errorf("write sync response to %s: %w", s.path, err)
|
||||
}
|
||||
|
||||
log.Debugf("sync response persisted to %s (%d bytes)", s.path, len(bs))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *diskStore) Get() (*mgmProto.SyncResponse, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
bs, err := os.ReadFile(s.path)
|
||||
if err != nil {
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
//nolint:nilnil // nil,nil means "nothing stored", per the Store contract; preserve the original behaviour
|
||||
return nil, nil
|
||||
}
|
||||
return nil, fmt.Errorf("read sync response from %s: %w", s.path, err)
|
||||
}
|
||||
|
||||
resp := &mgmProto.SyncResponse{}
|
||||
if err := proto.Unmarshal(bs, resp); err != nil {
|
||||
return nil, fmt.Errorf("unmarshal sync response: %w", err)
|
||||
}
|
||||
|
||||
log.Debugf("retrieving latest sync response from %s (%d bytes)", s.path, len(bs))
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (s *diskStore) Clear() error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
if err := os.Remove(s.path); err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||
return fmt.Errorf("remove sync response file %s: %w", s.path, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
//go:build ios
|
||||
|
||||
package syncstore
|
||||
|
||||
// New returns the platform default store. On iOS the sync response is
|
||||
// serialized to disk (in dir) to keep it out of the constrained process memory.
|
||||
func New(dir string) Store {
|
||||
return NewDiskStore(dir)
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
//go:build !ios
|
||||
|
||||
package syncstore
|
||||
|
||||
// New returns the platform default store. On all non-iOS platforms the sync
|
||||
// response is kept in memory; dir is unused.
|
||||
func New(_ string) Store {
|
||||
return NewMemoryStore()
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
package syncstore
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
"google.golang.org/protobuf/proto"
|
||||
|
||||
mgmProto "github.com/netbirdio/netbird/shared/management/proto"
|
||||
)
|
||||
|
||||
// memoryStore keeps the latest sync response in memory.
|
||||
type memoryStore struct {
|
||||
mu sync.RWMutex
|
||||
latest *mgmProto.SyncResponse
|
||||
}
|
||||
|
||||
// NewMemoryStore returns a Store that keeps the sync response in memory.
|
||||
func NewMemoryStore() Store {
|
||||
return &memoryStore{}
|
||||
}
|
||||
|
||||
func (s *memoryStore) Set(resp *mgmProto.SyncResponse) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
s.latest = resp
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *memoryStore) Get() (*mgmProto.SyncResponse, error) {
|
||||
s.mu.RLock()
|
||||
latest := s.latest
|
||||
s.mu.RUnlock()
|
||||
|
||||
if latest == nil {
|
||||
//nolint:nilnil // nil,nil means "nothing stored", per the Store contract; preserve the original behaviour
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
log.Debugf("retrieving latest sync response with size %d bytes", proto.Size(latest))
|
||||
sr, ok := proto.Clone(latest).(*mgmProto.SyncResponse)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("clone sync response")
|
||||
}
|
||||
return sr, nil
|
||||
}
|
||||
|
||||
func (s *memoryStore) Clear() error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
s.latest = nil
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
// Package syncstore stores the latest Management sync response (which carries
|
||||
// the network map) for debug bundle generation.
|
||||
//
|
||||
// The storage backend is selected at build time per operating system: on iOS
|
||||
// the response is serialized to disk to keep it out of the (tightly
|
||||
// constrained) process memory, while on all other platforms it is kept in
|
||||
// memory. The backend is chosen by the New constructor; see factory_ios.go and
|
||||
// factory_other.go.
|
||||
package syncstore
|
||||
|
||||
import (
|
||||
mgmProto "github.com/netbirdio/netbird/shared/management/proto"
|
||||
)
|
||||
|
||||
// Store persists the latest sync response and returns it on demand.
|
||||
//
|
||||
// Implementations must be safe for concurrent use.
|
||||
type Store interface {
|
||||
// Set stores the given sync response, replacing any previously stored one.
|
||||
Set(resp *mgmProto.SyncResponse) error
|
||||
|
||||
// Get returns the stored sync response, or nil if none is stored.
|
||||
// The returned value is an independent copy that the caller may retain.
|
||||
Get() (*mgmProto.SyncResponse, error)
|
||||
|
||||
// Clear removes any stored sync response. It is safe to call when nothing
|
||||
// is stored.
|
||||
Clear() error
|
||||
}
|
||||
@@ -19,8 +19,6 @@ import (
|
||||
|
||||
const (
|
||||
latestVersion = "latest"
|
||||
// this version will be ignored
|
||||
developmentVersion = "development"
|
||||
)
|
||||
|
||||
var errNoUpdateState = errors.New("no update state found")
|
||||
@@ -483,7 +481,7 @@ func (m *Manager) loadAndDeleteUpdateState(ctx context.Context) (*UpdateState, e
|
||||
}
|
||||
|
||||
func (m *Manager) shouldUpdate(updateVersion *v.Version, forceUpdate bool) bool {
|
||||
if m.currentVersion == developmentVersion {
|
||||
if version.IsDevelopmentVersion(m.currentVersion) {
|
||||
log.Debugf("skipping auto-update, running development version")
|
||||
return false
|
||||
}
|
||||
|
||||
@@ -13,7 +13,7 @@ import (
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.opentelemetry.io/otel"
|
||||
|
||||
"github.com/netbirdio/management-integrations/integrations"
|
||||
"github.com/netbirdio/netbird/management/server/integrations/integrated_validator/validator"
|
||||
|
||||
"github.com/netbirdio/netbird/management/internals/controllers/network_map/controller"
|
||||
"github.com/netbirdio/netbird/management/internals/controllers/network_map/update_channel"
|
||||
@@ -315,7 +315,7 @@ func startManagement(t *testing.T, signalAddr string, counter *int) (*grpc.Serve
|
||||
return nil, "", err
|
||||
}
|
||||
|
||||
ia, _ := integrations.NewIntegratedValidator(context.Background(), peersManager, settingsManagerMock, eventStore, cacheStore)
|
||||
ia, _ := validator.NewIntegratedValidator(context.Background(), peersManager, settingsManagerMock, eventStore, cacheStore)
|
||||
|
||||
metrics, err := telemetry.NewDefaultAppMetrics(context.Background())
|
||||
require.NoError(t, err)
|
||||
|
||||
+12
-27
@@ -3,15 +3,14 @@
|
||||
package system
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"os"
|
||||
"os/exec"
|
||||
"regexp"
|
||||
"runtime"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"golang.org/x/sys/unix"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
"github.com/zcalusic/sysinfo"
|
||||
|
||||
@@ -29,19 +28,11 @@ func UpdateStaticInfoAsync() {
|
||||
|
||||
// GetInfo retrieves and parses the system information
|
||||
func GetInfo(ctx context.Context) *Info {
|
||||
info := _getInfo()
|
||||
for strings.Contains(info, "broken pipe") {
|
||||
info = _getInfo()
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
}
|
||||
|
||||
osStr := strings.ReplaceAll(info, "\n", "")
|
||||
osStr = strings.ReplaceAll(osStr, "\r\n", "")
|
||||
osInfo := strings.Split(osStr, " ")
|
||||
kernelName, kernelVersion, kernelPlatform := kernelInfo()
|
||||
|
||||
osName, osVersion := readOsReleaseFile()
|
||||
if osName == "" {
|
||||
osName = osInfo[3]
|
||||
osName = kernelName
|
||||
}
|
||||
|
||||
systemHostname, _ := os.Hostname()
|
||||
@@ -58,8 +49,8 @@ func GetInfo(ctx context.Context) *Info {
|
||||
}
|
||||
|
||||
gio := &Info{
|
||||
Kernel: osInfo[0],
|
||||
Platform: osInfo[2],
|
||||
Kernel: kernelName,
|
||||
Platform: kernelPlatform,
|
||||
OS: osName,
|
||||
OSVersion: osVersion,
|
||||
Hostname: extractDeviceName(ctx, systemHostname),
|
||||
@@ -67,7 +58,7 @@ func GetInfo(ctx context.Context) *Info {
|
||||
CPUs: runtime.NumCPU(),
|
||||
NetbirdVersion: version.NetbirdVersion(),
|
||||
UIVersion: extractUserAgent(ctx),
|
||||
KernelVersion: osInfo[1],
|
||||
KernelVersion: kernelVersion,
|
||||
NetworkAddresses: addrs,
|
||||
SystemSerialNumber: si.SystemSerialNumber,
|
||||
SystemProductName: si.SystemProductName,
|
||||
@@ -78,18 +69,12 @@ func GetInfo(ctx context.Context) *Info {
|
||||
return gio
|
||||
}
|
||||
|
||||
func _getInfo() string {
|
||||
cmd := exec.Command("uname", "-srio")
|
||||
cmd.Stdin = strings.NewReader("some")
|
||||
var out bytes.Buffer
|
||||
var stderr bytes.Buffer
|
||||
cmd.Stdout = &out
|
||||
cmd.Stderr = &stderr
|
||||
err := cmd.Run()
|
||||
if err != nil {
|
||||
log.Warnf("getInfo: %s", err)
|
||||
func kernelInfo() (string, string, string) {
|
||||
var uts unix.Utsname
|
||||
if err := unix.Uname(&uts); err != nil {
|
||||
return "", "", ""
|
||||
}
|
||||
return out.String()
|
||||
return unix.ByteSliceToString(uts.Sysname[:]), unix.ByteSliceToString(uts.Release[:]), unix.ByteSliceToString(uts.Machine[:])
|
||||
}
|
||||
|
||||
func sysInfo() (string, string, string) {
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"fmt"
|
||||
"sync"
|
||||
"syscall/js"
|
||||
"time"
|
||||
|
||||
@@ -13,7 +14,7 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
certValidationTimeout = 60 * time.Second
|
||||
certValidationTimeout = 5 * time.Minute
|
||||
)
|
||||
|
||||
func (p *RDCleanPathProxy) validateCertificateWithJS(conn *proxyConnection, certChain [][]byte) (bool, error) {
|
||||
@@ -46,17 +47,31 @@ func (p *RDCleanPathProxy) validateCertificateWithJS(conn *proxyConnection, cert
|
||||
|
||||
promise := conn.wsHandlers.Call("onCertificateRequest", certInfo)
|
||||
|
||||
resultChan := make(chan bool)
|
||||
errorChan := make(chan error)
|
||||
resultChan := make(chan bool, 1)
|
||||
errorChan := make(chan error, 1)
|
||||
|
||||
promise.Call("then", js.FuncOf(func(this js.Value, args []js.Value) interface{} {
|
||||
result := args[0].Bool()
|
||||
resultChan <- result
|
||||
// Release from inside the callbacks so a post-timeout promise resolution
|
||||
// does not invoke an already-released func.
|
||||
var thenFn, catchFn js.Func
|
||||
var releaseOnce sync.Once
|
||||
release := func() {
|
||||
releaseOnce.Do(func() {
|
||||
thenFn.Release()
|
||||
catchFn.Release()
|
||||
})
|
||||
}
|
||||
thenFn = js.FuncOf(func(this js.Value, args []js.Value) interface{} {
|
||||
defer release()
|
||||
resultChan <- args[0].Bool()
|
||||
return nil
|
||||
})).Call("catch", js.FuncOf(func(this js.Value, args []js.Value) interface{} {
|
||||
})
|
||||
catchFn = js.FuncOf(func(this js.Value, args []js.Value) interface{} {
|
||||
defer release()
|
||||
errorChan <- fmt.Errorf("certificate validation failed")
|
||||
return nil
|
||||
}))
|
||||
})
|
||||
|
||||
promise.Call("then", thenFn).Call("catch", catchFn)
|
||||
|
||||
select {
|
||||
case result := <-resultChan:
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"io"
|
||||
"net"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"syscall/js"
|
||||
"time"
|
||||
|
||||
@@ -57,6 +58,8 @@ type RDCleanPathProxy struct {
|
||||
}
|
||||
activeConnections map[string]*proxyConnection
|
||||
destinations map[string]string
|
||||
pendingHandlers map[string]js.Func
|
||||
nextID atomic.Uint64
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
@@ -66,8 +69,15 @@ type proxyConnection struct {
|
||||
rdpConn net.Conn
|
||||
tlsConn *tls.Conn
|
||||
wsHandlers js.Value
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
// Go-side callbacks exposed to JS. js.FuncOf pins the Go closure in a
|
||||
// global handle map and MUST be released, otherwise every connection
|
||||
// leaks the Go memory the closure captures.
|
||||
wsHandlerFn js.Func
|
||||
onMessageFn js.Func
|
||||
onCloseFn js.Func
|
||||
cleanupOnce sync.Once
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
// NewRDCleanPathProxy creates a new RDCleanPath proxy
|
||||
@@ -80,7 +90,11 @@ func NewRDCleanPathProxy(client interface {
|
||||
}
|
||||
}
|
||||
|
||||
// CreateProxy creates a new proxy endpoint for the given destination
|
||||
// CreateProxy creates a new proxy endpoint for the given destination.
|
||||
// The registered handler fn and its destinations/pendingHandlers entries are
|
||||
// only released once a connection is established and cleanupConnection runs.
|
||||
// If a caller invokes CreateProxy but never connects to the returned URL,
|
||||
// those entries stay pinned for the lifetime of the page.
|
||||
func (p *RDCleanPathProxy) CreateProxy(hostname, port string) js.Value {
|
||||
destination := net.JoinHostPort(hostname, port)
|
||||
|
||||
@@ -88,7 +102,7 @@ func (p *RDCleanPathProxy) CreateProxy(hostname, port string) js.Value {
|
||||
resolve := args[0]
|
||||
|
||||
go func() {
|
||||
proxyID := fmt.Sprintf("proxy_%d", len(p.activeConnections))
|
||||
proxyID := fmt.Sprintf("proxy_%d", p.nextID.Add(1))
|
||||
|
||||
p.mu.Lock()
|
||||
if p.destinations == nil {
|
||||
@@ -100,7 +114,7 @@ func (p *RDCleanPathProxy) CreateProxy(hostname, port string) js.Value {
|
||||
proxyURL := fmt.Sprintf("%s://%s/%s", RDCleanPathProxyScheme, RDCleanPathProxyHost, proxyID)
|
||||
|
||||
// Register the WebSocket handler for this specific proxy
|
||||
js.Global().Set(fmt.Sprintf("handleRDCleanPathWebSocket_%s", proxyID), js.FuncOf(func(_ js.Value, args []js.Value) any {
|
||||
handlerFn := js.FuncOf(func(_ js.Value, args []js.Value) any {
|
||||
if len(args) < 1 {
|
||||
return js.ValueOf("error: requires WebSocket argument")
|
||||
}
|
||||
@@ -108,7 +122,14 @@ func (p *RDCleanPathProxy) CreateProxy(hostname, port string) js.Value {
|
||||
ws := args[0]
|
||||
p.HandleWebSocketConnection(ws, proxyID)
|
||||
return nil
|
||||
}))
|
||||
})
|
||||
p.mu.Lock()
|
||||
if p.pendingHandlers == nil {
|
||||
p.pendingHandlers = make(map[string]js.Func)
|
||||
}
|
||||
p.pendingHandlers[proxyID] = handlerFn
|
||||
p.mu.Unlock()
|
||||
js.Global().Set(fmt.Sprintf("handleRDCleanPathWebSocket_%s", proxyID), handlerFn)
|
||||
|
||||
log.Infof("Created RDCleanPath proxy endpoint: %s for destination: %s", proxyURL, destination)
|
||||
resolve.Invoke(proxyURL)
|
||||
@@ -142,6 +163,10 @@ func (p *RDCleanPathProxy) HandleWebSocketConnection(ws js.Value, proxyID string
|
||||
|
||||
p.mu.Lock()
|
||||
p.activeConnections[proxyID] = conn
|
||||
if fn, ok := p.pendingHandlers[proxyID]; ok {
|
||||
conn.wsHandlerFn = fn
|
||||
delete(p.pendingHandlers, proxyID)
|
||||
}
|
||||
p.mu.Unlock()
|
||||
|
||||
p.setupWebSocketHandlers(ws, conn)
|
||||
@@ -150,7 +175,7 @@ func (p *RDCleanPathProxy) HandleWebSocketConnection(ws js.Value, proxyID string
|
||||
}
|
||||
|
||||
func (p *RDCleanPathProxy) setupWebSocketHandlers(ws js.Value, conn *proxyConnection) {
|
||||
ws.Set("onGoMessage", js.FuncOf(func(this js.Value, args []js.Value) any {
|
||||
conn.onMessageFn = js.FuncOf(func(this js.Value, args []js.Value) any {
|
||||
if len(args) < 1 {
|
||||
return nil
|
||||
}
|
||||
@@ -158,13 +183,15 @@ func (p *RDCleanPathProxy) setupWebSocketHandlers(ws js.Value, conn *proxyConnec
|
||||
data := args[0]
|
||||
go p.handleWebSocketMessage(conn, data)
|
||||
return nil
|
||||
}))
|
||||
})
|
||||
ws.Set("onGoMessage", conn.onMessageFn)
|
||||
|
||||
ws.Set("onGoClose", js.FuncOf(func(_ js.Value, args []js.Value) any {
|
||||
conn.onCloseFn = js.FuncOf(func(_ js.Value, args []js.Value) any {
|
||||
log.Debug("WebSocket closed by JavaScript")
|
||||
conn.cancel()
|
||||
return nil
|
||||
}))
|
||||
})
|
||||
ws.Set("onGoClose", conn.onCloseFn)
|
||||
}
|
||||
|
||||
func (p *RDCleanPathProxy) handleWebSocketMessage(conn *proxyConnection, data js.Value) {
|
||||
@@ -261,25 +288,49 @@ func (p *RDCleanPathProxy) handleDirectRDP(conn *proxyConnection, firstPacket []
|
||||
}
|
||||
|
||||
func (p *RDCleanPathProxy) cleanupConnection(conn *proxyConnection) {
|
||||
log.Debugf("Cleaning up connection %s", conn.id)
|
||||
conn.cancel()
|
||||
if conn.tlsConn != nil {
|
||||
log.Debug("Closing TLS connection")
|
||||
if err := conn.tlsConn.Close(); err != nil {
|
||||
log.Debugf("Error closing TLS connection: %v", err)
|
||||
conn.cleanupOnce.Do(func() {
|
||||
log.Debugf("Cleaning up connection %s", conn.id)
|
||||
conn.cancel()
|
||||
if conn.tlsConn != nil {
|
||||
log.Debug("Closing TLS connection")
|
||||
if err := conn.tlsConn.Close(); err != nil {
|
||||
log.Debugf("Error closing TLS connection: %v", err)
|
||||
}
|
||||
conn.tlsConn = nil
|
||||
}
|
||||
conn.tlsConn = nil
|
||||
}
|
||||
if conn.rdpConn != nil {
|
||||
log.Debug("Closing TCP connection")
|
||||
if err := conn.rdpConn.Close(); err != nil {
|
||||
log.Debugf("Error closing TCP connection: %v", err)
|
||||
if conn.rdpConn != nil {
|
||||
log.Debug("Closing TCP connection")
|
||||
if err := conn.rdpConn.Close(); err != nil {
|
||||
log.Debugf("Error closing TCP connection: %v", err)
|
||||
}
|
||||
conn.rdpConn = nil
|
||||
}
|
||||
conn.rdpConn = nil
|
||||
}
|
||||
p.mu.Lock()
|
||||
delete(p.activeConnections, conn.id)
|
||||
p.mu.Unlock()
|
||||
js.Global().Delete(fmt.Sprintf("handleRDCleanPathWebSocket_%s", conn.id))
|
||||
|
||||
// Detach before releasing so late JS calls surface as TypeError instead
|
||||
// of silent "call to released function".
|
||||
if conn.wsHandlers.Truthy() {
|
||||
conn.wsHandlers.Set("onGoMessage", js.Undefined())
|
||||
conn.wsHandlers.Set("onGoClose", js.Undefined())
|
||||
}
|
||||
|
||||
// wsHandlerFn may be zero-value if the pending handler lookup missed.
|
||||
if conn.wsHandlerFn.Truthy() {
|
||||
conn.wsHandlerFn.Release()
|
||||
}
|
||||
if conn.onMessageFn.Truthy() {
|
||||
conn.onMessageFn.Release()
|
||||
}
|
||||
if conn.onCloseFn.Truthy() {
|
||||
conn.onCloseFn.Release()
|
||||
}
|
||||
|
||||
p.mu.Lock()
|
||||
delete(p.activeConnections, conn.id)
|
||||
delete(p.destinations, conn.id)
|
||||
delete(p.pendingHandlers, conn.id)
|
||||
p.mu.Unlock()
|
||||
})
|
||||
}
|
||||
|
||||
func (p *RDCleanPathProxy) sendToWebSocket(conn *proxyConnection, data []byte) {
|
||||
|
||||
@@ -13,7 +13,7 @@ import (
|
||||
func CreateJSInterface(client *Client) js.Value {
|
||||
jsInterface := js.Global().Get("Object").Call("create", js.Null())
|
||||
|
||||
jsInterface.Set("write", js.FuncOf(func(this js.Value, args []js.Value) any {
|
||||
writeFunc := js.FuncOf(func(this js.Value, args []js.Value) any {
|
||||
if len(args) < 1 {
|
||||
return js.ValueOf(false)
|
||||
}
|
||||
@@ -32,9 +32,10 @@ func CreateJSInterface(client *Client) js.Value {
|
||||
|
||||
_, err := client.Write(bytes)
|
||||
return js.ValueOf(err == nil)
|
||||
}))
|
||||
})
|
||||
jsInterface.Set("write", writeFunc)
|
||||
|
||||
jsInterface.Set("resize", js.FuncOf(func(this js.Value, args []js.Value) any {
|
||||
resizeFunc := js.FuncOf(func(this js.Value, args []js.Value) any {
|
||||
if len(args) < 2 {
|
||||
return js.ValueOf(false)
|
||||
}
|
||||
@@ -42,14 +43,26 @@ func CreateJSInterface(client *Client) js.Value {
|
||||
rows := args[1].Int()
|
||||
err := client.Resize(cols, rows)
|
||||
return js.ValueOf(err == nil)
|
||||
}))
|
||||
})
|
||||
jsInterface.Set("resize", resizeFunc)
|
||||
|
||||
jsInterface.Set("close", js.FuncOf(func(this js.Value, args []js.Value) any {
|
||||
closeFunc := js.FuncOf(func(this js.Value, args []js.Value) any {
|
||||
client.Close()
|
||||
return js.Undefined()
|
||||
}))
|
||||
})
|
||||
jsInterface.Set("close", closeFunc)
|
||||
|
||||
go readLoop(client, jsInterface)
|
||||
go func() {
|
||||
readLoop(client, jsInterface)
|
||||
// Detach before releasing so late JS calls surface as TypeError instead
|
||||
// of silent "call to released function".
|
||||
jsInterface.Set("write", js.Undefined())
|
||||
jsInterface.Set("resize", js.Undefined())
|
||||
jsInterface.Set("close", js.Undefined())
|
||||
writeFunc.Release()
|
||||
resizeFunc.Release()
|
||||
closeFunc.Release()
|
||||
}()
|
||||
|
||||
return jsInterface
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user