mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-10 15:39:07 +02:00
f0a40e4395
* implement certificate posture check * log signal address * add keychain and cert store support * read the console user's keychain through a user session helper A root daemon cannot reach a login keychain: securityd is per session and a key ACL needs a session to prompt in, so dropping uid is not enough. The daemon now answers certificate challenges from the System keychain itself, where MDM installs device identities, and launches "netbird posture cert-proof" into the console user's desktop session with launchctl asuser for the login keychain. Only the signature and the chain cross back, never the private key. The console user comes from SCDynamicStoreCopyConsoleUser, bound with purego like the keychain calls. The login window reports no user, root, or "loginwindow", and all three are treated as no keychain to read, so a Mac at the lock screen sends device proofs alone. Adds info logging across the path: the keychain search list, per class query status and item counts, the chain built per candidate, and the verification error for every rejected candidate. A run that sends nothing now says why. README.md documents the trust model, the console user limitation and how to read the logs. * read the signed-in user's certificate store on Windows A service reads LocalMachine\MY, where AD and Intune enrol device certificates. CurrentUser\MY lives in the signed-in user's registry hive with keys protected against their profile, and a service that opens it does not fail: "current user" resolves to HKU\S-1-5-18, so it silently reads the service account's own empty store. The service therefore reads the machine store itself and launches "netbird posture cert-proof" with the session token for the rest, mirroring the macOS console user helper. Windows lets a privileged service assume a user identity, so the token goes straight into the child process and no external tooling is involved. CREATE_NO_WINDOW keeps a console window from flashing on the desktop every sync. In-process impersonation would also work but is per OS thread while goroutines migrate, so the child process avoids that class of bug. Session selection prefers the physical console and falls back to any active session, so remote desktop and VDI hosts are covered. WTSQueryUserToken needs SE_TCB_NAME, so a user-run client skips the helper and reads the machine store alone. SystemStore takes a store location, gaining NewUserStore alongside NewSystemStore and the per candidate logging macOS already had. The request building and proof merging move to helper_spawn.go, shared by both platforms, and helperStore picks what the helper reads per platform. * start TPM support * split goreleaser to support pkcs11 and exclude on docker * update goreleaser * go mod tidy * add tpm pin to netbird config * split cert and key location and allow key lookup on tpm * add unsupported flag for mobile devices * Isolate the cert proof helper from the service environment and cap its output * Read the PKCS#11 token PIN from NB_TPM_PIN instead of the profile config * Bound certificate proof collection so a stuck token or keychain cannot hold the sync loop * Stop retrying a PKCS#11 PIN the token rejected * Log certificate posture details at debug level * Sign only nonces and peer keys of the size management issues * Skip certificate files whose key belongs to another certificate * Bound PKCS#11 driver sizes, pin template values, and log out only a login the session owns * Never pass NULL to CFRelease and skip unreadable keychain identities * Keep the macOS keychain code out of iOS and the PKCS#11 driver out of Android * Find a chain to each challenge's CAs through every intermediate the store holds * Require a token label whenever a PKCS#11 PIN is set * Read user certificates only from the session of the active profile's owner * Collect certificate proofs again when the owner's session changes and report lost proofs * Test the PKCS#11 build against SoftHSM in CI and warn once where the build has no driver * Document where an inline PKCS#11 PIN is stored and how it is protected * Refuse PKCS#11 URIs that this client cannot honour instead of widening the match * Trust certificate and key files only when no other user can write or redirect them * Explain a Windows certificate whose key only a legacy CryptoAPI provider holds * Use platform absolute module paths in tests and add a real owner session test for Windows * Match the Windows profile owner by name instead of resolving it through the domain controller * Keep the certificate stores and TPM library out of the WebAssembly build * [client] Read TSS2 key files on go-tpm, checked against the library it replaces The TSS2 parser was the only reason this repository depended on a crypto suite whose own build tooling it inherits. The replacement sits on go-tpm, which was already a direct dependency and is in fact what that suite calls underneath, so this removes a wrapper rather than porting onto a different library: the load, the derived storage root key and the signing commands are the same calls. Swapping a parser on the one path a customer actually runs is not something to assert, so the two are held side by side for this commit. One test feeds the replacement bytes the old library wrote and requires the same key type, empty auth flag, parent handle, blobs and decoded public key; the other feeds both the fixtures the tests are built on, so those are the shape the format calls for and not merely the shape the new parser reads. The scaffolding goes away with the dependency in the commit that follows. The encoder behind the fixtures is written out separately from the parser under test, so an encoder bug and a decoder bug cannot cancel each other out. * [client] Drop go.step.sm/crypto and the repo-wide upgrades it imposed The TSS2 parser was the only thing in the repository that used this module, and it brought 302 modules into the graph to do it — 35 of them linters, along with Google Cloud KMS and IAM, the AWS SDK and a terminal styling library. Those are the module's own development dependencies, which minimal version selection turns into floors in ours, and they are the whole reason gRPC, protobuf, the AWS SDK, OpenTelemetry, logrus and five x/ packages had moved. Management, signal, relay and proxy inherited every one of them for a feature none of them runs. Removing the import is not enough, because tidy never downgrades: the raised floors stay written in go.mod. Each one is pinned back to the version main had, then tidy is left to raise again whatever something still genuinely needs. It raised nothing: all 43 are back where they were, and go-tpm was already in the graph at the same version, so the certificate feature now costs no new module at all. The differential tests go with it. They existed to check the swap against the library while both were present, and there is nothing left to compare against. * [client] Clear the lint findings only the macOS and Windows runners see golangci-lint analyses one build at a time, so running it on Linux says nothing about the two platforms CI also lints. Against those builds the feature's packages reported eight findings, and the structural one is Config.dir: it is dead on macOS and Windows because neither reads a directory at all, their collectors take the configuration and discard it. Moving the method beside its only callers makes that visible in the layout instead of in a linter, and leaves the gap itself — no file or token store on those platforms — where it belongs, as something to decide rather than something to silence. An absent key file beside a certificate was reported as a nil signer with a nil error, which the caller then had to recognise by its nilness. It is a sentinel now, so the meaning is in the error rather than in the absence of one. The rest follow the standard library: the elliptic coordinates and the private scalar come from the encoding helpers rather than the deprecated big.Int fields, and an error string loses its trailing colon. Lint is clean on linux, darwin and windows; the hardware TPM path was exercised separately against a real device and passes. * Accept the TSS2 emptyAuth boolean OpenSSL writes and persistent parents on 32-bit builds * Count the certificates field in the peer meta store test * Check the store directory before listing it, refuse group-writable files, and reject a URI with two PIN sources * Share a PKCS#11 login between sessions and send each PIN at most once at a time * Collect certificate proofs again when the meta sync carrying them failed * Use no Windows user store when a domainless owner matches accounts of several domains * Use no user certificate store when the active profile's owner cannot be read * Document the PIN sources on CertPKCS11URI and keep the README PIN example off the command line * Test that the PKCS#11 URI stays out of the debug bundle and run the wrong-PIN test only on a disposable token * Refuse a TPM PSS signature request for the maximum salt length * Add the certificate fields to the network map golden data * Retry posture checks whose meta sync timed out instead of dropping them * Start no system info gathering while a timed-out one is still running * Guard the applied posture checks across goroutines and keep refreshing proofs while a pending update times out * Log what a successful certificate proof helper wrote to stderr * Send recollected certificate proofs to management only when the proven chains changed * Explain a macOS keychain key whose access list does not allow netbird * Kill the whole macOS certificate helper process group when it times out * End sudo option parsing before the macOS certificate helper binary * Hold off system info gathering only while a timed-out one is still running * Collect certificate proofs on the posture watcher instead of under the sync lock * Read the certificate store directory and PKCS#11 URI from the daemon environment, not the profile config * Install the RPM sysconfig file readable by root only and show the certificate posture variables * Move the certificate posture README into the package doc and the docs site * Name NB_CERT_PKCS11_URI in the PIN-without-token error * Keep the file check results of the latest-started system info refresh * Give the full import command for a keychain key netbird may not use, and correct the package doc * Restrict the service environment file to root on every package install * Search only the System keychain in the macOS daemon and only the login keychain in the user helper * Let the certificate proof helper read the PKCS#11 token from the environment on Linux * Ask a macOS user's keychain again only after an hour when it proved nothing * Clear the lint findings in certificate posture * Hold off the keychain helper only after a completed or timed-out run, independent of CA order * Keep free functions out of the method lists of PKCS11Store, URI and Challenger * Name the post-install permission helper in snake case and shorten the sysconfig certificate block * Drop the certificate store directory from certproof.Config, which only NB_CERT_STORE_DIR sets * [management] Renew certificate challenge nonces on quiet accounts A certificate challenge nonce is accepted for its own window and the one before it, and it only reaches a peer attached to a network map. An account where nothing changes sends no map, so after a day the peer re-sends the nonce it still holds, verification rejects its whole proof set, and the certificates stored for it are dropped. It fails the certificate check and loses every policy gated on it until some unrelated change happens to push a map. The outage repairs itself in seconds, which is what makes it expensive: it is intermittent, it only hits stable networks, and it is not reproducible on demand. Push the account's peers an update often enough that the nonce they hold is never close to expiring. Only accounts whose posture checks actually ask for a certificate are tracked, so a deployment without the feature does no extra work. The refresh runs from one goroutine over a map of accounts rather than a timer per account: the period is hours, so one pass every few minutes costs nothing next to it, and there is no timer to re-arm when an account that falls due sooner appears. Each account's first run is offset by a hash of its ID, because the challenge window is global and an instance restart would otherwise arm every account in the same moment. The push carries no administrative change, so it is counted as a refresh rather than an update and stays out of the figures that track what was edited. (cherry picked from commit7ad4a0df37) * [management] Make the certificate challenge window one knob to turn Renewal was timed against the window in two different ways: the period derived from it, the sweep interval did not. Shortening the window to watch a renewal in an end-to-end run would have left the refresher still looking for due accounts every quarter of an hour, so nothing would have been renewed in time and the test would have reported the feature broken. Derive the sweep from the period, within bounds that keep a very short window from spinning and a normal one from checking less often than is useful, and allow the window itself to be set through the environment so a run can take seconds instead of half a day. A value that cannot be parsed or falls outside the bounds keeps the default, because a window nobody intended is a security property nobody chose, and an override is logged at warning level since it sets how long a device keeps passing the check after its key is gone. Every instance has to be given the same value: the window is part of the nonce, so instances that disagree reject each other's. (cherry picked from commit0e38fcf409) * [management] Pin the property that makes per-peer nonce state unnecessary A nonce carries the window it was minted in, not the instant, and is accepted for that window and the one before it. So a peer re-stamped at least once per window can never be left holding one outside the accepted pair, whenever it was last served and however much life its own nonce had left. That is the whole reason management tracks nothing per peer, and it was resting on an argument rather than a test. The phases are part of the property, not decoration: accounts are deliberately given a refresh phase of their own, so the guarantee has to hold off the window boundary too. The negative case shows why that matters — a cadence of exactly two windows lands inside the grace window when it is aligned to the boundary and leaves a gap when it is not. (cherry picked from commitdee68facfd) * [management] Renew challenges only for the peers that answer one The refresh pushed an update to every connected peer of the account, while only the peers a certificate check applies to carry a nonce. On an account where a handful of peers sit behind the check and the rest do not, everyone was woken several times a day to be handed a map that changed nothing for them. Push to the sources of the enabled policies whose posture checks include a certificate check, which is exactly the set that is sent a challenge. Resolving the set the other way round than the gRPC layer does is the risk here: a peer the refresh forgets stops being renewed and falls out of its policies silently, which is the failure this whole mechanism exists to prevent. So the selection is held against processPeerPostureChecks, the per-peer rule that decides who receives a challenge in the first place, by a test that asks both the same question and requires the same answer. (cherry picked from commitdc4d0e0274) * [management] Derive certificate challenge nonces from the stored encryption key The nonce secret came from the server's WireGuard key, which is generated afresh in every process and never persisted. A nonce carries no state, so the only thing that lets one instance verify what another issued is deriving the same secret — and that premise, written in the comment above the challenger, was not met: every instance had its own key. A peer reconnecting after a restart therefore presented a nonce minted under the previous secret, verification failed with a mismatch, its whole proof set was rejected and the certificates stored for it were dropped until it signed again. Reproduced three times on the lab, each one logging "nonce was not issued to this peer", which only a changed secret produces. On a single instance it costs seconds of lost policy access per restart; across instances it is not transient at all, because every reconnect that lands elsewhere is rejected the same way. Derive from the data store encryption key instead: it is generated once, written back to the configuration and read by every instance, so it survives restarts and is shared. Where none is configured the secret falls back to the WireGuard key with a warning — degraded but still unpredictable, which is the property that matters most: a peer able to guess it could mint the nonces of future windows, sign them while its key is present and keep passing after it is gone. The challenger is now built once and passed to the two places that need it, rather than re-derived per message. (cherry picked from commit278f2f3807) * [management] Register an account for renewal where its nonce is issued Renewal was armed when a peer connected or when a posture check was saved, both of which ask the store whether the account has a certificate check. That misses the case it most needs to catch: the check is created through one instance while the peers are connected to another, so the instance serving them never learns it has anything to renew and their nonce expires. It also charged a query to every peer connect in every account, including the ones that will never use the feature, which a fleet reconnecting after a restart pays all at once. Register where the nonce is actually stamped instead. A nonce is verified from a shared secret and so travels between instances, but the renewal that keeps it fresh cannot: only the instance holding a peer's stream can push to it. Issuing and renewing now line up by construction — an instance renews exactly the accounts it has issued nonces for — and an instance that never issues one has nothing to renew, so there is no case left to miss. The registration is a map insert with no store access, which is what lets it sit on a path taken by every login and every initial sync. Reported by Viktor Liu, who also proposed registering at the point of issue. (cherry picked from commit 2d16dd7d7cf54762f2e64c5630ea092f32ef63ab) * [management] Register for renewal on pushed updates, not only on connect Registering where the nonce is stamped only covered the login and the initial sync, which both happen when a peer opens a stream. That left out the path the mechanism exists for. On the cloud the network map controller is wrapped so that an update publishes to an event bus instead of pushing locally: an instance handling a REST change broadcasts, and every instance holding a peer of that account pushes to its own. Those pushes stamp a nonce through the update handler, and nothing there registered, so an instance learned about an account only when one of its peers happened to reconnect. For a quiet fleet that is the original bug: the check is created, the peers are told about it, and nobody renews what they were told. Registering on the pushed update closes it, and is the difference between stamping and marking a peer connected — one happens on every push, the other only when a stream opens. Reported by Viktor Liu; the broadcast that makes it work was pointed out by Pascal Fischer. (cherry picked from commit 59efe8d93e53bacdf57cb546f4ab2c19dc4eddab) * [management] Let the challenge refresh loop stop with the manager that owns it The loop was started on a context explicitly detached from the caller's, so nothing could ever stop it. Production is unaffected either way, since BuildManager is called with context.Background(), but a test that builds a manager leaked a sweeping goroutine for the rest of the run, and a shutdown path added later would have had no way to reach it. Take the manager's context as the request buffer built on the line above already does. The test pins the contract the loop offers, so a detached context cannot come back inside Start either. * [management] Bound one account's challenge refresh so it cannot starve the rest Resolving which peers answer a challenge reads the store three times, and the refresher sweeps accounts one after another on a single goroutine. A read that never returns held the sweep for the life of the process, so every other account on the instance stopped being renewed and its peers fell out of the policies gated on the check: one account's bad luck became an outage for all of them. Give each refresh the sweep interval it is allowed to occupy, capped at 30s so a 12-hour window does not grant minutes to a query that should take milliseconds. A refresh that runs out of time keeps its account tracked, since a deadline says nothing about whether that account still has a certificate check. * [management] Send challenge refreshes down the path the rest of management uses The refresh dispatched through UpdateAffectedPeers, the one variant that takes no reason, so it was missing from the update counters and coalesced with nothing. An administrator editing a policy while the sweep ran made the account's network map twice over, and UpdateOperationRefresh, added for exactly this caller, was never referenced. Buffer it with a posture_check/refresh reason instead. The periodic push is now visible in the metrics as what it is, distinct from an edit, and the send detaches from the sweep deadline on its own, so that deadline bounds the store reads it was meant for. * [management] Keep the certificate challenge comments to what the history does not say Four of these ran to three and four times the comment budget, the longest at 992 characters. Most of the excess argued against designs that were never written or explained a bug that no longer exists in the code, which is what the commit that fixed it is for. What is left is the part a reader cannot recover from the code: that the nonce secret has to be persisted and unpredictable, that stamping and renewing are decided together because only the serving instance can push, and that the target rule is the inverse of processPeerPostureChecks. * Keep the newest posture checks pending whatever made their meta sync fail * Report no lost certificate when the engine stops during a proof collection * Share the proof collection single-flight across engine restarts * Close a PKCS#11 module that loads but cannot be used * Fix the pending checks comments * Renew certificate challenges only for the peers streamed to this instance * Ignore a challenge stamp from an older sync stream of the same peer * Kill the Windows certificate proof helper with its whole process tree * Expect the challenge untrack in the session ownership test * Drop an invalid certificate proof without discarding the valid ones * Start a system info gathering beside one that has been stuck for ten timeouts --------- Co-authored-by: pascal <pascal@netbird.io> Co-authored-by: mlsmaycon <mlsmaycon@gmail.com> Co-authored-by: riccardom <riccardomanfrin@gmail.com>
2990 lines
92 KiB
Go
2990 lines
92 KiB
Go
package internal
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"math/rand"
|
|
"net"
|
|
"net/netip"
|
|
"net/url"
|
|
"os"
|
|
"runtime"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/hashicorp/go-multierror"
|
|
"github.com/pion/ice/v4"
|
|
"github.com/pion/stun/v3"
|
|
log "github.com/sirupsen/logrus"
|
|
wgdevice "golang.zx2c4.com/wireguard/device"
|
|
"golang.zx2c4.com/wireguard/tun/netstack"
|
|
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
|
|
|
|
"github.com/netbirdio/netbird/client/anonymize"
|
|
nberrors "github.com/netbirdio/netbird/client/errors"
|
|
"github.com/netbirdio/netbird/client/firewall"
|
|
"github.com/netbirdio/netbird/client/firewall/firewalld"
|
|
firewallManager "github.com/netbirdio/netbird/client/firewall/manager"
|
|
"github.com/netbirdio/netbird/client/firewall/uspfilter/forwarder"
|
|
"github.com/netbirdio/netbird/client/iface"
|
|
"github.com/netbirdio/netbird/client/iface/device"
|
|
nbnetstack "github.com/netbirdio/netbird/client/iface/netstack"
|
|
"github.com/netbirdio/netbird/client/iface/udpmux"
|
|
"github.com/netbirdio/netbird/client/iface/wgaddr"
|
|
"github.com/netbirdio/netbird/client/internal/acl"
|
|
"github.com/netbirdio/netbird/client/internal/certproof"
|
|
"github.com/netbirdio/netbird/client/internal/debug"
|
|
"github.com/netbirdio/netbird/client/internal/dns"
|
|
dnsconfig "github.com/netbirdio/netbird/client/internal/dns/config"
|
|
"github.com/netbirdio/netbird/client/internal/dnsfwd"
|
|
"github.com/netbirdio/netbird/client/internal/expose"
|
|
"github.com/netbirdio/netbird/client/internal/lazyconn"
|
|
"github.com/netbirdio/netbird/client/internal/metrics"
|
|
"github.com/netbirdio/netbird/client/internal/netflow"
|
|
nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
|
|
"github.com/netbirdio/netbird/client/internal/networkmonitor"
|
|
"github.com/netbirdio/netbird/client/internal/peer"
|
|
"github.com/netbirdio/netbird/client/internal/peer/guard"
|
|
icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
|
|
"github.com/netbirdio/netbird/client/internal/peerstore"
|
|
"github.com/netbirdio/netbird/client/internal/portforward"
|
|
"github.com/netbirdio/netbird/client/internal/profilemanager"
|
|
"github.com/netbirdio/netbird/client/internal/relay"
|
|
"github.com/netbirdio/netbird/client/internal/rosenpass"
|
|
"github.com/netbirdio/netbird/client/internal/routemanager"
|
|
"github.com/netbirdio/netbird/client/internal/statemanager"
|
|
"github.com/netbirdio/netbird/client/internal/stdnet"
|
|
"github.com/netbirdio/netbird/client/internal/syncstore"
|
|
"github.com/netbirdio/netbird/client/internal/updater"
|
|
"github.com/netbirdio/netbird/client/jobexec"
|
|
"github.com/netbirdio/netbird/client/netevents"
|
|
cProto "github.com/netbirdio/netbird/client/proto"
|
|
"github.com/netbirdio/netbird/client/system"
|
|
nbdns "github.com/netbirdio/netbird/dns"
|
|
"github.com/netbirdio/netbird/route"
|
|
mgm "github.com/netbirdio/netbird/shared/management/client"
|
|
"github.com/netbirdio/netbird/shared/management/domain"
|
|
sharedgrpc "github.com/netbirdio/netbird/shared/management/grpc"
|
|
nbnetworkmap "github.com/netbirdio/netbird/shared/management/networkmap"
|
|
mgmProto "github.com/netbirdio/netbird/shared/management/proto"
|
|
types "github.com/netbirdio/netbird/shared/management/types"
|
|
"github.com/netbirdio/netbird/shared/netiputil"
|
|
auth "github.com/netbirdio/netbird/shared/relay/auth/hmac"
|
|
relayClient "github.com/netbirdio/netbird/shared/relay/client"
|
|
signal "github.com/netbirdio/netbird/shared/signal/client"
|
|
sProto "github.com/netbirdio/netbird/shared/signal/proto"
|
|
"github.com/netbirdio/netbird/util"
|
|
"github.com/netbirdio/netbird/util/capture"
|
|
"github.com/netbirdio/netbird/version"
|
|
)
|
|
|
|
// PeerConnectionTimeoutMax is a timeout of an initial connection attempt to a remote peer.
|
|
// E.g. this peer will wait PeerConnectionTimeoutMax for the remote peer to respond,
|
|
// if not successful then it will retry the connection attempt.
|
|
// Todo pass timeout at EnginConfig
|
|
const (
|
|
PeerConnectionTimeoutMax = 45000 // ms
|
|
PeerConnectionTimeoutMin = 30000 // ms
|
|
disableAutoUpdate = "disabled"
|
|
|
|
// systemInfoTimeout bounds how long the sync loop waits for system info / posture
|
|
// check gathering. The gathering runs uncancellable system calls (process scan,
|
|
// exec, os.Stat); without this bound a single stuck call freezes handleSync, and
|
|
// thus syncMsgMux, for as long as the call hangs (observed multi-minute freezes).
|
|
systemInfoTimeout = 15 * time.Second
|
|
|
|
// dnsForwarderStopTimeout bounds how long stopping the DNS forwarder waits
|
|
// for the queries still in flight. One waiting on an unresponsive upstream
|
|
// would otherwise hold the stop for the whole upstream timeout, and the
|
|
// stop runs with syncMsgMux held. The sockets are closed either way, so
|
|
// giving up costs a query that was already failing.
|
|
dnsForwarderStopTimeout = 2 * time.Second
|
|
)
|
|
|
|
var ErrResetConnection = fmt.Errorf("reset connection")
|
|
|
|
var ErrEngineAlreadyStarted = errors.New("engine already started")
|
|
|
|
type EngineConfig struct {
|
|
WgPort int
|
|
WgIfaceName string
|
|
|
|
// WgAddr is the Wireguard local address (Netbird Network IP).
|
|
// Contains both v4 and optional v6 overlay addresses.
|
|
WgAddr wgaddr.Address
|
|
|
|
// WgPrivateKey is a Wireguard private key of our peer (it MUST never leave the machine)
|
|
WgPrivateKey wgtypes.Key
|
|
|
|
// NetworkMonitor is a flag to enable network monitoring
|
|
NetworkMonitor bool
|
|
|
|
// IFaceBlackList is a list of network interfaces to ignore when discovering connection candidates (ICE related)
|
|
IFaceBlackList []string
|
|
DisableIPv6Discovery bool
|
|
|
|
PreSharedKey *wgtypes.Key
|
|
|
|
// UDPMuxPort default value 0 - the system will pick an available port
|
|
UDPMuxPort int
|
|
|
|
// UDPMuxSrflxPort default value 0 - the system will pick an available port
|
|
UDPMuxSrflxPort int
|
|
|
|
// SSHKey is a private SSH key in a PEM format
|
|
SSHKey []byte
|
|
|
|
NATExternalIPs []string
|
|
|
|
CustomDNSAddress string
|
|
|
|
RosenpassEnabled bool
|
|
RosenpassPermissive bool
|
|
|
|
ServerSSHAllowed bool
|
|
RemoteJobsAllowed bool
|
|
EnableSSHRoot *bool
|
|
EnableSSHSFTP *bool
|
|
EnableSSHLocalPortForwarding *bool
|
|
EnableSSHRemotePortForwarding *bool
|
|
DisableSSHAuth *bool
|
|
|
|
DNSRouteInterval time.Duration
|
|
|
|
DisableClientRoutes bool
|
|
DisableServerRoutes bool
|
|
DisableDNS bool
|
|
DisableFirewall bool
|
|
BlockLANAccess bool
|
|
BlockInbound bool
|
|
DisableIPv6 bool
|
|
SyncMessageVersion *int
|
|
|
|
// LazyConnection is the MDM-sourced lazy-connection override; StateUnset defers to
|
|
// the env var and management feature flag.
|
|
LazyConnection lazyconn.State
|
|
|
|
MTU uint16
|
|
|
|
CertStore certproof.Config
|
|
|
|
// for debug bundle generation
|
|
ProfileConfig *profilemanager.Config
|
|
|
|
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.
|
|
type EngineServices struct {
|
|
SignalClient signal.Client
|
|
MgmClient mgm.Client
|
|
RelayManager *relayClient.Manager
|
|
StatusRecorder *peer.Status
|
|
Checks []*mgmProto.Checks
|
|
StateManager *statemanager.Manager
|
|
UpdateManager *updater.Manager
|
|
ClientMetrics *metrics.ClientMetrics
|
|
MetricsCtx context.Context
|
|
// NetMgr gates the reconnection loops on OS-reported network
|
|
// availability; nil disables gating.
|
|
NetMgr *netevents.Manager
|
|
}
|
|
|
|
// Engine is a mechanism responsible for reacting on Signal and Management stream events and managing connections to the remote peers.
|
|
type Engine struct {
|
|
// signal is a Signal Service client
|
|
signal signal.Client
|
|
signaler *peer.Signaler
|
|
// mgmClient is a Management Service client
|
|
mgmClient mgm.Client
|
|
// peerConns is a map that holds all the peers that are known to this peer
|
|
peerStore *peerstore.Store
|
|
|
|
connMgr *ConnMgr
|
|
|
|
// rpManager is a Rosenpass manager
|
|
rpManager *rosenpass.Manager
|
|
|
|
// syncMsgMux is used to guarantee sequential Management Service message processing
|
|
syncMsgMux *sync.Mutex
|
|
|
|
config *EngineConfig
|
|
mobileDep MobileDependency
|
|
|
|
// netMgr gates the peer reconnection guards on OS-reported network
|
|
// availability; nil disables gating.
|
|
netMgr *netevents.Manager
|
|
|
|
// STUNs is a list of STUN servers used by ICE
|
|
STUNs []*stun.URI
|
|
// TURNs is a list of STUN servers used by ICE
|
|
TURNs []*stun.URI
|
|
stunTurn icemaker.StunTurn
|
|
|
|
clientCtx context.Context
|
|
clientCancel context.CancelFunc
|
|
|
|
ctx context.Context
|
|
cancel context.CancelFunc
|
|
|
|
started bool
|
|
|
|
wgInterface WGIface
|
|
|
|
// wgDevice is a lock-free handle on the WireGuard device behind
|
|
// wgInterface. Reaching the device through wgInterface requires
|
|
// syncMsgMux, which handleSync holds while it adds and removes peers;
|
|
// SetPerformance must stay reachable exactly when that work is stuck.
|
|
wgDevice atomic.Pointer[wgdevice.Device]
|
|
|
|
udpMux *udpmux.UniversalUDPMuxDefault
|
|
|
|
// wgDetector is shared by every ICE agent through the ICE config.
|
|
wgDetector *stdnet.WGDetector
|
|
|
|
// networkSerial is the latest CurrentSerial (state ID) of the network sent by the Management service
|
|
networkSerial uint64
|
|
|
|
// latestComponents is the most-recent NetworkMapComponents decoded from
|
|
// a NetworkMapEnvelope (capability=3 peers only). Held alongside the
|
|
// NetworkMap that Calculate() produced from it so future incremental
|
|
// updates have a base to apply changes against. nil for legacy-format
|
|
// peers. Guarded by syncMsgMux.
|
|
latestComponents *types.NetworkMapComponents
|
|
|
|
networkMonitor *networkmonitor.NetworkMonitor
|
|
|
|
sshServer sshServer
|
|
|
|
statusRecorder *peer.Status
|
|
|
|
firewall firewallManager.Manager
|
|
routeManager routemanager.Manager
|
|
acl acl.Manager
|
|
dnsForwardMgr *dnsfwd.Manager
|
|
|
|
dnsServer dns.Server
|
|
|
|
// checks are the client-applied posture checks that need to be evaluated on the client.
|
|
// Writers hold syncMsgMux and checksMu; readers hold either, see appliedChecks.
|
|
checks []*mgmProto.Checks
|
|
checksMu sync.RWMutex
|
|
// pendingChecks are the newest received checks whose meta sync failed; the posture
|
|
// watcher retries them. Both are guarded by syncMsgMux, and
|
|
// hasPendingChecks lets the watcher skip the lock when nothing is pending.
|
|
pendingChecks []*mgmProto.Checks
|
|
hasPendingChecks atomic.Bool
|
|
// infoTimeout overrides systemInfoTimeout when set.
|
|
infoTimeout time.Duration
|
|
|
|
// certProofs answers the certificate challenges in checks within a bounded time, and
|
|
// certState remembers what it last proved.
|
|
certProofs certproof.Collector
|
|
certState certPostureState
|
|
// certWake wakes the posture watcher, which owns proof collection; certSendMu orders
|
|
// the meta syncs that carry proofs.
|
|
certWake chan struct{}
|
|
certSendMu sync.Mutex
|
|
|
|
infoSource system.InfoSource
|
|
|
|
relayManager *relayClient.Manager
|
|
stateManager *statemanager.Manager
|
|
portForwardManager *portforward.Manager
|
|
srWatcher *guard.SRWatcher
|
|
|
|
afpacketCapture *capture.AFPacketCapture
|
|
|
|
// 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
|
|
updateManager *updater.Manager
|
|
|
|
// WireGuard interface monitor
|
|
wgIfaceMonitor *WGIfaceMonitor
|
|
|
|
// shutdownWg tracks all long-running goroutines to ensure clean shutdown
|
|
shutdownWg sync.WaitGroup
|
|
|
|
probeStunTurn *relay.StunTurnProbe
|
|
|
|
// clientMetrics collects and pushes metrics
|
|
clientMetrics *metrics.ClientMetrics
|
|
metricsCtx context.Context
|
|
|
|
jobExecutor *jobexec.Executor
|
|
jobExecutorWG sync.WaitGroup
|
|
|
|
exposeManager *expose.Manager
|
|
|
|
sessionWatcher sessionDeadlineWatcher
|
|
}
|
|
|
|
// sessionDeadlineWatcher is the engine-facing surface of the SSO session
|
|
// expiry watcher. The concrete implementation (sessionwatch.Watcher) is wired
|
|
// in via newSessionWatcher, which is build-tagged so the js/wasm build links a
|
|
// no-op stub instead of pulling the full sessionwatch package (and its timer
|
|
// machinery) into the binary — the wasm client never runs the engine's
|
|
// session-warning flow.
|
|
type sessionDeadlineWatcher interface {
|
|
Update(deadline time.Time) error
|
|
Dismiss()
|
|
Close()
|
|
}
|
|
|
|
// Peer is an instance of the Connection Peer
|
|
type Peer struct {
|
|
WgPubKey string
|
|
WgAllowedIps string
|
|
}
|
|
|
|
type localIpUpdater interface {
|
|
UpdateLocalIPs() error
|
|
}
|
|
|
|
// overlayRebind rebuilds one subsystem's sockets on the current interface. The
|
|
// error it returns names its own subsystem, since the caller can only log it.
|
|
type overlayRebind func() error
|
|
|
|
// NewEngine creates a new Connection Engine with probes attached
|
|
func NewEngine(
|
|
clientCtx context.Context,
|
|
clientCancel context.CancelFunc,
|
|
config *EngineConfig,
|
|
services EngineServices,
|
|
mobileDep MobileDependency,
|
|
) *Engine {
|
|
// The engine is single-use: a fresh instance is built per connection
|
|
// cycle (see Client.run), so the run context is created once here rather
|
|
// than in Start.
|
|
ctx, cancel := context.WithCancel(clientCtx)
|
|
engine := &Engine{
|
|
clientCtx: clientCtx,
|
|
clientCancel: clientCancel,
|
|
ctx: ctx,
|
|
cancel: cancel,
|
|
signal: services.SignalClient,
|
|
signaler: peer.NewSignaler(services.SignalClient, config.WgPrivateKey),
|
|
mgmClient: services.MgmClient,
|
|
relayManager: services.RelayManager,
|
|
peerStore: peerstore.NewConnStore(),
|
|
wgDetector: stdnet.NewWGDetector(),
|
|
syncMsgMux: &sync.Mutex{},
|
|
config: config,
|
|
mobileDep: mobileDep,
|
|
netMgr: services.NetMgr,
|
|
STUNs: []*stun.URI{},
|
|
TURNs: []*stun.URI{},
|
|
networkSerial: 0,
|
|
statusRecorder: services.StatusRecorder,
|
|
stateManager: services.StateManager,
|
|
portForwardManager: portforward.NewManager(),
|
|
checks: services.Checks,
|
|
certWake: make(chan struct{}, 1),
|
|
probeStunTurn: relay.NewStunTurnProbe(relay.DefaultCacheTTL),
|
|
jobExecutor: jobexec.NewExecutor(),
|
|
clientMetrics: services.ClientMetrics,
|
|
metricsCtx: services.MetricsCtx,
|
|
updateManager: services.UpdateManager,
|
|
syncStoreDir: config.StateDir,
|
|
}
|
|
// sessionWatcher keeps the SubscribeStatus consumers in sync with the
|
|
// session expiry deadline. Deadline-change ticks come for free via
|
|
// Status.SetSessionExpiresAt; the watcher exists to push a wake-up at
|
|
// T-WarningLead and T-FinalWarningLead so the UI repaints the remaining
|
|
// time / warning state even when nothing else changed, and to publish
|
|
// two SystemEvents (the warning composition lives in sessionwatch so
|
|
// the wire format stays owned by one package):
|
|
// - T-WarningLead → interactive "Extend now / Dismiss" notification
|
|
// - T-FinalWarningLead → auto-opened SessionAboutToExpire dialog,
|
|
// suppressed when the user dismissed the earlier warning
|
|
engine.sessionWatcher = newSessionWatcher(engine.statusRecorder)
|
|
|
|
log.Infof("I am: %s", config.WgPrivateKey.PublicKey().String())
|
|
return engine
|
|
}
|
|
|
|
func (e *Engine) Stop() error {
|
|
if e == nil {
|
|
// this seems to be a very odd case but there was the possibility if the netbird down command comes before the engine is fully started
|
|
log.Debugf("tried stopping engine that is nil")
|
|
return nil
|
|
}
|
|
e.cancel()
|
|
e.syncMsgMux.Lock()
|
|
|
|
e.stopLocked()
|
|
|
|
e.syncMsgMux.Unlock()
|
|
|
|
timeout := e.calculateShutdownTimeout()
|
|
log.Debugf("waiting for goroutines to finish with timeout: %v", timeout)
|
|
shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout)
|
|
defer cancel()
|
|
|
|
if err := waitWithContext(shutdownCtx, &e.shutdownWg); err != nil {
|
|
log.Warnf("shutdown timeout exceeded after %v, some goroutines may still be running", timeout)
|
|
}
|
|
|
|
log.Infof("stopped Netbird Engine")
|
|
|
|
return nil
|
|
}
|
|
|
|
// stopLocked tears down everything Start may have brought up, in the order
|
|
// teardown requires (DNS before the interface goes down, flow manager after).
|
|
// The caller must hold syncMsgMux. It is shared by Stop and by Start's failure
|
|
// path, so a partially-initialized engine is cleaned up the same way; every
|
|
// step is nil-guarded. It does not wait on shutdownWg — the caller does that
|
|
// after releasing the lock, since the goroutines also take syncMsgMux.
|
|
func (e *Engine) stopLocked() {
|
|
if e.connMgr != nil {
|
|
e.connMgr.Close()
|
|
}
|
|
|
|
// stopping network monitor first to avoid starting the engine again
|
|
if e.networkMonitor != nil {
|
|
e.networkMonitor.Stop()
|
|
}
|
|
log.Info("Network monitor: stopped")
|
|
|
|
if err := e.stopSSHServer(); err != nil {
|
|
log.Warnf("failed to stop SSH server: %v", err)
|
|
}
|
|
|
|
e.cleanupSSHConfig()
|
|
|
|
if e.srWatcher != nil {
|
|
e.srWatcher.Close()
|
|
}
|
|
|
|
if e.sessionWatcher != nil {
|
|
e.sessionWatcher.Close()
|
|
}
|
|
|
|
if e.updateManager != nil {
|
|
e.updateManager.ResetMode()
|
|
}
|
|
|
|
log.Info("cleaning up status recorder states")
|
|
e.statusRecorder.ReplaceOfflinePeers([]peer.State{})
|
|
e.statusRecorder.UpdateDNSStates([]peer.NSGroupState{})
|
|
e.statusRecorder.UpdateRelayStates([]relay.ProbeResult{})
|
|
|
|
if err := e.removeAllPeers(); err != nil {
|
|
log.Errorf("failed to remove all peers: %s", err)
|
|
}
|
|
|
|
if e.routeManager != nil {
|
|
e.routeManager.Stop(e.stateManager)
|
|
}
|
|
|
|
e.stopDNSForwarder()
|
|
|
|
// stop/restore DNS after peers are closed but before interface goes down
|
|
// so dbus and friends don't complain because of a missing interface
|
|
e.stopDNSServer()
|
|
|
|
e.jobExecutorWG.Wait() // block until job goroutines finish
|
|
|
|
e.close()
|
|
|
|
// stop flow manager after wg interface is gone
|
|
if e.flowManager != nil {
|
|
e.flowManager.Close()
|
|
}
|
|
|
|
stateCtx, stateCancel := context.WithTimeout(context.Background(), 3*time.Second)
|
|
defer stateCancel()
|
|
|
|
if err := e.stateManager.Stop(stateCtx); err != nil {
|
|
log.Errorf("failed to stop state manager: %v", err)
|
|
}
|
|
if err := e.stateManager.PersistState(context.Background()); err != nil {
|
|
log.Errorf("failed to persist state: %v", err)
|
|
}
|
|
}
|
|
|
|
// calculateShutdownTimeout returns shutdown timeout: 10s base + 100ms per peer, capped at 30s.
|
|
func (e *Engine) calculateShutdownTimeout() time.Duration {
|
|
peerCount := len(e.peerStore.PeersPubKey())
|
|
|
|
baseTimeout := 10 * time.Second
|
|
perPeerTimeout := time.Duration(peerCount) * 100 * time.Millisecond
|
|
timeout := baseTimeout + perPeerTimeout
|
|
|
|
maxTimeout := 30 * time.Second
|
|
if timeout > maxTimeout {
|
|
timeout = maxTimeout
|
|
}
|
|
|
|
return timeout
|
|
}
|
|
|
|
// waitWithContext waits for WaitGroup with timeout, returns ctx.Err() on timeout.
|
|
func waitWithContext(ctx context.Context, wg *sync.WaitGroup) error {
|
|
done := make(chan struct{})
|
|
go func() {
|
|
wg.Wait()
|
|
close(done)
|
|
}()
|
|
|
|
select {
|
|
case <-done:
|
|
return nil
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
}
|
|
|
|
// Start creates a new WireGuard tunnel interface and listens to events from Signal and Management services
|
|
// Connections to remote peers are not established here.
|
|
// However, they will be established once an event with a list of peers to connect to will be received from Management Service
|
|
func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL) (err error) {
|
|
e.syncMsgMux.Lock()
|
|
defer e.syncMsgMux.Unlock()
|
|
|
|
// The engine is single-use. Reject a duplicate start and a start on an
|
|
// already-stopped engine (run context cancelled).
|
|
if e.started {
|
|
return ErrEngineAlreadyStarted
|
|
}
|
|
|
|
if ctxErr := e.ctx.Err(); ctxErr != nil {
|
|
return fmt.Errorf("engine already stopped: %w", ctxErr)
|
|
}
|
|
|
|
e.started = true
|
|
|
|
// Tear down any partially-initialized state on a failed start. Cancel the
|
|
// run context first so goroutines started before the failure (connMgr,
|
|
// srWatcher, monitors) unwind, then stopLocked mirrors Stop's teardown (we
|
|
// already hold syncMsgMux), cleaning up route/DNS/flow/state managers too,
|
|
// not just what close() covers.
|
|
defer func() {
|
|
if err != nil {
|
|
e.cancel()
|
|
e.stopLocked()
|
|
}
|
|
}()
|
|
|
|
if err = iface.ValidateMTU(e.config.MTU); err != nil {
|
|
return fmt.Errorf("invalid MTU configuration: %w", err)
|
|
}
|
|
|
|
e.exposeManager = expose.NewManager(e.ctx, e.mgmClient)
|
|
|
|
wgIface, err := e.newWgIface()
|
|
if err != nil {
|
|
log.Errorf("failed creating wireguard interface instance %s: [%s]", e.config.WgIfaceName, err)
|
|
return fmt.Errorf("new wg interface: %w", err)
|
|
}
|
|
e.wgInterface = wgIface
|
|
e.statusRecorder.SetWgIface(wgIface)
|
|
|
|
// start flow manager right after interface creation
|
|
publicKey := e.config.WgPrivateKey.PublicKey()
|
|
e.flowManager = netflow.NewManager(e.wgInterface, publicKey[:], e.statusRecorder)
|
|
|
|
if e.config.RosenpassEnabled {
|
|
log.Infof("rosenpass is enabled")
|
|
if e.config.RosenpassPermissive {
|
|
log.Infof("running rosenpass in permissive mode")
|
|
} else {
|
|
log.Infof("running rosenpass in strict mode")
|
|
}
|
|
e.rpManager, err = rosenpass.NewManager(e.config.PreSharedKey, e.config.WgIfaceName, publicKey)
|
|
if err != nil {
|
|
return fmt.Errorf("create rosenpass manager: %w", err)
|
|
}
|
|
if err := e.rpManager.Run(); err != nil {
|
|
return fmt.Errorf("run rosenpass manager: %w", err)
|
|
}
|
|
}
|
|
e.stateManager.Start()
|
|
|
|
dnsServer, err := e.newDnsServer()
|
|
if err != nil {
|
|
return fmt.Errorf("create dns server: %w", err)
|
|
}
|
|
e.dnsServer = dnsServer
|
|
|
|
// Populate DNS cache with NetbirdConfig and management URL for early resolution
|
|
if err := e.PopulateNetbirdConfig(netbirdConfig, mgmtURL); err != nil {
|
|
log.Warnf("failed to populate DNS cache: %v", err)
|
|
}
|
|
|
|
e.routeManager = routemanager.NewManager(routemanager.ManagerConfig{
|
|
Context: e.ctx,
|
|
PublicKey: e.config.WgPrivateKey.PublicKey().String(),
|
|
DNSRouteInterval: e.config.DNSRouteInterval,
|
|
WGInterface: e.wgInterface,
|
|
StatusRecorder: e.statusRecorder,
|
|
RelayManager: e.relayManager,
|
|
StateManager: e.stateManager,
|
|
DNSServer: dnsServer,
|
|
PeerStore: e.peerStore,
|
|
DisableClientRoutes: e.config.DisableClientRoutes,
|
|
DisableServerRoutes: e.config.DisableServerRoutes,
|
|
})
|
|
if err := e.routeManager.Init(); err != nil {
|
|
log.Errorf("Failed to initialize route manager: %s", err)
|
|
}
|
|
|
|
e.routeManager.SetRouteChangeListener(e.mobileDep.NetworkChangeListener)
|
|
|
|
e.dnsServer.SetRouteSources(e.routeManager.GetSelectedClientRoutes, e.routeManager.GetActiveClientRoutes)
|
|
|
|
if err = e.wgInterfaceCreate(); err != nil {
|
|
log.Errorf("failed creating tunnel interface %s: [%s]", e.config.WgIfaceName, err.Error())
|
|
return fmt.Errorf("create wg interface: %w", err)
|
|
}
|
|
|
|
if filteredDevice := e.wgInterface.GetDevice(); filteredDevice != nil {
|
|
filteredDevice.SetPanicHandler(e.triggerClientRestart)
|
|
}
|
|
|
|
if err := e.createFirewall(); err != nil {
|
|
return err
|
|
}
|
|
|
|
// Inject firewall into DNS server now that it's available.
|
|
// The DNS server is created before the firewall because the route manager
|
|
// depends on the DNS server, and the firewall depends on the wg interface.
|
|
e.dnsServer.SetFirewall(e.firewall)
|
|
|
|
e.udpMux, err = e.wgInterface.Up()
|
|
if err != nil {
|
|
log.Errorf("failed to pull up wgInterface [%s]: %s", e.wgInterface.Name(), err.Error())
|
|
return fmt.Errorf("up wg interface: %w", err)
|
|
}
|
|
e.wgDevice.Store(e.wgInterface.GetWGDevice())
|
|
|
|
// Start after interface is up since port may have been resolved from 0 or changed if occupied
|
|
e.shutdownWg.Add(1)
|
|
go func() {
|
|
defer e.shutdownWg.Done()
|
|
e.portForwardManager.Start(e.ctx, uint16(e.config.WgPort))
|
|
}()
|
|
|
|
e.shutdownWg.Add(1)
|
|
go func(ctx context.Context) {
|
|
defer e.shutdownWg.Done()
|
|
e.watchCertificatePosture(ctx)
|
|
}(e.ctx)
|
|
|
|
// Set the WireGuard interface for rosenpass after interface is up
|
|
if e.rpManager != nil {
|
|
e.rpManager.SetInterface(e.wgInterface)
|
|
}
|
|
|
|
// if inbound conns are blocked there is no need to create the ACL manager
|
|
if e.firewall != nil && !e.config.BlockInbound {
|
|
e.acl = acl.NewDefaultManager(e.firewall)
|
|
}
|
|
|
|
if err := e.dnsServer.Initialize(); err != nil {
|
|
return fmt.Errorf("initialize dns server: %w", err)
|
|
}
|
|
|
|
iceCfg := e.createICEConfig()
|
|
|
|
e.connMgr = NewConnMgr(e.config, e.statusRecorder, e.peerStore, wgIface)
|
|
e.connMgr.SetRoutedIPsReconciler(func(peerKey string) error {
|
|
if e.routeManager == nil {
|
|
return nil
|
|
}
|
|
return e.routeManager.ReconcilePeerAllowedIPs(peerKey)
|
|
})
|
|
e.connMgr.Start(e.ctx)
|
|
|
|
// Wire DNS-time lazy-connection warm-up now that the connection manager
|
|
// exists (it does not at DNS-server construction time). A DNS answer that
|
|
// points at an idle peer then wakes it before the client's first request.
|
|
e.dnsServer.SetPeerActivator(&dnsPeerActivator{
|
|
connMgr: e.connMgr,
|
|
peerStore: e.peerStore,
|
|
status: e.statusRecorder,
|
|
ctx: e.ctx,
|
|
})
|
|
|
|
e.srWatcher = guard.NewSRWatcher(e.signal, e.relayManager, e.mobileDep.IFaceDiscover, iceCfg)
|
|
e.srWatcher.Start(peer.IsForceRelayed())
|
|
|
|
if err = e.receiveSignalEvents(); err != nil {
|
|
return err
|
|
}
|
|
e.receiveManagementEvents()
|
|
e.receiveJobEvents()
|
|
|
|
// starting network monitor at the very last to avoid disruptions
|
|
e.startNetworkMonitor()
|
|
|
|
// monitor WireGuard interface lifecycle and restart engine on changes
|
|
e.wgIfaceMonitor = NewWGIfaceMonitor()
|
|
e.shutdownWg.Add(1)
|
|
wgIfaceName := e.wgInterface.Name()
|
|
|
|
go func() {
|
|
defer e.shutdownWg.Done()
|
|
|
|
if shouldRestart, err := e.wgIfaceMonitor.Start(e.ctx, wgIfaceName); shouldRestart {
|
|
log.Infof("WireGuard interface monitor: %s, restarting engine", err)
|
|
e.triggerClientRestart()
|
|
} else if err != nil {
|
|
log.Warnf("WireGuard interface monitor: %s", err)
|
|
}
|
|
}()
|
|
|
|
return nil
|
|
}
|
|
|
|
func (e *Engine) createFirewall() error {
|
|
if e.config.DisableFirewall {
|
|
log.Infof("firewall is disabled")
|
|
return nil
|
|
}
|
|
|
|
firewalld.SetParentContext(e.ctx)
|
|
|
|
var err error
|
|
e.firewall, err = firewall.NewFirewall(e.wgInterface, e.stateManager, e.flowManager.GetLogger(), e.config.DisableServerRoutes, e.config.MTU)
|
|
if err != nil {
|
|
return fmt.Errorf("create firewall manager: %w", err)
|
|
}
|
|
if e.firewall == nil {
|
|
return fmt.Errorf("create firewall manager: received nil manager")
|
|
}
|
|
|
|
if err := e.initFirewall(); err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (e *Engine) initFirewall() error {
|
|
if err := e.routeManager.SetFirewall(e.firewall); err != nil {
|
|
return fmt.Errorf("set firewall: %w", err)
|
|
}
|
|
|
|
// TODO: the firewall backends dedup filter rules by content, so a
|
|
// management route ACL with identical content would collapse onto the
|
|
// untracked drop rules installed here, and a later management delete
|
|
// could remove them. Needs backend refcounting or per-consumer key
|
|
// namespacing.
|
|
if e.config.BlockLANAccess {
|
|
e.blockLanAccess()
|
|
}
|
|
|
|
if e.rpManager == nil || !e.config.RosenpassEnabled {
|
|
return nil
|
|
}
|
|
|
|
rosenpassPort := e.rpManager.GetAddress().Port
|
|
port := firewallManager.Port{Values: []uint16{uint16(rosenpassPort)}}
|
|
|
|
// IPv4-only: rosenpass peers connect via AllowedIps[0] which is always v4.
|
|
if _, err := e.firewall.AddFilterRule(
|
|
nil,
|
|
[]netip.Prefix{netip.PrefixFrom(netip.IPv4Unspecified(), 0)},
|
|
firewallManager.Network{},
|
|
firewallManager.ProtocolUDP,
|
|
nil,
|
|
&port,
|
|
firewallManager.ActionAccept,
|
|
); err != nil {
|
|
log.Errorf("failed to allow rosenpass interface traffic: %v", err)
|
|
return nil
|
|
}
|
|
|
|
log.Infof("rosenpass interface traffic allowed on port %d", rosenpassPort)
|
|
|
|
return nil
|
|
}
|
|
|
|
func (e *Engine) blockLanAccess() {
|
|
if e.config.BlockInbound {
|
|
// no need to set up extra deny rules if inbound is already blocked in general
|
|
return
|
|
}
|
|
|
|
var merr *multierror.Error
|
|
|
|
// TODO: keep this updated
|
|
toBlock, err := getInterfacePrefixes()
|
|
if err != nil {
|
|
merr = multierror.Append(merr, fmt.Errorf("get local addresses: %w", err))
|
|
}
|
|
|
|
log.Infof("blocking route LAN access for networks: %v", toBlock)
|
|
v4 := netip.PrefixFrom(netip.IPv4Unspecified(), 0)
|
|
v6 := netip.PrefixFrom(netip.IPv6Unspecified(), 0)
|
|
for _, network := range toBlock {
|
|
source := v4
|
|
if network.Addr().Is6() {
|
|
source = v6
|
|
}
|
|
if _, err := e.firewall.AddFilterRule(
|
|
nil,
|
|
[]netip.Prefix{source},
|
|
firewallManager.Network{Prefix: network},
|
|
firewallManager.ProtocolALL,
|
|
nil,
|
|
nil,
|
|
firewallManager.ActionDrop,
|
|
); err != nil {
|
|
merr = multierror.Append(merr, fmt.Errorf("add fw rule for network %s: %w", network, err))
|
|
}
|
|
}
|
|
|
|
if merr != nil {
|
|
log.Warnf("encountered errors blocking IPs to block LAN access: %v", nberrors.FormatErrorOrNil(merr))
|
|
}
|
|
}
|
|
|
|
// modifyPeers updates peers that have been modified (e.g. IP address has been changed).
|
|
// It closes the existing connection, removes it from the peerConns map, and creates a new one.
|
|
func (e *Engine) modifyPeers(peersUpdate []*mgmProto.RemotePeerConfig) error {
|
|
|
|
// first, check if peers have been modified
|
|
var modified []*mgmProto.RemotePeerConfig
|
|
for _, p := range peersUpdate {
|
|
peerPubKey := p.GetWgPubKey()
|
|
currentPeer, ok := e.peerStore.PeerConn(peerPubKey)
|
|
if !ok {
|
|
continue
|
|
}
|
|
|
|
if currentPeer.AgentVersionString() != p.AgentVersion {
|
|
modified = append(modified, p)
|
|
continue
|
|
}
|
|
|
|
allowedIPs, ok := e.peerStore.AllowedIPs(peerPubKey)
|
|
if !ok {
|
|
continue
|
|
}
|
|
if !compareNetIPLists(allowedIPs, e.filterAllowedIPs(p.GetAllowedIps())) {
|
|
modified = append(modified, p)
|
|
continue
|
|
}
|
|
|
|
if err := e.statusRecorder.UpdatePeerFQDN(peerPubKey, p.GetFqdn()); err != nil {
|
|
log.Warnf("error updating peer's %s fqdn in the status recorder, got error: %v", peerPubKey, err)
|
|
}
|
|
}
|
|
|
|
// second, close all modified connections and remove them from the state map
|
|
for _, p := range modified {
|
|
err := e.removePeer(p.GetWgPubKey())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
// third, add the peer connections again
|
|
for _, p := range modified {
|
|
if err := e.addNewPeer(p); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// removePeers finds and removes peers that do not exist anymore in the network map received from the Management Service.
|
|
// It also removes peers that have been modified (e.g. change of IP address). They will be added again in addPeers method.
|
|
func (e *Engine) removePeers(peersUpdate []*mgmProto.RemotePeerConfig) error {
|
|
newPeers := make([]string, 0, len(peersUpdate))
|
|
for _, p := range peersUpdate {
|
|
newPeers = append(newPeers, p.GetWgPubKey())
|
|
}
|
|
|
|
toRemove := util.SliceDiff(e.peerStore.PeersPubKey(), newPeers)
|
|
|
|
for _, p := range toRemove {
|
|
err := e.removePeer(p)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
log.Infof("removed peer %s", p)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (e *Engine) removeAllPeers() error {
|
|
log.Debugf("removing all peer connections")
|
|
for _, p := range e.peerStore.PeersPubKey() {
|
|
err := e.removePeer(p)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// removePeer closes an existing peer connection and removes a peer
|
|
func (e *Engine) removePeer(peerKey string) error {
|
|
log.Debugf("removing peer from engine %s", peerKey)
|
|
|
|
e.connMgr.RemovePeerConn(peerKey)
|
|
|
|
err := e.statusRecorder.RemovePeer(peerKey)
|
|
if err != nil {
|
|
log.Warnf("received error when removing peer %s from status recorder: %v", peerKey, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// PopulateNetbirdConfig populates the DNS cache with infrastructure domains from login response
|
|
func (e *Engine) PopulateNetbirdConfig(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL) error {
|
|
if e.dnsServer == nil {
|
|
return nil
|
|
}
|
|
|
|
// Populate management URL if provided
|
|
if mgmtURL != nil {
|
|
if err := e.dnsServer.PopulateManagementDomain(mgmtURL); err != nil {
|
|
log.Warnf("failed to populate DNS cache with management URL: %v", err)
|
|
}
|
|
}
|
|
|
|
// Populate NetbirdConfig domains if provided
|
|
if netbirdConfig != nil {
|
|
serverDomains := dnsconfig.ExtractFromNetbirdConfig(netbirdConfig)
|
|
if err := e.dnsServer.UpdateServerConfig(serverDomains); err != nil {
|
|
return fmt.Errorf("update DNS server config from NetbirdConfig: %w", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (e *Engine) handleAutoUpdateVersion(autoUpdateSettings *mgmProto.AutoUpdateSettings) {
|
|
if e.updateManager == nil {
|
|
return
|
|
}
|
|
|
|
if autoUpdateSettings == nil {
|
|
log.Infof("no auto-update settings received, defaulting to download-only")
|
|
e.updateManager.SetDownloadOnly()
|
|
return
|
|
}
|
|
|
|
if autoUpdateSettings.Version == disableAutoUpdate {
|
|
log.Infof("auto-update is disabled, switching to download-only")
|
|
e.updateManager.SetDownloadOnly()
|
|
return
|
|
}
|
|
|
|
e.updateManager.SetVersion(autoUpdateSettings.Version, autoUpdateSettings.AlwaysUpdate)
|
|
}
|
|
|
|
// phase times a sync sub-phase: it returns a function that records the elapsed
|
|
// duration when called. Starting the timer at the call site keeps inter-phase
|
|
// glue code out of the measurement.
|
|
func (e *Engine) phase(name string) func() {
|
|
start := time.Now()
|
|
return func() {
|
|
e.clientMetrics.RecordSyncPhase(e.ctx, name, time.Since(start))
|
|
}
|
|
}
|
|
|
|
func (e *Engine) handleSync(update *mgmProto.SyncResponse) error {
|
|
started := time.Now()
|
|
defer func() {
|
|
duration := time.Since(started)
|
|
log.Infof("sync finished in %s", duration)
|
|
e.clientMetrics.RecordSyncDuration(e.ctx, duration)
|
|
}()
|
|
e.syncMsgMux.Lock()
|
|
defer e.syncMsgMux.Unlock()
|
|
|
|
// Check context INSIDE lock to ensure atomicity with shutdown
|
|
if e.ctx.Err() != nil {
|
|
return e.ctx.Err()
|
|
}
|
|
|
|
e.ApplySessionDeadline(update.GetSessionExpiresAt())
|
|
|
|
// Envelope sync responses carry PeerConfig at the top level; legacy
|
|
// NetworkMap syncs carry it under NetworkMap.PeerConfig.
|
|
if pc := update.GetPeerConfig(); pc != nil {
|
|
e.handleAutoUpdateVersion(pc.GetAutoUpdate())
|
|
} else if nm := update.GetNetworkMap(); nm != nil && nm.GetPeerConfig() != nil {
|
|
e.handleAutoUpdateVersion(nm.GetPeerConfig().GetAutoUpdate())
|
|
}
|
|
|
|
done := e.phase("netbird_config")
|
|
err := e.updateNetbirdConfig(update.GetNetbirdConfig())
|
|
done()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Decode the network map from either the components envelope or the
|
|
// legacy proto.NetworkMap before the posture-check gating below, so the
|
|
// "is there a network map" decision covers both wire shapes.
|
|
var (
|
|
nm *mgmProto.NetworkMap
|
|
components *types.NetworkMapComponents
|
|
)
|
|
if version := update.GetVersion(); version == int32(sharedgrpc.ComponentNetworkMap) {
|
|
// Components-format peer: decode the envelope back to typed
|
|
// components, run Calculate() locally, and convert to the wire
|
|
// NetworkMap shape the rest of the engine consumes. Components are
|
|
// retained so future incremental updates can apply deltas instead
|
|
// of doing a full reconstruction.
|
|
envelope := update.GetNetworkMapEnvelope()
|
|
if envelope == nil {
|
|
return fmt.Errorf("received a SyncReponse indicating use of components network map, but components are missing")
|
|
}
|
|
|
|
localKey := e.config.WgPrivateKey.PublicKey().String()
|
|
dnsName := ""
|
|
if pc := update.GetPeerConfig(); pc != nil {
|
|
// PeerConfig.Fqdn = "<dns_label>.<dns_domain>" — extract the
|
|
// shared domain by stripping the peer's own label prefix. Falls
|
|
// back to empty if the FQDN doesn't have the expected shape.
|
|
dnsName = extractDNSDomainFromFQDN(pc.GetFqdn())
|
|
}
|
|
// With the firewall disabled there is no ACL manager to program, so
|
|
// RoutesFirewallRules would be built and then dropped. On a peer that
|
|
// routes many network resources that is the single most expensive
|
|
// step of the sync.
|
|
result, err := nbnetworkmap.EnvelopeToNetworkMap(e.ctx, envelope, localKey, dnsName, e.config.DisableFirewall)
|
|
if err != nil {
|
|
return fmt.Errorf("decode network map envelope: %w", err)
|
|
}
|
|
nm = result.NetworkMap
|
|
components = result.Components
|
|
} else {
|
|
nm = update.GetNetworkMap()
|
|
}
|
|
|
|
// Posture checks are bound to the network map presence:
|
|
// NetworkMap != nil, checks present -> apply the received checks
|
|
// NetworkMap != nil, checks nil -> posture checks were removed, clear them
|
|
// NetworkMap == nil -> config-only update (e.g. relay token rotation),
|
|
// leave the previously applied checks untouched
|
|
if nm == nil {
|
|
return nil
|
|
}
|
|
|
|
done = e.phase("checks")
|
|
err = e.updateChecksIfNew(update.Checks)
|
|
done()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
done = e.phase("persist")
|
|
// Only retain the components view when the server sent the envelope
|
|
// path. A legacy proto.NetworkMap means components == nil; writing it
|
|
// here would clobber a previously-cached snapshot, breaking the
|
|
// incremental-delta base on a future envelope sync.
|
|
if components != nil {
|
|
e.latestComponents = components
|
|
}
|
|
|
|
e.persistSyncResponse(update)
|
|
done()
|
|
|
|
// only apply new changes and ignore old ones
|
|
if err := e.updateNetworkMap(nm); err != nil {
|
|
return err
|
|
}
|
|
|
|
e.statusRecorder.PublishEvent(cProto.SystemEvent_INFO, cProto.SystemEvent_SYSTEM, "Network map updated", "", nil)
|
|
|
|
return nil
|
|
}
|
|
|
|
// extractDNSDomainFromFQDN returns the trailing dotted domain part of the
|
|
// receiving peer's FQDN — the same value the management server fills as
|
|
// dnsName when it builds the legacy NetworkMap. "peer42.netbird.cloud" →
|
|
// "netbird.cloud". An empty string is returned for unrecognized formats.
|
|
func extractDNSDomainFromFQDN(fqdn string) string {
|
|
for i := 0; i < len(fqdn); i++ {
|
|
if fqdn[i] == '.' && i+1 < len(fqdn) {
|
|
return fqdn[i+1:]
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// updateNetbirdConfig applies the management-provided NetBird configuration:
|
|
// STUN/TURN and relay servers, flow logging and DNS settings. A nil config is a no-op,
|
|
// which is the case for sync updates carrying only a network map.
|
|
func (e *Engine) updateNetbirdConfig(wCfg *mgmProto.NetbirdConfig) error {
|
|
if wCfg == nil {
|
|
return nil
|
|
}
|
|
|
|
if err := e.updateTURNs(wCfg.GetTurns()); err != nil {
|
|
return fmt.Errorf("update TURNs: %w", err)
|
|
}
|
|
|
|
if err := e.updateSTUNs(wCfg.GetStuns()); err != nil {
|
|
return fmt.Errorf("update STUNs: %w", err)
|
|
}
|
|
|
|
var stunTurn []*stun.URI
|
|
stunTurn = append(stunTurn, e.STUNs...)
|
|
stunTurn = append(stunTurn, e.TURNs...)
|
|
e.stunTurn.Store(stunTurn)
|
|
|
|
if err := e.handleRelayUpdate(wCfg.GetRelay()); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := e.handleFlowUpdate(wCfg.GetFlow()); err != nil {
|
|
return fmt.Errorf("handle the flow configuration: %w", err)
|
|
}
|
|
|
|
e.handleMetricsUpdate(wCfg.GetMetrics())
|
|
|
|
if err := e.PopulateNetbirdConfig(wCfg, nil); err != nil {
|
|
log.Warnf("Failed to update DNS server config: %v", err)
|
|
}
|
|
|
|
// todo update signal
|
|
|
|
return nil
|
|
}
|
|
|
|
// persistSyncResponse stores the full sync response so it can be restored on the next
|
|
// startup. Persistence is enabled only when syncStore is set. The dedicated syncRespMux
|
|
// (not syncMsgMux) is held 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.
|
|
func (e *Engine) persistSyncResponse(update *mgmProto.SyncResponse) {
|
|
e.syncRespMux.RLock()
|
|
defer e.syncRespMux.RUnlock()
|
|
|
|
if e.syncStore == nil {
|
|
return
|
|
}
|
|
|
|
if err := e.syncStore.Set(update); err != nil {
|
|
log.Errorf("failed to persist sync response: %v", err)
|
|
return
|
|
}
|
|
|
|
log.Debugf("sync response persisted with serial %d", update.GetNetworkMap().GetSerial())
|
|
}
|
|
|
|
func (e *Engine) handleRelayUpdate(update *mgmProto.RelayConfig) error {
|
|
if update != nil {
|
|
// when we receive token we expect valid address list too
|
|
c := &auth.Token{
|
|
Payload: update.GetTokenPayload(),
|
|
Signature: update.GetTokenSignature(),
|
|
}
|
|
if err := e.relayManager.UpdateToken(c); err != nil {
|
|
return fmt.Errorf("update relay token: %w", err)
|
|
}
|
|
|
|
urls := update.Urls
|
|
if override, ok := peer.OverrideRelayURLs(); ok {
|
|
log.Infof("overriding relay URLs from %s: %v", peer.EnvKeyNBHomeRelayServers, override)
|
|
urls = override
|
|
}
|
|
e.relayManager.UpdateServerURLs(urls)
|
|
|
|
// Just in case the agent started with an MGM server where the relay was disabled but was later enabled.
|
|
// We can ignore all errors because the guard will manage the reconnection retries.
|
|
_ = e.relayManager.Serve()
|
|
} else {
|
|
e.relayManager.UpdateServerURLs(nil)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (e *Engine) handleFlowUpdate(config *mgmProto.FlowConfig) error {
|
|
if config == nil {
|
|
return nil
|
|
}
|
|
|
|
flowConfig, err := toFlowLoggerConfig(config)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return e.flowManager.Update(flowConfig)
|
|
}
|
|
|
|
func (e *Engine) handleMetricsUpdate(config *mgmProto.MetricsConfig) {
|
|
if config == nil {
|
|
return
|
|
}
|
|
log.Infof("received metrics configuration from management: enabled=%v", config.GetEnabled())
|
|
e.clientMetrics.UpdatePushFromMgm(e.metricsCtx, config.GetEnabled())
|
|
}
|
|
|
|
func toFlowLoggerConfig(config *mgmProto.FlowConfig) (*nftypes.FlowConfig, error) {
|
|
if config.GetInterval() == nil {
|
|
return nil, errors.New("flow interval is nil")
|
|
}
|
|
return &nftypes.FlowConfig{
|
|
Enabled: config.GetEnabled(),
|
|
Counters: config.GetCounters(),
|
|
URL: config.GetUrl(),
|
|
TokenPayload: config.GetTokenPayload(),
|
|
TokenSignature: config.GetTokenSignature(),
|
|
Interval: config.GetInterval().AsDuration(),
|
|
DNSCollection: config.GetDnsCollection(),
|
|
ExitNodeCollection: config.GetExitNodeCollection(),
|
|
}, nil
|
|
}
|
|
|
|
// updateChecksIfNew updates checks if there are changes and sync new meta with management
|
|
func (e *Engine) updateChecksIfNew(checks []*mgmProto.Checks) error {
|
|
// if checks are equal, we skip the update
|
|
if isChecksEqual(e.checks, checks) {
|
|
e.clearPendingChecks()
|
|
return nil
|
|
}
|
|
if err := e.syncChecksMeta(checks); err != nil {
|
|
// The newest checks become the pending ones whatever failed, so the watcher
|
|
// never retries a set they superseded.
|
|
e.pendingChecks = checks
|
|
e.hasPendingChecks.Store(true)
|
|
if errors.Is(err, errSystemInfoTimeout) {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
e.setAppliedChecks(checks)
|
|
e.clearPendingChecks()
|
|
return nil
|
|
}
|
|
|
|
// appliedChecks returns the posture checks in effect, for callers that do not hold
|
|
// syncMsgMux. The slice is replaced, never modified, so it may be read after return.
|
|
func (e *Engine) appliedChecks() []*mgmProto.Checks {
|
|
e.checksMu.RLock()
|
|
defer e.checksMu.RUnlock()
|
|
return e.checks
|
|
}
|
|
|
|
// setAppliedChecks replaces the posture checks in effect. The caller holds syncMsgMux.
|
|
func (e *Engine) setAppliedChecks(checks []*mgmProto.Checks) {
|
|
e.checksMu.Lock()
|
|
e.checks = checks
|
|
e.checksMu.Unlock()
|
|
e.wakeCertificatePosture()
|
|
}
|
|
|
|
// clearPendingChecks drops checks whose meta sync was still owed. The caller holds
|
|
// syncMsgMux.
|
|
func (e *Engine) clearPendingChecks() {
|
|
e.pendingChecks = nil
|
|
e.hasPendingChecks.Store(false)
|
|
}
|
|
|
|
func (e *Engine) infoGatherTimeout() time.Duration {
|
|
if e.infoTimeout > 0 {
|
|
return e.infoTimeout
|
|
}
|
|
return systemInfoTimeout
|
|
}
|
|
|
|
// applyInfoFlags sets the engine's config-derived feature flags on the gathered system info.
|
|
func (e *Engine) applyInfoFlags(info *system.Info) {
|
|
info.SetFlags(
|
|
e.config.RosenpassEnabled,
|
|
e.config.RosenpassPermissive,
|
|
&e.config.ServerSSHAllowed,
|
|
e.config.DisableClientRoutes,
|
|
e.config.DisableServerRoutes,
|
|
e.config.DisableDNS,
|
|
e.config.DisableFirewall,
|
|
e.config.BlockLANAccess,
|
|
e.config.BlockInbound,
|
|
e.config.DisableIPv6,
|
|
e.config.SyncMessageVersion,
|
|
e.config.EnableSSHRoot,
|
|
e.config.EnableSSHSFTP,
|
|
e.config.EnableSSHLocalPortForwarding,
|
|
e.config.EnableSSHRemotePortForwarding,
|
|
e.config.DisableSSHAuth,
|
|
&e.config.RemoteJobsAllowed,
|
|
)
|
|
}
|
|
|
|
func (e *Engine) currentSystemInfo(ctx context.Context) *system.Info {
|
|
info := e.infoSource.Current(ctx, e.overlayAddresses()...)
|
|
e.applyInfoFlags(info)
|
|
e.attachCertificateProofs(info, e.appliedChecks())
|
|
return info
|
|
}
|
|
|
|
// syncInfoFunc returns the info callback for the management sync stream. The
|
|
// first connect sends the info refreshed right before it instead of gathering
|
|
// again; every reconnect gathers a fresh one. The stream retry loop calls the
|
|
// callback sequentially, so the handoff needs no synchronization.
|
|
func (e *Engine) syncInfoFunc(refreshed *system.Info) func(ctx context.Context) *system.Info {
|
|
return func(ctx context.Context) *system.Info {
|
|
if refreshed == nil {
|
|
return e.currentSystemInfo(ctx)
|
|
}
|
|
info := refreshed
|
|
refreshed = nil
|
|
e.applyInfoFlags(info)
|
|
e.attachCertificateProofs(info, e.appliedChecks())
|
|
return info
|
|
}
|
|
}
|
|
|
|
// overlayAddresses returns our own WireGuard overlay address (v4 and v6) so it
|
|
// can be excluded from the reported network addresses; the interface coming and
|
|
// going otherwise churns the peer meta on the management server.
|
|
func (e *Engine) overlayAddresses() []netip.Addr {
|
|
var ips []netip.Addr
|
|
if e.config.WgAddr.IP.IsValid() {
|
|
ips = append(ips, e.config.WgAddr.IP)
|
|
}
|
|
if e.config.WgAddr.HasIPv6() {
|
|
ips = append(ips, e.config.WgAddr.IPv6)
|
|
}
|
|
return ips
|
|
}
|
|
|
|
func (e *Engine) updateConfig(conf *mgmProto.PeerConfig) error {
|
|
if e.wgInterface == nil {
|
|
return errors.New("wireguard interface is not initialized")
|
|
}
|
|
|
|
if e.wgInterface.Address().String() != conf.Address {
|
|
log.Infof("peer IP address changed from %s to %s, restarting client", e.wgInterface.Address().String(), conf.Address)
|
|
_ = CtxGetState(e.ctx).Wrap(ErrResetConnection)
|
|
e.clientCancel()
|
|
return ErrResetConnection
|
|
}
|
|
|
|
if !e.config.DisableIPv6 && e.hasIPv6Changed(conf) {
|
|
log.Infof("peer IPv6 address changed, restarting client")
|
|
_ = CtxGetState(e.ctx).Wrap(ErrResetConnection)
|
|
e.clientCancel()
|
|
return ErrResetConnection
|
|
}
|
|
|
|
if conf.GetSshConfig() != nil {
|
|
if err := e.updateSSH(conf.GetSshConfig()); err != nil {
|
|
log.Warnf("failed handling SSH server setup: %v", err)
|
|
}
|
|
}
|
|
|
|
state := e.statusRecorder.GetLocalPeerState()
|
|
state.IP = e.wgInterface.Address().String()
|
|
state.IPv6 = e.wgInterface.Address().IPv6String()
|
|
state.PubKey = e.config.WgPrivateKey.PublicKey().String()
|
|
state.KernelInterface = !e.wgInterface.IsUserspaceBind()
|
|
state.FQDN = conf.GetFqdn()
|
|
state.WgPort = e.config.WgPort
|
|
|
|
e.statusRecorder.UpdateLocalPeerState(state)
|
|
|
|
return nil
|
|
}
|
|
|
|
// hasIPv6Changed reports whether the IPv6 overlay address in the peer config
|
|
// differs from the configured address (added, removed, or changed).
|
|
// Compares against e.config.WgAddr (not the interface address, which may have
|
|
// been cleared by ClearIPv6 if OS assignment failed).
|
|
func (e *Engine) hasIPv6Changed(conf *mgmProto.PeerConfig) bool {
|
|
current := e.config.WgAddr
|
|
raw := conf.GetAddressV6()
|
|
|
|
if len(raw) == 0 {
|
|
return current.HasIPv6()
|
|
}
|
|
|
|
prefix, err := netiputil.DecodePrefix(raw)
|
|
if err != nil {
|
|
log.Errorf("decode v6 overlay address: %v", err)
|
|
return false
|
|
}
|
|
|
|
return !current.HasIPv6() || current.IPv6 != prefix.Addr() || current.IPv6Net != prefix.Masked()
|
|
}
|
|
|
|
func (e *Engine) receiveJobEvents() {
|
|
e.jobExecutorWG.Add(1)
|
|
go func() {
|
|
defer e.jobExecutorWG.Done()
|
|
err := e.mgmClient.Job(e.ctx, func(msg *mgmProto.JobRequest) *mgmProto.JobResponse {
|
|
resp := mgmProto.JobResponse{
|
|
ID: msg.ID,
|
|
Status: mgmProto.JobStatus_failed,
|
|
}
|
|
// Remote jobs are an explicit opt-in. When not enabled on this
|
|
// peer, every job is refused before any work is done.
|
|
if !e.config.RemoteJobsAllowed {
|
|
log.Warnf("refusing remote job: remote jobs are not enabled on this peer (enable with --allow-remote-jobs)")
|
|
resp.Reason = []byte("remote jobs are not enabled on this peer")
|
|
return &resp
|
|
}
|
|
switch params := msg.WorkloadParameters.(type) {
|
|
case *mgmProto.JobRequest_Bundle:
|
|
bundleResult, err := e.handleBundle(params.Bundle)
|
|
if err != nil {
|
|
log.Errorf("handling bundle: %v", err)
|
|
resp.Reason = []byte(err.Error())
|
|
return &resp
|
|
}
|
|
resp.Status = mgmProto.JobStatus_succeeded
|
|
resp.WorkloadResults = bundleResult
|
|
return &resp
|
|
default:
|
|
resp.Reason = []byte(jobexec.ErrJobNotImplemented.Error())
|
|
return &resp
|
|
}
|
|
})
|
|
if err != nil {
|
|
// happens if management is unavailable for a long time.
|
|
// We want to cancel the operation of the whole client
|
|
_ = CtxGetState(e.ctx).Wrap(ErrResetConnection)
|
|
e.clientCancel()
|
|
return
|
|
}
|
|
log.Info("stopped receiving jobs from Management Service")
|
|
}()
|
|
log.Info("connecting to Management Service jobs stream")
|
|
}
|
|
|
|
func (e *Engine) handleBundle(params *mgmProto.BundleParameters) (*mgmProto.JobResponse_Bundle, error) {
|
|
// The upload URL can carry a host, credentials, or query tokens, so it is
|
|
// kept out of the info-level line; the full parameters stay available at
|
|
// debug level for troubleshooting.
|
|
log.Infof("handle remote debug bundle request: anonymize=%v anonymize_level=%q log_file_count=%d bundle_for=%v bundle_for_time=%d",
|
|
params.GetAnonymize(), params.GetAnonymizeLevel(), params.GetLogFileCount(), params.GetBundleFor(), params.GetBundleForTime())
|
|
log.Debugf("remote debug bundle request parameters: %s", params.String())
|
|
|
|
// Resolve the upload destination: an MDM override, when set, takes
|
|
// precedence over the management-supplied URL. Both are validated the same
|
|
// way; an empty result falls back to the default upload server downstream.
|
|
uploadURL := params.GetUploadUrl()
|
|
if override := e.config.ProfileConfig.DebugBundleUploadURL; override != "" {
|
|
log.Infof("using MDM debug bundle upload URL override instead of the management-supplied value")
|
|
uploadURL = override
|
|
}
|
|
if err := validateBundleUploadURL(uploadURL); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
syncResponse, err := e.GetLatestSyncResponse()
|
|
if err != nil {
|
|
log.Warnf("get latest sync response: %v", err)
|
|
}
|
|
|
|
bundleDeps := debug.GeneratorDependencies{
|
|
InternalConfig: e.config.ProfileConfig,
|
|
StatusRecorder: e.statusRecorder,
|
|
SyncResponse: syncResponse,
|
|
LogPath: e.config.LogPath,
|
|
TempDir: e.config.TempDir,
|
|
ClientMetrics: e.clientMetrics,
|
|
DaemonVersion: version.NetbirdVersion(),
|
|
RefreshStatus: func() {
|
|
e.RunHealthProbes(e.ctx, true)
|
|
},
|
|
}
|
|
|
|
bundleJobParams := debug.BundleConfig{
|
|
Anonymize: params.Anonymize,
|
|
AnonymizeLevel: anonymize.ParseLevel(params.AnonymizeLevel),
|
|
IncludeSystemInfo: true,
|
|
LogFileCount: uint32(params.LogFileCount),
|
|
}
|
|
|
|
waitFor := time.Duration(params.BundleForTime) * time.Minute
|
|
|
|
uploadKey, err := e.jobExecutor.BundleJob(e.ctx, bundleDeps, bundleJobParams, waitFor, e.config.ProfileConfig.ManagementURL.String(), uploadURL)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
response := &mgmProto.JobResponse_Bundle{
|
|
Bundle: &mgmProto.BundleResult{
|
|
UploadKey: uploadKey,
|
|
},
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
// validateBundleUploadURL sanity-checks a management-supplied upload URL for a
|
|
// remote debug bundle job. It delegates to profilemanager.ValidateBundleUploadURL
|
|
// so the executor and the MDM policy override share one definition of the rule
|
|
// (empty accepted; otherwise a well-formed https URL with a host) and cannot
|
|
// drift. The host is deliberately left unconstrained pending a decision on
|
|
// management-directed uploads.
|
|
func validateBundleUploadURL(raw string) error {
|
|
return profilemanager.ValidateBundleUploadURL(raw)
|
|
}
|
|
|
|
// receiveManagementEvents connects to the Management Service event stream to receive updates from the management service
|
|
// E.g. when a new peer has been registered and we are allowed to connect to it.
|
|
func (e *Engine) receiveManagementEvents() {
|
|
e.shutdownWg.Add(1)
|
|
go func() {
|
|
defer e.shutdownWg.Done()
|
|
info, ok := e.infoSource.Refresh(e.ctx, e.infoGatherTimeout(), e.appliedChecks(), e.overlayAddresses()...)
|
|
if !ok {
|
|
log.Warnf("posture checks not refreshed before the sync connect, sending the previous results")
|
|
}
|
|
err := e.mgmClient.Sync(e.ctx, e.syncInfoFunc(info), e.handleSync)
|
|
if err != nil {
|
|
// happens if management is unavailable for a long time.
|
|
// We want to cancel the operation of the whole client
|
|
_ = CtxGetState(e.ctx).Wrap(ErrResetConnection)
|
|
e.clientCancel()
|
|
return
|
|
}
|
|
log.Debugf("stopped receiving updates from Management Service")
|
|
}()
|
|
log.Infof("connecting to Management Service updates stream")
|
|
}
|
|
|
|
func (e *Engine) updateSTUNs(stuns []*mgmProto.HostConfig) error {
|
|
if len(stuns) == 0 {
|
|
return nil
|
|
}
|
|
var newSTUNs []*stun.URI
|
|
log.Debugf("got STUNs update from Management Service, updating")
|
|
for _, s := range stuns {
|
|
url, err := stun.ParseURI(s.Uri)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
newSTUNs = append(newSTUNs, url)
|
|
}
|
|
e.STUNs = newSTUNs
|
|
|
|
return nil
|
|
}
|
|
|
|
func (e *Engine) updateTURNs(turns []*mgmProto.ProtectedHostConfig) error {
|
|
if len(turns) == 0 {
|
|
return nil
|
|
}
|
|
var newTURNs []*stun.URI
|
|
log.Debugf("got TURNs update from Management Service, updating")
|
|
for _, turn := range turns {
|
|
url, err := stun.ParseURI(turn.HostConfig.Uri)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
url.Username = turn.User
|
|
url.Password = turn.Password
|
|
newTURNs = append(newTURNs, url)
|
|
}
|
|
e.TURNs = newTURNs
|
|
|
|
return nil
|
|
}
|
|
|
|
func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap) error {
|
|
// intentionally leave it before checking serial because for now it can happen that peer IP changed but serial didn't
|
|
if networkMap.GetPeerConfig() != nil {
|
|
err := e.updateConfig(networkMap.GetPeerConfig())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
serial := networkMap.GetSerial()
|
|
if e.networkSerial > serial {
|
|
log.Debugf("received outdated NetworkMap with serial %d, ignoring", serial)
|
|
return nil
|
|
}
|
|
|
|
// Only update the flag when the sync carries a peer config; a nil peer config
|
|
// (e.g. a partial update) must not reset the cached flag to false.
|
|
if peerConfig := networkMap.GetPeerConfig(); peerConfig != nil {
|
|
if err := e.connMgr.UpdatedRemoteFeatureFlag(e.ctx, peerConfig.GetLazyConnectionEnabled()); err != nil {
|
|
log.Errorf("failed to update lazy connection feature flag: %v", err)
|
|
}
|
|
}
|
|
|
|
if e.firewall != nil {
|
|
if localipfw, ok := e.firewall.(localIpUpdater); ok {
|
|
if err := localipfw.UpdateLocalIPs(); err != nil {
|
|
log.Errorf("failed to update local IPs: %v", err)
|
|
}
|
|
}
|
|
|
|
// If we got empty rules list but management did not set the networkMap.FirewallRulesIsEmpty flag,
|
|
// then the mgmt server is older than the client, and we need to allow all traffic for routes.
|
|
// This needs to be toggled before applying routes.
|
|
isLegacy := len(networkMap.RoutesFirewallRules) == 0 && !networkMap.RoutesFirewallRulesIsEmpty
|
|
if err := e.firewall.SetLegacyManagement(isLegacy); err != nil {
|
|
log.Errorf("failed to set legacy management flag: %v", err)
|
|
}
|
|
}
|
|
|
|
protoDNSConfig := networkMap.GetDNSConfig()
|
|
if protoDNSConfig == nil {
|
|
protoDNSConfig = &mgmProto.DNSConfig{}
|
|
}
|
|
|
|
dnsConfig := toDNSConfig(protoDNSConfig, e.wgInterface.Address())
|
|
|
|
done := e.phase("dns_server")
|
|
if err := e.dnsServer.UpdateDNSServer(serial, dnsConfig); err != nil {
|
|
log.Errorf("failed to update dns server, err: %v", err)
|
|
}
|
|
done()
|
|
|
|
e.routeManager.SetDNSForwarderPort(dnsConfig.ForwarderPort)
|
|
|
|
// apply routes first, route related actions might depend on routing being enabled
|
|
done = e.phase("routes_classify")
|
|
routes := toRoutes(networkMap.GetRoutes())
|
|
serverRoutes, clientRoutes := e.routeManager.ClassifyRoutes(routes)
|
|
|
|
// lazy mgr needs to be aware of which routes are available before they are applied
|
|
if e.connMgr != nil {
|
|
e.connMgr.UpdateRouteHAMap(clientRoutes)
|
|
log.Debugf("updated lazy connection manager with %d HA groups", len(clientRoutes))
|
|
}
|
|
done()
|
|
|
|
done = e.phase("routes_apply")
|
|
dnsRouteFeatureFlag := toDNSFeatureFlag(networkMap)
|
|
if err := e.routeManager.UpdateRoutes(serial, serverRoutes, clientRoutes, dnsRouteFeatureFlag); err != nil {
|
|
log.Errorf("failed to update routes: %v", err)
|
|
}
|
|
done()
|
|
|
|
done = e.phase("filtering")
|
|
if e.acl != nil {
|
|
e.acl.ApplyFiltering(networkMap, dnsRouteFeatureFlag)
|
|
}
|
|
done()
|
|
|
|
done = e.phase("dns_forwarder")
|
|
fwdEntries := toRouteDomains(e.config.WgPrivateKey.PublicKey().String(), routes)
|
|
e.updateDNSForwarder(dnsRouteFeatureFlag, fwdEntries)
|
|
done()
|
|
|
|
log.Debugf("got peers update from Management Service, total peers to connect to = %d", len(networkMap.GetRemotePeers()))
|
|
|
|
done = e.phase("offline_peers")
|
|
e.updateOfflinePeers(networkMap.GetOfflinePeers())
|
|
done()
|
|
|
|
remotePeers, err := e.reconcilePeers(networkMap)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// must set the exclude list after the peers are added. Without it the manager can not figure out the peers parameters from the store
|
|
done = e.phase("lazy_exclude")
|
|
e.connMgr.SetExcludeList(e.ctx, e.toExcludedLazyPeers(remotePeers))
|
|
done()
|
|
|
|
e.networkSerial = serial
|
|
|
|
return nil
|
|
}
|
|
|
|
// reconcilePeers applies the remote peer list from the network map (removing,
|
|
// modifying and adding peers, then updating SSH config) and returns the remote
|
|
// peers with our own peer filtered out, for use by later sync steps.
|
|
func (e *Engine) reconcilePeers(networkMap *mgmProto.NetworkMap) ([]*mgmProto.RemotePeerConfig, error) {
|
|
// Filter out own peer from the remote peers list
|
|
localPubKey := e.config.WgPrivateKey.PublicKey().String()
|
|
remotePeers := make([]*mgmProto.RemotePeerConfig, 0, len(networkMap.GetRemotePeers()))
|
|
for _, p := range networkMap.GetRemotePeers() {
|
|
if p.GetWgPubKey() != localPubKey {
|
|
remotePeers = append(remotePeers, p)
|
|
}
|
|
}
|
|
|
|
// cleanup request, most likely our peer has been deleted
|
|
if networkMap.GetRemotePeersIsEmpty() {
|
|
err := e.removeAllPeers()
|
|
e.statusRecorder.FinishPeerListModifications()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return remotePeers, nil
|
|
}
|
|
|
|
done := e.phase("removed_peers")
|
|
err := e.removePeers(remotePeers)
|
|
done()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
done = e.phase("modified_peers")
|
|
err = e.modifyPeers(remotePeers)
|
|
done()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
done = e.phase("added_peers")
|
|
err = e.addNewPeers(remotePeers)
|
|
done()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
e.statusRecorder.FinishPeerListModifications()
|
|
|
|
e.updatePeerSSHHostKeys(remotePeers)
|
|
|
|
if err := e.updateSSHClientConfig(remotePeers); err != nil {
|
|
log.Warnf("failed to update SSH client config: %v", err)
|
|
}
|
|
|
|
e.updateSSHServerAuth(networkMap.GetSshAuth())
|
|
|
|
return remotePeers, nil
|
|
}
|
|
|
|
func toDNSFeatureFlag(networkMap *mgmProto.NetworkMap) bool {
|
|
if networkMap.PeerConfig != nil {
|
|
return networkMap.PeerConfig.RoutingPeerDnsResolutionEnabled
|
|
}
|
|
return false
|
|
}
|
|
|
|
func toRoutes(protoRoutes []*mgmProto.Route) []*route.Route {
|
|
if protoRoutes == nil {
|
|
protoRoutes = []*mgmProto.Route{}
|
|
}
|
|
|
|
routes := make([]*route.Route, 0)
|
|
for _, protoRoute := range protoRoutes {
|
|
var prefix netip.Prefix
|
|
if len(protoRoute.Domains) == 0 {
|
|
var err error
|
|
if prefix, err = netip.ParsePrefix(protoRoute.Network); err != nil {
|
|
log.Errorf("Failed to parse prefix %s: %v", protoRoute.Network, err)
|
|
continue
|
|
}
|
|
}
|
|
|
|
convertedRoute := &route.Route{
|
|
ID: route.ID(protoRoute.ID),
|
|
Network: prefix.Masked(),
|
|
Domains: domain.FromPunycodeList(protoRoute.Domains),
|
|
NetID: route.NetID(protoRoute.NetID),
|
|
NetworkType: route.NetworkType(protoRoute.NetworkType),
|
|
Peer: protoRoute.Peer,
|
|
Metric: int(protoRoute.Metric),
|
|
Masquerade: protoRoute.Masquerade,
|
|
KeepRoute: protoRoute.KeepRoute,
|
|
SkipAutoApply: protoRoute.SkipAutoApply,
|
|
}
|
|
routes = append(routes, convertedRoute)
|
|
}
|
|
return routes
|
|
}
|
|
|
|
func toRouteDomains(myPubKey string, routes []*route.Route) []*dnsfwd.ForwarderEntry {
|
|
var entries []*dnsfwd.ForwarderEntry
|
|
for _, route := range routes {
|
|
if len(route.Domains) == 0 {
|
|
continue
|
|
}
|
|
if route.Peer == myPubKey {
|
|
domainSet := firewallManager.NewDomainSet(route.Domains)
|
|
for _, d := range route.Domains {
|
|
entries = append(entries, &dnsfwd.ForwarderEntry{
|
|
Domain: d,
|
|
Set: domainSet,
|
|
ResID: route.GetResourceID(),
|
|
})
|
|
}
|
|
}
|
|
}
|
|
return entries
|
|
}
|
|
|
|
func toDNSConfig(protoDNSConfig *mgmProto.DNSConfig, addr wgaddr.Address) nbdns.Config {
|
|
network := addr.Network
|
|
networkV6 := addr.IPv6Net
|
|
//nolint
|
|
forwarderPort := uint16(protoDNSConfig.GetForwarderPort())
|
|
if forwarderPort == 0 {
|
|
forwarderPort = nbdns.ForwarderClientPort
|
|
}
|
|
|
|
dnsUpdate := nbdns.Config{
|
|
ServiceEnable: protoDNSConfig.GetServiceEnable(),
|
|
CustomZones: make([]nbdns.CustomZone, 0),
|
|
NameServerGroups: make([]*nbdns.NameServerGroup, 0),
|
|
ForwarderPort: forwarderPort,
|
|
}
|
|
|
|
protoZones := protoDNSConfig.GetCustomZones()
|
|
// Treat single zone as authoritative for backward compatibility with old servers
|
|
// that only send the peer FQDN zone without setting field 4.
|
|
singleZoneCompat := len(protoZones) == 1
|
|
|
|
for _, zone := range protoZones {
|
|
dnsZone := nbdns.CustomZone{
|
|
Domain: zone.GetDomain(),
|
|
SearchDomainDisabled: zone.GetSearchDomainDisabled(),
|
|
NonAuthoritative: zone.GetNonAuthoritative() && !singleZoneCompat,
|
|
}
|
|
for _, record := range zone.Records {
|
|
dnsRecord := nbdns.SimpleRecord{
|
|
Name: record.GetName(),
|
|
Type: int(record.GetType()),
|
|
Class: record.GetClass(),
|
|
TTL: int(record.GetTTL()),
|
|
RData: record.GetRData(),
|
|
}
|
|
dnsZone.Records = append(dnsZone.Records, dnsRecord)
|
|
}
|
|
dnsUpdate.CustomZones = append(dnsUpdate.CustomZones, dnsZone)
|
|
}
|
|
|
|
for _, nsGroup := range protoDNSConfig.GetNameServerGroups() {
|
|
dnsNSGroup := &nbdns.NameServerGroup{
|
|
Primary: nsGroup.GetPrimary(),
|
|
Domains: nsGroup.GetDomains(),
|
|
SearchDomainsEnabled: nsGroup.GetSearchDomainsEnabled(),
|
|
}
|
|
for _, ns := range nsGroup.GetNameServers() {
|
|
dnsNS := nbdns.NameServer{
|
|
IP: netip.MustParseAddr(ns.GetIP()),
|
|
NSType: nbdns.NameServerType(ns.GetNSType()),
|
|
Port: int(ns.GetPort()),
|
|
}
|
|
dnsNSGroup.NameServers = append(dnsNSGroup.NameServers, dnsNS)
|
|
}
|
|
dnsUpdate.NameServerGroups = append(dnsUpdate.NameServerGroups, dnsNSGroup)
|
|
}
|
|
|
|
if len(dnsUpdate.CustomZones) > 0 {
|
|
addReverseZone(&dnsUpdate, network)
|
|
if networkV6.IsValid() {
|
|
addReverseZone(&dnsUpdate, networkV6)
|
|
}
|
|
}
|
|
|
|
return dnsUpdate
|
|
}
|
|
|
|
func (e *Engine) updateOfflinePeers(offlinePeers []*mgmProto.RemotePeerConfig) {
|
|
replacement := make([]peer.State, len(offlinePeers))
|
|
for i, offlinePeer := range offlinePeers {
|
|
log.Debugf("added offline peer %s", offlinePeer.Fqdn)
|
|
v4, v6 := overlayAddrsFromAllowedIPs(offlinePeer.GetAllowedIps(), e.wgInterface.Address().IPv6Net)
|
|
replacement[i] = peer.State{
|
|
IP: addrToString(v4),
|
|
IPv6: addrToString(v6),
|
|
PubKey: offlinePeer.GetWgPubKey(),
|
|
FQDN: offlinePeer.GetFqdn(),
|
|
ConnStatus: peer.StatusIdle,
|
|
ConnStatusUpdate: time.Now(),
|
|
Mux: new(sync.RWMutex),
|
|
}
|
|
}
|
|
e.statusRecorder.ReplaceOfflinePeers(replacement)
|
|
}
|
|
|
|
// overlayAddrsFromAllowedIPs extracts the peer's v4 and v6 overlay addresses
|
|
// from AllowedIPs strings. Only host routes (/32, /128) are considered; v6 must
|
|
// fall within ourV6Net to distinguish overlay addresses from routed prefixes.
|
|
func overlayAddrsFromAllowedIPs(allowedIPs []string, ourV6Net netip.Prefix) (v4, v6 netip.Addr) {
|
|
for _, cidr := range allowedIPs {
|
|
prefix, err := netip.ParsePrefix(cidr)
|
|
if err != nil {
|
|
log.Warnf("failed to parse AllowedIP %q: %v", cidr, err)
|
|
continue
|
|
}
|
|
addr := prefix.Addr().Unmap()
|
|
switch {
|
|
case addr.Is4() && prefix.Bits() == 32 && !v4.IsValid():
|
|
v4 = addr
|
|
case addr.Is6() && prefix.Bits() == 128 && ourV6Net.Contains(addr) && !v6.IsValid():
|
|
v6 = addr
|
|
}
|
|
if v4.IsValid() && v6.IsValid() {
|
|
break
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
func addrToString(addr netip.Addr) string {
|
|
if !addr.IsValid() {
|
|
return ""
|
|
}
|
|
return addr.String()
|
|
}
|
|
|
|
// addNewPeers adds peers that were not know before but arrived from the Management service with the update
|
|
func (e *Engine) addNewPeers(peersUpdate []*mgmProto.RemotePeerConfig) error {
|
|
for _, p := range peersUpdate {
|
|
if err := e.addNewPeer(p); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// addNewPeer add peer if connection doesn't exist. A peer that is not lazy by
|
|
// policy gets an always-active connection instead.
|
|
func (e *Engine) addNewPeer(peerConfig *mgmProto.RemotePeerConfig) error {
|
|
peerKey := peerConfig.GetWgPubKey()
|
|
peerIPs := make([]netip.Prefix, 0, len(peerConfig.GetAllowedIps()))
|
|
if _, ok := e.peerStore.PeerConn(peerKey); ok {
|
|
return nil
|
|
}
|
|
|
|
for _, ipString := range peerConfig.GetAllowedIps() {
|
|
allowedNetIP, err := netip.ParsePrefix(ipString)
|
|
if err != nil {
|
|
log.Errorf("failed to parse allowedIPS: %v", err)
|
|
return err
|
|
}
|
|
if allowedNetIP.Addr().Is6() && !e.wgInterface.Address().HasIPv6() {
|
|
continue
|
|
}
|
|
peerIPs = append(peerIPs, allowedNetIP)
|
|
}
|
|
|
|
if len(peerIPs) == 0 {
|
|
return fmt.Errorf("peer %s has no usable AllowedIPs", peerKey)
|
|
}
|
|
|
|
conn, err := e.createPeerConn(peerKey, peerIPs, peerConfig.AgentVersion)
|
|
if err != nil {
|
|
return fmt.Errorf("create peer connection: %w", err)
|
|
}
|
|
|
|
peerV4, peerV6 := overlayAddrsFromAllowedIPs(peerConfig.GetAllowedIps(), e.wgInterface.Address().IPv6Net)
|
|
err = e.statusRecorder.AddPeer(peerKey, peerConfig.Fqdn, addrToString(peerV4), addrToString(peerV6))
|
|
if err != nil {
|
|
log.Warnf("error adding peer %s to status recorder, got error: %v", peerKey, err)
|
|
}
|
|
|
|
permanent := !e.connMgr.PeerLazyDefault(peerConfig.GetLazyState())
|
|
if exists := e.connMgr.AddPeerConn(e.ctx, peerKey, conn, permanent); exists {
|
|
conn.Close(false)
|
|
return fmt.Errorf("peer already exists: %s", peerKey)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (e *Engine) createPeerConn(pubKey string, allowedIPs []netip.Prefix, agentVersion string) (*peer.Conn, error) {
|
|
log.Debugf("creating peer connection %s", pubKey)
|
|
|
|
wgConfig := peer.WgConfig{
|
|
RemoteKey: pubKey,
|
|
WgListenPort: e.config.WgPort,
|
|
WgInterface: e.wgInterface,
|
|
AllowedIps: allowedIPs,
|
|
PreSharedKey: e.config.PreSharedKey,
|
|
}
|
|
|
|
// randomize connection timeout
|
|
timeout := time.Duration(rand.Intn(PeerConnectionTimeoutMax-PeerConnectionTimeoutMin)+PeerConnectionTimeoutMin) * time.Millisecond
|
|
config := peer.ConnConfig{
|
|
Key: pubKey,
|
|
LocalKey: e.config.WgPrivateKey.PublicKey().String(),
|
|
AgentVersion: agentVersion,
|
|
Timeout: timeout,
|
|
WgConfig: wgConfig,
|
|
LocalWgPort: e.config.WgPort,
|
|
RosenpassConfig: peer.RosenpassConfig{
|
|
PubKey: e.getRosenpassPubKey(),
|
|
Addr: e.getRosenpassAddr(),
|
|
PermissiveMode: e.config.RosenpassPermissive,
|
|
},
|
|
ICEConfig: e.createICEConfig(),
|
|
NetMgr: e.netMgr,
|
|
}
|
|
|
|
serviceDependencies := peer.ServiceDependencies{
|
|
StatusRecorder: e.statusRecorder,
|
|
Signaler: e.signaler,
|
|
IFaceDiscover: e.mobileDep.IFaceDiscover,
|
|
RelayManager: e.relayManager,
|
|
SrWatcher: e.srWatcher,
|
|
PortForwardManager: e.portForwardManager,
|
|
MetricsRecorder: e.clientMetrics,
|
|
}
|
|
peerConn, err := peer.NewConn(config, serviceDependencies)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if e.rpManager != nil {
|
|
peerConn.SetOnConnected(e.rpManager.OnConnected)
|
|
peerConn.SetOnDisconnected(e.rpManager.OnDisconnected)
|
|
peerConn.SetRosenpassInitializedPresharedKeyValidator(e.rpManager.IsPresharedKeyInitialized)
|
|
}
|
|
|
|
return peerConn, nil
|
|
}
|
|
|
|
// receiveSignalEvents connects to the Signal Service event stream to negotiate connection with remote peers
|
|
func (e *Engine) receiveSignalEvents() error {
|
|
e.shutdownWg.Add(1)
|
|
go func() {
|
|
defer e.shutdownWg.Done()
|
|
// connect to a stream of messages coming from the signal server
|
|
err := e.signal.Receive(e.ctx, func(msg *sProto.Message) error {
|
|
start := time.Now()
|
|
e.syncMsgMux.Lock()
|
|
defer e.syncMsgMux.Unlock()
|
|
gotLock := time.Since(start)
|
|
|
|
// Check context INSIDE lock to ensure atomicity with shutdown
|
|
if e.ctx.Err() != nil {
|
|
return e.ctx.Err()
|
|
}
|
|
|
|
// Self-addressed heartbeat: the signal client's receive watchdog
|
|
// round-trips this through the server to confirm the receive stream
|
|
// is delivering. Liveness is already recorded before this handler.
|
|
if msg.GetBody().GetType() == sProto.Body_HEARTBEAT {
|
|
return nil
|
|
}
|
|
|
|
conn, ok := e.peerStore.PeerConn(msg.Key)
|
|
if !ok {
|
|
return fmt.Errorf("wrongly addressed message %s", msg.Key)
|
|
}
|
|
|
|
msgType := msg.GetBody().GetType()
|
|
if msgType != sProto.Body_GO_IDLE {
|
|
e.connMgr.ActivatePeer(e.ctx, conn)
|
|
}
|
|
|
|
switch msg.GetBody().Type {
|
|
case sProto.Body_OFFER, sProto.Body_ANSWER:
|
|
offerAnswer, err := convertToOfferAnswer(msg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
log.Debugf("receiveMSG: took %s to get lock for peer %s with session id %s", gotLock, msg.Key, offerAnswer.SessionID)
|
|
|
|
if msg.Body.Type == sProto.Body_OFFER {
|
|
conn.OnRemoteOffer(*offerAnswer)
|
|
} else {
|
|
conn.OnRemoteAnswer(*offerAnswer)
|
|
}
|
|
case sProto.Body_CANDIDATE:
|
|
candidate, err := ice.UnmarshalCandidate(msg.GetBody().Payload)
|
|
if err != nil {
|
|
log.Errorf("failed on parsing remote candidate %s -> %s", candidate, err)
|
|
return err
|
|
}
|
|
|
|
go conn.OnRemoteCandidate(candidate, e.routeManager.GetClientRoutes())
|
|
case sProto.Body_MODE:
|
|
case sProto.Body_GO_IDLE:
|
|
e.connMgr.DeactivatePeer(conn)
|
|
}
|
|
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
// happens if signal is unavailable for a long time.
|
|
// We want to cancel the operation of the whole client
|
|
_ = CtxGetState(e.ctx).Wrap(ErrResetConnection)
|
|
e.clientCancel()
|
|
return
|
|
}
|
|
}()
|
|
|
|
// todo: consider to remove this blocker. I do not see benefit to block the Start operations
|
|
e.signal.WaitStreamConnected(e.ctx)
|
|
if err := e.ctx.Err(); err != nil {
|
|
return fmt.Errorf("wait for signal stream: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (e *Engine) parseNATExternalIPMappings() []string {
|
|
var mappedIPs []string
|
|
var ignoredIFaces = make(map[string]interface{})
|
|
for _, iFace := range e.config.IFaceBlackList {
|
|
ignoredIFaces[iFace] = nil
|
|
}
|
|
for _, mapping := range e.config.NATExternalIPs {
|
|
var external, internal string
|
|
var externalIP, internalIP net.IP
|
|
var err error
|
|
|
|
split := strings.Split(mapping, "/")
|
|
if len(split) > 2 {
|
|
log.Warnf("ignoring invalid external mapping '%s', too many delimiters", mapping)
|
|
break
|
|
}
|
|
if len(split) > 1 {
|
|
internal = split[1]
|
|
internalIP = net.ParseIP(internal)
|
|
if internalIP == nil {
|
|
// not a properly formatted IP address, maybe it's interface name?
|
|
if _, present := ignoredIFaces[internal]; present {
|
|
log.Warnf("internal interface '%s' in blacklist, ignoring external mapping '%s'", internal, mapping)
|
|
break
|
|
}
|
|
internalIP, err = findIPFromInterfaceName(internal)
|
|
if err != nil {
|
|
log.Warnf("error finding interface IP for interface '%s', ignoring external mapping '%s': %v", internal, mapping, err)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
external = split[0]
|
|
externalIP = net.ParseIP(external)
|
|
if externalIP == nil {
|
|
log.Warnf("invalid external IP, %s, ignoring external IP mapping '%s'", external, mapping)
|
|
break
|
|
}
|
|
mappedIP := externalIP.String()
|
|
if internalIP != nil {
|
|
mappedIP = mappedIP + "/" + internalIP.String()
|
|
}
|
|
mappedIPs = append(mappedIPs, mappedIP)
|
|
log.Infof("parsed external IP mapping of '%s' as '%s'", mapping, mappedIP)
|
|
}
|
|
if len(mappedIPs) != len(e.config.NATExternalIPs) {
|
|
log.Warnf("one or more external IP mappings failed to parse, ignoring all mappings")
|
|
return nil
|
|
}
|
|
return mappedIPs
|
|
}
|
|
|
|
func (e *Engine) close() {
|
|
if e.afpacketCapture != nil {
|
|
e.afpacketCapture.Stop()
|
|
e.afpacketCapture = nil
|
|
}
|
|
|
|
log.Debugf("removing Netbird interface %s", e.config.WgIfaceName)
|
|
|
|
if e.wgInterface != nil {
|
|
// Drop the handle before the close starts: a retune that loads it
|
|
// afterwards would touch a device on its way out and report success
|
|
// for an engine that is already gone.
|
|
e.wgDevice.Store(nil)
|
|
if err := e.wgInterface.Close(); err != nil {
|
|
log.Errorf("failed closing Netbird interface %s %v", e.config.WgIfaceName, err)
|
|
}
|
|
e.wgInterface = nil
|
|
e.statusRecorder.SetWgIface(nil)
|
|
}
|
|
|
|
if e.firewall != nil {
|
|
err := e.firewall.Close(e.stateManager)
|
|
if err != nil {
|
|
log.Warnf("failed to reset firewall: %s", err)
|
|
}
|
|
}
|
|
|
|
if e.rpManager != nil {
|
|
_ = e.rpManager.Close()
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
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) newWgIface() (*iface.WGIface, error) {
|
|
transportNet := e.newStdNet()
|
|
|
|
opts := iface.WGIFaceOpts{
|
|
IFaceName: e.config.WgIfaceName,
|
|
Address: e.config.WgAddr,
|
|
WGPort: e.config.WgPort,
|
|
WGPrivKey: e.config.WgPrivateKey.String(),
|
|
MTU: e.config.MTU,
|
|
TransportNet: transportNet,
|
|
DisableDNS: e.config.DisableDNS,
|
|
}
|
|
|
|
switch runtime.GOOS {
|
|
case "android":
|
|
opts.MobileArgs = &device.MobileIFaceArguments{
|
|
TunAdapter: e.mobileDep.TunAdapter,
|
|
TunFd: int(e.mobileDep.FileDescriptor),
|
|
}
|
|
case "ios":
|
|
opts.MobileArgs = &device.MobileIFaceArguments{
|
|
TunFd: int(e.mobileDep.FileDescriptor),
|
|
}
|
|
}
|
|
|
|
return iface.NewWGIFace(opts)
|
|
}
|
|
|
|
func (e *Engine) wgInterfaceCreate() (err error) {
|
|
switch runtime.GOOS {
|
|
case "android":
|
|
err = e.wgInterface.CreateOnAndroid(e.routeManager.CurrentRouteRange(), e.dnsServer.DnsIP().String(), e.dnsServer.SearchDomains())
|
|
case "ios":
|
|
e.mobileDep.NetworkChangeListener.SetInterfaceIP(e.config.WgAddr.String())
|
|
if e.config.WgAddr.HasIPv6() {
|
|
e.mobileDep.NetworkChangeListener.SetInterfaceIPv6(e.config.WgAddr.IPv6String())
|
|
}
|
|
err = e.wgInterface.Create()
|
|
default:
|
|
err = e.wgInterface.Create()
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (e *Engine) newDnsServer() (dns.Server, error) {
|
|
// due to tests where we are using a mocked version of the DNS server
|
|
if e.dnsServer != nil {
|
|
return e.dnsServer, nil
|
|
}
|
|
|
|
switch runtime.GOOS {
|
|
case "android":
|
|
dnsServer := dns.NewDefaultServerPermanentUpstream(
|
|
e.ctx,
|
|
e.wgInterface,
|
|
e.mobileDep.HostDNSAddresses,
|
|
nbdns.Config{},
|
|
e.mobileDep.NetworkChangeListener,
|
|
e.statusRecorder,
|
|
e.config.DisableDNS,
|
|
)
|
|
go e.mobileDep.DnsReadyListener.OnReady()
|
|
return dnsServer, nil
|
|
|
|
case "ios":
|
|
dnsServer := dns.NewDefaultServerIos(e.ctx, e.wgInterface, e.mobileDep.DnsManager, e.statusRecorder, e.config.DisableDNS)
|
|
return dnsServer, nil
|
|
|
|
default:
|
|
|
|
dnsServer, err := dns.NewDefaultServer(e.ctx, dns.DefaultServerConfig{
|
|
WgInterface: e.wgInterface,
|
|
CustomAddress: e.config.CustomDNSAddress,
|
|
StatusRecorder: e.statusRecorder,
|
|
StateManager: e.stateManager,
|
|
DisableSys: e.config.DisableDNS,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return dnsServer, nil
|
|
}
|
|
}
|
|
|
|
// GetRouteManager returns the route manager
|
|
func (e *Engine) GetRouteManager() routemanager.Manager {
|
|
return e.routeManager
|
|
}
|
|
|
|
// GetFirewallManager returns the firewall manager.
|
|
func (e *Engine) GetFirewallManager() firewallManager.Manager {
|
|
return e.firewall
|
|
}
|
|
|
|
// GetExposeManager returns the expose session manager.
|
|
func (e *Engine) GetExposeManager() *expose.Manager {
|
|
e.syncMsgMux.Lock()
|
|
defer e.syncMsgMux.Unlock()
|
|
return e.exposeManager
|
|
}
|
|
|
|
// IsBlockInbound returns whether inbound connections are blocked.
|
|
func (e *Engine) IsBlockInbound() bool {
|
|
return e.config.BlockInbound
|
|
}
|
|
|
|
// GetClientMetrics returns the client metrics
|
|
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.
|
|
//
|
|
// It deliberately does not take syncMsgMux. Raising the buffer pool cap is the
|
|
// recovery path for a device whose pool is exhausted, and an exhausted pool
|
|
// blocks peer removal inside handleSync, which holds syncMsgMux for as long as
|
|
// it stays blocked. Taking the lock here would make the retune unreachable in
|
|
// the one situation that needs it.
|
|
func (e *Engine) SetPerformance(t Performance) error {
|
|
dev := e.wgDevice.Load()
|
|
if dev == nil {
|
|
return errors.New("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 {
|
|
return nil, err
|
|
}
|
|
return findIPFromInterface(iface)
|
|
}
|
|
|
|
func findIPFromInterface(iface *net.Interface) (net.IP, error) {
|
|
ifaceAddrs, err := iface.Addrs()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, addr := range ifaceAddrs {
|
|
if ipv4Addr := addr.(*net.IPNet).IP.To4(); ipv4Addr != nil {
|
|
return ipv4Addr, nil
|
|
}
|
|
}
|
|
return nil, fmt.Errorf("interface %s don't have an ipv4 address", iface.Name)
|
|
}
|
|
|
|
func (e *Engine) getRosenpassPubKey() []byte {
|
|
if e.rpManager != nil {
|
|
return e.rpManager.GetPubKey()
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (e *Engine) getRosenpassAddr() string {
|
|
if e.rpManager != nil {
|
|
return e.rpManager.GetAddress().String()
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// RunHealthProbes executes health checks for Signal, Management, Relay, and WireGuard services
|
|
// and updates the status recorder with the latest states.
|
|
//
|
|
// ctx scopes the (potentially slow) STUN/TURN probing: a caller that gives up —
|
|
// e.g. a Status RPC whose client disconnected — cancels its ctx and the probe
|
|
// returns instead of running to its per-component timeout. The engine's own
|
|
// lifetime ctx still applies independently, so an engine shutdown aborts the
|
|
// probe even if the caller's ctx is context.Background().
|
|
func (e *Engine) RunHealthProbes(ctx context.Context, waitForResult bool) bool {
|
|
// Tie the caller's ctx to the engine lifetime: either cancelling aborts
|
|
// the probe below.
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
stop := context.AfterFunc(e.ctx, cancel)
|
|
defer stop()
|
|
|
|
e.syncMsgMux.Lock()
|
|
|
|
signalHealthy := e.signal.IsHealthy()
|
|
log.Debugf("signal health check: healthy=%t", signalHealthy)
|
|
|
|
managementHealthy := e.mgmClient.IsHealthy()
|
|
log.Debugf("management health check: healthy=%t", managementHealthy)
|
|
|
|
stuns := slices.Clone(e.STUNs)
|
|
turns := slices.Clone(e.TURNs)
|
|
|
|
if err := e.statusRecorder.RefreshWireGuardStats(); err != nil {
|
|
log.Debugf("failed to refresh WireGuard stats: %v", err)
|
|
}
|
|
|
|
e.syncMsgMux.Unlock()
|
|
|
|
// Skip STUN/TURN probing for JS/WASM as it's not available
|
|
relayHealthy := true
|
|
if runtime.GOOS != "js" {
|
|
var results []relay.ProbeResult
|
|
if waitForResult {
|
|
results = e.probeStunTurn.ProbeAllWaitResult(ctx, stuns, turns)
|
|
} else {
|
|
results = e.probeStunTurn.ProbeAll(ctx, stuns, turns)
|
|
}
|
|
e.statusRecorder.UpdateRelayStates(results)
|
|
|
|
for _, res := range results {
|
|
if res.Err != nil {
|
|
relayHealthy = false
|
|
break
|
|
}
|
|
}
|
|
log.Debugf("relay health check: healthy=%t", relayHealthy)
|
|
}
|
|
|
|
allHealthy := signalHealthy && managementHealthy && relayHealthy
|
|
log.Debugf("all health checks completed: healthy=%t", allHealthy)
|
|
return allHealthy
|
|
}
|
|
|
|
// triggerClientRestart triggers a full client restart by cancelling the client context.
|
|
// Note: This does NOT just restart the engine - it cancels the entire client context,
|
|
// which causes the connect client's retry loop to create a completely new engine.
|
|
func (e *Engine) triggerClientRestart() {
|
|
e.syncMsgMux.Lock()
|
|
defer e.syncMsgMux.Unlock()
|
|
|
|
if e.ctx.Err() != nil {
|
|
return
|
|
}
|
|
|
|
log.Info("restarting engine")
|
|
CtxGetState(e.ctx).Set(StatusConnecting)
|
|
_ = CtxGetState(e.ctx).Wrap(ErrResetConnection)
|
|
log.Infof("cancelling client context, engine will be recreated")
|
|
e.clientCancel()
|
|
}
|
|
|
|
func (e *Engine) startNetworkMonitor() {
|
|
if !e.config.NetworkMonitor || nbnetstack.IsEnabled() {
|
|
log.Infof("Network monitor is disabled, not starting")
|
|
return
|
|
}
|
|
|
|
e.networkMonitor = networkmonitor.New()
|
|
e.shutdownWg.Add(1)
|
|
go func() {
|
|
defer e.shutdownWg.Done()
|
|
if err := e.networkMonitor.Listen(e.ctx); err != nil {
|
|
if errors.Is(err, context.Canceled) {
|
|
log.Infof("network monitor stopped")
|
|
return
|
|
}
|
|
log.Errorf("network monitor error: %v", err)
|
|
return
|
|
}
|
|
|
|
log.Infof("Network monitor: detected network change, triggering client restart")
|
|
e.triggerClientRestart()
|
|
}()
|
|
}
|
|
|
|
func (e *Engine) stopDNSServer() {
|
|
if e.dnsServer == nil {
|
|
return
|
|
}
|
|
e.dnsServer.Stop()
|
|
e.dnsServer = nil
|
|
err := fmt.Errorf("DNS server stopped")
|
|
nsGroupStates := e.statusRecorder.GetDNSStates()
|
|
for i := range nsGroupStates {
|
|
nsGroupStates[i].Enabled = false
|
|
nsGroupStates[i].Error = err
|
|
}
|
|
e.statusRecorder.UpdateDNSStates(nsGroupStates)
|
|
}
|
|
|
|
// 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.syncStore != nil) {
|
|
return
|
|
}
|
|
log.Debugf("Sync response persistence is set to %t", enabled)
|
|
|
|
if !enabled {
|
|
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()
|
|
defer e.syncRespMux.RUnlock()
|
|
|
|
if e.syncStore == nil {
|
|
return nil, errors.New("sync response persistence is disabled")
|
|
}
|
|
|
|
//nolint:nilnil
|
|
return e.syncStore.Get()
|
|
}
|
|
|
|
// GetWgAddr returns the wireguard address
|
|
func (e *Engine) GetWgAddr() netip.Addr {
|
|
if e.wgInterface == nil {
|
|
return netip.Addr{}
|
|
}
|
|
return e.wgInterface.Address().IP
|
|
}
|
|
|
|
// GetWgV6Addr returns the IPv6 overlay address of the WireGuard interface.
|
|
func (e *Engine) GetWgV6Addr() netip.Addr {
|
|
if e.wgInterface == nil {
|
|
return netip.Addr{}
|
|
}
|
|
return e.wgInterface.Address().IPv6
|
|
}
|
|
|
|
func (e *Engine) RenewTun(fd int) error {
|
|
e.syncMsgMux.Lock()
|
|
wgInterface := e.wgInterface
|
|
e.syncMsgMux.Unlock()
|
|
|
|
if wgInterface == nil {
|
|
return fmt.Errorf("wireguard interface not initialized")
|
|
}
|
|
|
|
if err := wgInterface.RenewTun(fd); err != nil {
|
|
return err
|
|
}
|
|
|
|
e.rebindOverlayListeners()
|
|
return nil
|
|
}
|
|
|
|
// rebindOverlayListeners gives the servers that listen on an overlay address
|
|
// sockets on the interface as it is now.
|
|
//
|
|
// A socket belongs to the interface generation it was created on. Renewing the
|
|
// TUN builds a new interface and moves the overlay addresses to it, which
|
|
// leaves the old sockets in LISTEN with the uspfilter still logging packets
|
|
// arriving for them, while every accept fails with EINVAL for the life of the
|
|
// socket: from the outside the server looks alive and answers nothing. On
|
|
// Android this happens during a normal startup, where the first TUN is
|
|
// established before the routes are known and replaced once they arrive.
|
|
//
|
|
// Rebinding costs whatever those sockets were carrying, which the renewal has
|
|
// already broken. Errors are logged rather than returned: the renewal itself
|
|
// succeeded, and failing it would hand the caller a working interface and an
|
|
// error.
|
|
func (e *Engine) rebindOverlayListeners() {
|
|
e.syncMsgMux.Lock()
|
|
defer e.syncMsgMux.Unlock()
|
|
|
|
for _, rebind := range e.overlayRebinds() {
|
|
if err := rebind(); err != nil {
|
|
log.Errorf("after TUN renewal: %v", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// overlayRebinds is every subsystem of this engine that holds sockets bound to
|
|
// an overlay address, and how to rebuild each one's.
|
|
//
|
|
// A subsystem that starts listening on an overlay address belongs in this list.
|
|
// Leaving it out costs nothing that review would notice and produces a listener
|
|
// that stays in LISTEN, is logged as receiving packets, and refuses every
|
|
// connection for the life of the process.
|
|
func (e *Engine) overlayRebinds() []overlayRebind {
|
|
return []overlayRebind{
|
|
e.restartSSHListeners,
|
|
e.restartDNSForwarder,
|
|
}
|
|
}
|
|
|
|
// restartDNSForwarder rebuilds the DNS forwarder serving the same domains.
|
|
// No-op when it is not running. See Engine.rebindOverlayListeners.
|
|
func (e *Engine) restartDNSForwarder() error {
|
|
if e.dnsForwardMgr == nil {
|
|
return nil
|
|
}
|
|
// Read from the forwarder before it goes away, so the replacement serves
|
|
// the domains in force now rather than a copy kept somewhere else.
|
|
entries := e.dnsForwardMgr.Domains()
|
|
e.stopDNSForwarder()
|
|
// Both halves log their own failures, so the only thing left to report is
|
|
// the outcome: a start that failed left the manager nil, and the forwarder
|
|
// is now down rather than merely rebound.
|
|
e.startDNSForwarder(entries)
|
|
if e.dnsForwardMgr == nil {
|
|
return errors.New("rebind DNS forwarder: it did not come back up")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// updateDNSForwarder start or stop the DNS forwarder based on the domains and the feature flag
|
|
func (e *Engine) updateDNSForwarder(
|
|
enabled bool,
|
|
fwdEntries []*dnsfwd.ForwarderEntry,
|
|
) {
|
|
if e.config.DisableServerRoutes || e.config.BlockInbound {
|
|
return
|
|
}
|
|
|
|
if !enabled {
|
|
e.stopDNSForwarder()
|
|
return
|
|
}
|
|
|
|
if len(fwdEntries) > 0 {
|
|
if e.dnsForwardMgr == nil {
|
|
e.startDNSForwarder(fwdEntries)
|
|
} else {
|
|
e.dnsForwardMgr.UpdateDomains(fwdEntries)
|
|
}
|
|
} else if e.dnsForwardMgr != nil {
|
|
log.Infof("disable domain router service")
|
|
e.stopDNSForwarder()
|
|
}
|
|
}
|
|
|
|
func (e *Engine) startDNSForwarder(fwdEntries []*dnsfwd.ForwarderEntry) {
|
|
e.dnsForwardMgr = dnsfwd.NewManager(e.firewall, e.statusRecorder, e.wgInterface)
|
|
|
|
if err := e.dnsForwardMgr.Start(fwdEntries); err != nil {
|
|
log.Errorf("failed to start DNS forward: %v", err)
|
|
e.dnsForwardMgr = nil
|
|
return
|
|
}
|
|
|
|
log.Infof("started domain router service with %d entries", len(fwdEntries))
|
|
}
|
|
|
|
func (e *Engine) stopDNSForwarder() {
|
|
if e.dnsForwardMgr == nil {
|
|
return
|
|
}
|
|
|
|
// Bounded because the shutdown waits for queries still in flight, and one
|
|
// waiting on an unresponsive upstream holds it for as long as that lookup
|
|
// is allowed to take. This runs with syncMsgMux held, so that wait is one
|
|
// the whole engine spends.
|
|
ctx, cancel := context.WithTimeout(context.Background(), dnsForwarderStopTimeout)
|
|
defer cancel()
|
|
|
|
if err := e.dnsForwardMgr.Stop(ctx); err != nil {
|
|
log.Errorf("failed to stop DNS forward: %v", err)
|
|
}
|
|
|
|
e.dnsForwardMgr = nil
|
|
}
|
|
|
|
func (e *Engine) GetNet() (*netstack.Net, error) {
|
|
e.syncMsgMux.Lock()
|
|
intf := e.wgInterface
|
|
e.syncMsgMux.Unlock()
|
|
if intf == nil {
|
|
return nil, errors.New("wireguard interface not initialized")
|
|
}
|
|
|
|
nsnet := intf.GetNet()
|
|
if nsnet == nil {
|
|
return nil, errors.New("failed to get netstack")
|
|
}
|
|
return nsnet, nil
|
|
}
|
|
|
|
func (e *Engine) Address() (netip.Addr, error) {
|
|
e.syncMsgMux.Lock()
|
|
intf := e.wgInterface
|
|
e.syncMsgMux.Unlock()
|
|
if intf == nil {
|
|
return netip.Addr{}, errors.New("wireguard interface not initialized")
|
|
}
|
|
|
|
return e.wgInterface.Address().IP, nil
|
|
}
|
|
|
|
// SetCapture sets or clears packet capture on the WireGuard device.
|
|
// On userspace WireGuard, it taps the FilteredDevice directly.
|
|
// On kernel WireGuard (Linux), it falls back to AF_PACKET raw socket capture.
|
|
// Pass nil to disable capture.
|
|
func (e *Engine) SetCapture(pc device.PacketCapture) error {
|
|
e.syncMsgMux.Lock()
|
|
defer e.syncMsgMux.Unlock()
|
|
|
|
intf := e.wgInterface
|
|
if intf == nil {
|
|
return errors.New("wireguard interface not initialized")
|
|
}
|
|
|
|
if e.afpacketCapture != nil {
|
|
e.afpacketCapture.Stop()
|
|
e.afpacketCapture = nil
|
|
}
|
|
|
|
dev := intf.GetDevice()
|
|
if dev != nil {
|
|
dev.SetCapture(pc)
|
|
e.setForwarderCapture(pc)
|
|
return nil
|
|
}
|
|
|
|
// Kernel mode: no FilteredDevice. Use AF_PACKET on Linux.
|
|
if pc == nil {
|
|
return nil
|
|
}
|
|
sess, ok := pc.(*capture.Session)
|
|
if !ok {
|
|
return errors.New("filtered device not available and AF_PACKET requires *capture.Session")
|
|
}
|
|
|
|
afc := capture.NewAFPacketCapture(intf.Name(), sess)
|
|
if err := afc.Start(); err != nil { //nolint:staticcheck // always errors on non-Linux builds
|
|
return fmt.Errorf("start AF_PACKET capture on %s: %w", intf.Name(), err)
|
|
}
|
|
e.afpacketCapture = afc
|
|
return nil
|
|
}
|
|
|
|
// setForwarderCapture propagates capture to the USP filter's forwarder endpoint.
|
|
// This captures outbound response packets that bypass the FilteredDevice in netstack mode.
|
|
func (e *Engine) setForwarderCapture(pc device.PacketCapture) {
|
|
if e.firewall == nil {
|
|
return
|
|
}
|
|
type forwarderCapturer interface {
|
|
SetPacketCapture(pc forwarder.PacketCapture)
|
|
}
|
|
if fc, ok := e.firewall.(forwarderCapturer); ok {
|
|
fc.SetPacketCapture(pc)
|
|
}
|
|
}
|
|
|
|
// toExcludedLazyPeers returns the peers that must have an always-active
|
|
// connection: those that are not lazy by policy (the per-peer lazy state or the
|
|
// account flag, subject to the local override).
|
|
func (e *Engine) toExcludedLazyPeers(peers []*mgmProto.RemotePeerConfig) map[string]bool {
|
|
excludedPeers := make(map[string]bool)
|
|
for _, p := range peers {
|
|
if !e.connMgr.PeerLazyDefault(p.GetLazyState()) {
|
|
excludedPeers[p.GetWgPubKey()] = true
|
|
}
|
|
}
|
|
return excludedPeers
|
|
}
|
|
|
|
// isChecksEqual checks if two slices of checks are equal.
|
|
func isChecksEqual(checks1, checks2 []*mgmProto.Checks) bool {
|
|
normalize := func(checks []*mgmProto.Checks) []string {
|
|
normalized := make([]string, len(checks))
|
|
|
|
for i, check := range checks {
|
|
sortedFiles := slices.Clone(check.Files)
|
|
sort.Strings(sortedFiles)
|
|
normalized[i] = strings.Join(sortedFiles, "|")
|
|
if challenge := check.GetCertificateChallenge(); challenge != nil {
|
|
sortedCAs := slices.Clone(challenge.GetCaCertificates())
|
|
sort.Strings(sortedCAs)
|
|
normalized[i] += fmt.Sprintf("#%x|%s", challenge.GetNonce(), strings.Join(sortedCAs, "|"))
|
|
}
|
|
}
|
|
|
|
sort.Strings(normalized)
|
|
return normalized
|
|
}
|
|
|
|
n1 := normalize(checks1)
|
|
n2 := normalize(checks2)
|
|
|
|
return slices.Equal(n1, n2)
|
|
}
|
|
|
|
func getInterfacePrefixes() ([]netip.Prefix, error) {
|
|
ifaces, err := net.Interfaces()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("get interfaces: %w", err)
|
|
}
|
|
|
|
var prefixes []netip.Prefix
|
|
var merr *multierror.Error
|
|
|
|
for _, iface := range ifaces {
|
|
addrs, err := iface.Addrs()
|
|
if err != nil {
|
|
merr = multierror.Append(merr, fmt.Errorf("get addresses for interface %s: %w", iface.Name, err))
|
|
continue
|
|
}
|
|
for _, addr := range addrs {
|
|
ipNet, ok := addr.(*net.IPNet)
|
|
if !ok {
|
|
merr = multierror.Append(merr, fmt.Errorf("cast address to IPNet: %v", addr))
|
|
continue
|
|
}
|
|
addr, ok := netip.AddrFromSlice(ipNet.IP)
|
|
if !ok {
|
|
merr = multierror.Append(merr, fmt.Errorf("cast IPNet to netip.Addr: %v", ipNet.IP))
|
|
continue
|
|
}
|
|
ones, _ := ipNet.Mask.Size()
|
|
prefix := netip.PrefixFrom(addr.Unmap(), ones).Masked()
|
|
ip := prefix.Addr()
|
|
|
|
if ip.IsLoopback() || ip.IsMulticast() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() {
|
|
continue
|
|
}
|
|
|
|
prefixes = append(prefixes, prefix)
|
|
}
|
|
}
|
|
|
|
return prefixes, nberrors.FormatErrorOrNil(merr)
|
|
}
|
|
|
|
// filterAllowedIPs strips IPv6 entries when the local interface has no v6 address.
|
|
// This covers both the explicit --disable-ipv6 flag (v6 never assigned) and the
|
|
// case where OS v6 assignment failed (ClearIPv6). Without this, WireGuard would
|
|
// accept v6 traffic that the native firewall cannot filter.
|
|
func (e *Engine) filterAllowedIPs(ips []string) []string {
|
|
if e.wgInterface.Address().HasIPv6() {
|
|
return ips
|
|
}
|
|
filtered := make([]string, 0, len(ips))
|
|
for _, s := range ips {
|
|
p, err := netip.ParsePrefix(s)
|
|
if err != nil || !p.Addr().Is6() {
|
|
filtered = append(filtered, s)
|
|
}
|
|
}
|
|
return filtered
|
|
}
|
|
|
|
// compareNetIPLists compares a list of netip.Prefix with a list of strings.
|
|
// return true if both lists are equal, false otherwise.
|
|
func compareNetIPLists(list1 []netip.Prefix, list2 []string) bool {
|
|
if len(list1) != len(list2) {
|
|
return false
|
|
}
|
|
|
|
freq := make(map[string]int, len(list1))
|
|
for _, p := range list1 {
|
|
freq[p.String()]++
|
|
}
|
|
|
|
for _, s := range list2 {
|
|
p, err := netip.ParsePrefix(s)
|
|
if err != nil {
|
|
return false // invalid prefix in list2.
|
|
}
|
|
key := p.String()
|
|
if freq[key] == 0 {
|
|
return false
|
|
}
|
|
freq[key]--
|
|
}
|
|
|
|
// all counts should be zero if lists are equal.
|
|
for _, count := range freq {
|
|
if count != 0 {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func fileExists(path string) bool {
|
|
_, err := os.Stat(path)
|
|
return !os.IsNotExist(err)
|
|
}
|
|
|
|
func createFile(path string) error {
|
|
file, err := os.Create(path)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return file.Close()
|
|
}
|
|
|
|
func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) {
|
|
remoteCred, err := signal.UnMarshalCredential(msg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var (
|
|
rosenpassPubKey []byte
|
|
rosenpassAddr string
|
|
)
|
|
if cfg := msg.GetBody().GetRosenpassConfig(); cfg != nil {
|
|
rosenpassPubKey = cfg.GetRosenpassPubKey()
|
|
rosenpassAddr = cfg.GetRosenpassServerAddr()
|
|
}
|
|
|
|
// Handle optional SessionID
|
|
var sessionID *peer.ICESessionID
|
|
if sessionBytes := msg.GetBody().GetSessionId(); sessionBytes != nil {
|
|
if id, err := peer.ICESessionIDFromBytes(sessionBytes); err != nil {
|
|
log.Warnf("Invalid session ID in message: %v", err)
|
|
sessionID = nil // Set to nil if conversion fails
|
|
} else {
|
|
sessionID = &id
|
|
}
|
|
}
|
|
|
|
relayIP := decodeRelayIP(msg.GetBody().GetRelayServerIP())
|
|
|
|
offerAnswer := peer.OfferAnswer{
|
|
IceCredentials: peer.IceCredentials{
|
|
UFrag: remoteCred.UFrag,
|
|
Pwd: remoteCred.Pwd,
|
|
},
|
|
WgListenPort: int(msg.GetBody().GetWgListenPort()),
|
|
Version: msg.GetBody().GetNetBirdVersion(),
|
|
RosenpassPubKey: rosenpassPubKey,
|
|
RosenpassAddr: rosenpassAddr,
|
|
RelaySrvAddress: msg.GetBody().GetRelayServerAddress(),
|
|
RelaySrvIP: relayIP,
|
|
SessionID: sessionID,
|
|
}
|
|
return &offerAnswer, nil
|
|
}
|
|
|
|
// decodeRelayIP decodes the proto relayServerIP bytes (4 or 16) into a
|
|
// netip.Addr. Returns the zero value for empty input and logs a warning
|
|
// for malformed payloads.
|
|
func decodeRelayIP(b []byte) netip.Addr {
|
|
if len(b) == 0 {
|
|
return netip.Addr{}
|
|
}
|
|
ip, ok := netip.AddrFromSlice(b)
|
|
if !ok {
|
|
log.Warnf("invalid relayServerIP in signal message (%d bytes), ignoring", len(b))
|
|
return netip.Addr{}
|
|
}
|
|
return ip.Unmap()
|
|
}
|