mirror of
https://github.com/netbirdio/netbird.git
synced 2026-08-24 00:21:29 +02:00
Compare commits
7 Commits
agent-netw
...
debug-ui-m
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7582015891 | ||
|
|
1fec88a51e | ||
|
|
ce32805f69 | ||
|
|
d0ea1d77b2 | ||
|
|
bc73e6bd71 | ||
|
|
04ca92d42b | ||
|
|
079aee8d63 |
13
.github/workflows/agent-network-e2e.yml
vendored
13
.github/workflows/agent-network-e2e.yml
vendored
@@ -12,13 +12,6 @@ on:
|
||||
AWS issues it. Leave empty for the Sonnet 4.6 default.
|
||||
required: false
|
||||
default: ""
|
||||
test_pattern:
|
||||
description: >-
|
||||
Package pattern to run. Defaults to the whole suite; narrow it to one
|
||||
package (e.g. ./e2e/agentnetwork/...) when a run only needs that
|
||||
package's answer and not the sixteen minutes the container suite costs.
|
||||
required: false
|
||||
default: "./e2e/..."
|
||||
|
||||
concurrency:
|
||||
group: ${{ github.workflow }}-${{ github.ref }}
|
||||
@@ -84,8 +77,4 @@ jobs:
|
||||
GOOGLE_VERTEX_PROJECT: ${{ secrets.E2E_GOOGLE_VERTEX_PROJECT }}
|
||||
GOOGLE_VERTEX_REGION: ${{ secrets.E2E_GOOGLE_VERTEX_REGION }}
|
||||
GOOGLE_VERTEX_MODEL: ${{ secrets.E2E_GOOGLE_VERTEX_MODEL }}
|
||||
# Read through an env var rather than interpolated into the run
|
||||
# script: a dispatch input reaching a shell command directly is a
|
||||
# script-injection seam, however trusted the dispatcher.
|
||||
TEST_PATTERN: ${{ inputs.test_pattern || './e2e/...' }}
|
||||
run: go test -tags e2e -timeout 40m -v "$TEST_PATTERN"
|
||||
run: go test -tags e2e -timeout 40m -v ./e2e/...
|
||||
|
||||
16
.github/workflows/mobile-build-validation.yml
vendored
16
.github/workflows/mobile-build-validation.yml
vendored
@@ -43,19 +43,8 @@ jobs:
|
||||
run: /usr/local/lib/android/sdk/cmdline-tools/7.0/bin/sdkmanager --install "ndk;23.1.7779620"
|
||||
- name: install gomobile
|
||||
run: go install golang.org/x/mobile/cmd/gomobile@v0.0.0-20251113184115-a159579294ab
|
||||
# `gomobile init` re-installs gobind from golang.org/x/mobile@latest
|
||||
# regardless of the pin above (cmd/gomobile/init.go: "Make sure gobind is
|
||||
# up to date"), so this step resolves a version nobody chose, on every run.
|
||||
#
|
||||
# setup-go sets GOTOOLCHAIN=local, so that install fails outright once
|
||||
# x/mobile@latest declares a newer Go than go.mod does — which it did on
|
||||
# 2026-08-21, breaking both jobs on every branch at once. GOTOOLCHAIN=auto
|
||||
# lets this one install fetch the toolchain it asks for. Scoped to the
|
||||
# step: the repo's own Go version, and every build below, is unaffected.
|
||||
- name: gomobile init
|
||||
run: gomobile init
|
||||
env:
|
||||
GOTOOLCHAIN: auto
|
||||
- name: build android netbird lib
|
||||
run: PATH=$PATH:$(go env GOPATH) gomobile bind -o $GITHUB_WORKSPACE/netbird.aar -javapkg=io.netbird.gomobile -ldflags="-checklinkname=0 -X golang.zx2c4.com/wireguard/ipc.socketDirectory=/data/data/io.netbird.client/cache/wireguard -X github.com/netbirdio/netbird/version.version=buildtest" $GITHUB_WORKSPACE/client/android
|
||||
env:
|
||||
@@ -75,13 +64,8 @@ jobs:
|
||||
go-version-file: "go.mod"
|
||||
- name: install gomobile
|
||||
run: go install golang.org/x/mobile/cmd/gomobile@v0.0.0-20251113184115-a159579294ab
|
||||
# See the Android job: `gomobile init` re-installs gobind from
|
||||
# golang.org/x/mobile@latest regardless of the pin above, and needs a
|
||||
# toolchain it may pick newer than go.mod's.
|
||||
- name: gomobile init
|
||||
run: gomobile init
|
||||
env:
|
||||
GOTOOLCHAIN: auto
|
||||
- name: build iOS netbird lib
|
||||
run: PATH=$PATH:$(go env GOPATH) gomobile bind -target=ios -bundleid=io.netbird.framework -ldflags="-X github.com/netbirdio/netbird/version.version=buildtest" -o ./NetBirdSDK.xcframework ./client/ios/NetBirdSDK
|
||||
env:
|
||||
|
||||
78
.github/workflows/no-new-replace.yml
vendored
78
.github/workflows/no-new-replace.yml
vendored
@@ -1,78 +0,0 @@
|
||||
name: No New Replace Directives
|
||||
|
||||
on:
|
||||
pull_request:
|
||||
paths:
|
||||
- "go.mod"
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
concurrency:
|
||||
group: ${{ github.workflow }}-${{ github.ref }}-${{ github.head_ref || github.actor_id }}
|
||||
cancel-in-progress: true
|
||||
|
||||
jobs:
|
||||
check-replace-directives:
|
||||
name: check-replace-directives
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
|
||||
with:
|
||||
persist-credentials: false
|
||||
fetch-depth: 0
|
||||
|
||||
- name: Install Go
|
||||
uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 # v6.5.0
|
||||
with:
|
||||
go-version-file: go.mod
|
||||
|
||||
- name: Compare replace directives against the base branch
|
||||
env:
|
||||
BASE_SHA: ${{ github.event.pull_request.base.sha }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
|
||||
# A replace directive only applies when this module is the main
|
||||
# module. Anything importing netbird as a library, the embedded
|
||||
# clients among them, resolves the replaced path upstream instead and
|
||||
# fails to build against whatever the replacement provides. Requiring
|
||||
# a fork under its own module path avoids that; a replace does not.
|
||||
#
|
||||
# go.mod is parsed rather than diffed so that reordering, comments and
|
||||
# single-line versus block syntax do not register as changes.
|
||||
#
|
||||
# Versions are part of the key because a replace can be scoped to one
|
||||
# version of a module. Keyed on paths alone, retargeting such a
|
||||
# directive at a different version would read as unchanged.
|
||||
list_replaces() {
|
||||
go mod edit -json "$1" \
|
||||
| jq -r '
|
||||
def ref: .Path + (if (.Version // "") == "" then "" else " " + .Version end);
|
||||
(.Replace // [])[] | "\(.Old | ref) => \(.New | ref)"
|
||||
' \
|
||||
| sort
|
||||
}
|
||||
|
||||
git show "${BASE_SHA}:go.mod" > /tmp/base-go.mod
|
||||
list_replaces /tmp/base-go.mod > /tmp/base-replaces
|
||||
list_replaces go.mod > /tmp/head-replaces
|
||||
|
||||
added=$(comm -13 /tmp/base-replaces /tmp/head-replaces)
|
||||
if [ -n "$added" ]; then
|
||||
echo "::error::This PR adds a replace directive to go.mod:"
|
||||
echo "$added" | sed 's/^/ /'
|
||||
echo ""
|
||||
echo "A replace directive applies only to the main module, so it does not"
|
||||
echo "reach anything that imports netbird as a library. Require the module"
|
||||
echo "under a path you control instead, as done for github.com/netbirdio/go-nat."
|
||||
exit 1
|
||||
fi
|
||||
|
||||
removed=$(comm -23 /tmp/base-replaces /tmp/head-replaces)
|
||||
if [ -n "$removed" ]; then
|
||||
echo "This PR removes replace directives:"
|
||||
echo "$removed" | sed 's/^/ /'
|
||||
fi
|
||||
echo "No new replace directives."
|
||||
@@ -40,35 +40,6 @@ You can then use this private endpoint to configure your AI agents, whether that
|
||||
Full step-by-step setup:
|
||||
**https://docs.netbird.io/agent-network/quickstart**
|
||||
|
||||
## Client settings that don't follow the endpoint
|
||||
|
||||
Most of an agent's traffic follows the base URL you hand it, but a few
|
||||
client-side checks call their vendor directly and never reach the proxy. On a
|
||||
network that blocks direct egress they fail even though inference works, so
|
||||
they are worth setting once when you roll the endpoint out.
|
||||
|
||||
For Claude Code:
|
||||
|
||||
- **Fast mode** checks availability against `api.anthropic.com` rather than the
|
||||
configured base URL. Set `CLAUDE_CODE_SKIP_FAST_MODE_ORG_CHECK=1` when the
|
||||
agent authenticates with `ANTHROPIC_AUTH_TOKEN` alone (the usual shape when
|
||||
the proxy injects the real provider key) or when a TLS-inspecting proxy
|
||||
answers the check itself. Set
|
||||
`CLAUDE_CODE_SKIP_FAST_MODE_NETWORK_ERRORS=1` when the network refuses the
|
||||
connection outright. Fast mode is an Anthropic-API feature, so it is
|
||||
unavailable on a Bedrock- or Vertex-backed endpoint whatever you set.
|
||||
- **Model discovery** is off by default. Set
|
||||
`CLAUDE_CODE_ENABLE_GATEWAY_MODEL_DISCOVERY=1` for the picker to list the
|
||||
models your policies authorise; the proxy filters the response to that set.
|
||||
The client gives discovery a three-second budget and treats any redirect as
|
||||
a failure, so the endpoint must serve `/v1/models` directly.
|
||||
- **The WebFetch domain safety check** also calls `api.anthropic.com` directly
|
||||
and is unaffected by the variables above.
|
||||
|
||||
Allowing direct egress to `api.anthropic.com` covers the network cases but not
|
||||
the credential one, where the check reaches Anthropic and is rejected because
|
||||
the agent presents a proxy-issued key.
|
||||
|
||||
## Architecture
|
||||
|
||||
Agent Network is built on two existing NetBird capabilities:
|
||||
|
||||
@@ -26,8 +26,6 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/routemanager"
|
||||
"github.com/netbirdio/netbird/client/internal/stdnet"
|
||||
"github.com/netbirdio/netbird/client/net"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
"github.com/netbirdio/netbird/client/system"
|
||||
"github.com/netbirdio/netbird/formatter"
|
||||
"github.com/netbirdio/netbird/route"
|
||||
@@ -42,6 +40,11 @@ const (
|
||||
AnonymizeLevelStrict = nbAnonymize.LevelStrictString
|
||||
)
|
||||
|
||||
// ConnectionListener export internal Listener for mobile
|
||||
type ConnectionListener interface {
|
||||
peer.Listener
|
||||
}
|
||||
|
||||
// TunAdapter export internal TunAdapter for mobile
|
||||
type TunAdapter interface {
|
||||
device.TunAdapter
|
||||
@@ -82,13 +85,6 @@ type Client struct {
|
||||
deviceName string
|
||||
uiVersion string
|
||||
networkChangeListener listener.NetworkChangeListener
|
||||
// netState outlives engine restarts: it mirrors the OS connectivity, not
|
||||
// the engine lifecycle. Run and RunWithoutLogin inject it into each new
|
||||
// ConnectClient, which distributes it to every reconnection loop.
|
||||
netState *netstate.State
|
||||
|
||||
// sweeper also outlives engine restarts; NotifyNetworkChange sweeps it.
|
||||
sweeper *netsweep.Sweeper
|
||||
|
||||
stateMu sync.RWMutex
|
||||
connectClient *internal.ConnectClient
|
||||
@@ -152,7 +148,6 @@ func NewClient(androidSDKVersion int, deviceName string, uiVersion string, tunAd
|
||||
execWorkaround(androidSDKVersion)
|
||||
|
||||
net.SetAndroidProtectSocketFn(tunAdapter.ProtectSocket)
|
||||
system.SetIFaceDiscover(iFaceDiscover)
|
||||
return &Client{
|
||||
deviceName: deviceName,
|
||||
uiVersion: uiVersion,
|
||||
@@ -161,8 +156,6 @@ func NewClient(androidSDKVersion int, deviceName string, uiVersion string, tunAd
|
||||
recorder: peer.NewRecorder(""),
|
||||
ctxCancelLock: &sync.Mutex{},
|
||||
networkChangeListener: networkChangeListener,
|
||||
netState: netstate.New(),
|
||||
sweeper: netsweep.New(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -203,8 +196,7 @@ func (c *Client) Run(platformFiles PlatformFiles, urlOpener URLOpener, isAndroid
|
||||
}
|
||||
// todo do not throw error in case of cancelled context
|
||||
ctx = internal.CtxInitState(ctx)
|
||||
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder,
|
||||
internal.WithNetworkState(c.netState), internal.WithSweeper(c.sweeper))
|
||||
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder)
|
||||
c.setState(cfg, cacheDir, cfgFile, connectClient)
|
||||
// This path runs the interactive SSO flow, so reaching here means the peer
|
||||
// is authenticated again — release the latch Status() reports from. Clear
|
||||
@@ -245,8 +237,7 @@ func (c *Client) RunWithoutLogin(platformFiles PlatformFiles, dns *DNSList, dnsR
|
||||
|
||||
// todo do not throw error in case of cancelled context
|
||||
ctx = internal.CtxInitState(ctx)
|
||||
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder,
|
||||
internal.WithNetworkState(c.netState), internal.WithSweeper(c.sweeper))
|
||||
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder)
|
||||
c.setState(cfg, cacheDir, cfgFile, connectClient)
|
||||
return connectClient.RunOnAndroid(c.tunAdapter, c.iFaceDiscover, c.networkChangeListener, slices.Clone(dns.items), dnsReadyListener, stateFile, cacheDir)
|
||||
}
|
||||
@@ -294,24 +285,6 @@ func (c *Client) GetTunSettings() (*TunSettings, error) {
|
||||
}, nil
|
||||
}
|
||||
|
||||
// SetNetworkAvailable feeds OS-reported network availability into the client.
|
||||
// While unavailable, the internal reconnect loops suspend their attempts and
|
||||
// the connection listener reports NoNetwork instead of Connecting; when
|
||||
// availability returns, the loops resume immediately with a fresh backoff.
|
||||
func (c *Client) SetNetworkAvailable(available bool) {
|
||||
c.netState.Set(available)
|
||||
c.recorder.SetNetworkAvailable(available)
|
||||
}
|
||||
|
||||
// NotifyNetworkChange marks the management, signal and relay connections
|
||||
// stale after the OS switched networks and schedules a sweep that cuts
|
||||
// whatever has not redialed on the new network by then. The engine and the
|
||||
// TUN device stay untouched.
|
||||
func (c *Client) NotifyNetworkChange() {
|
||||
c.sweeper.MarkNetworkChange()
|
||||
log.Infof("network change: connections marked stale")
|
||||
}
|
||||
|
||||
// DebugBundle generates a debug bundle, uploads it, and returns the upload key.
|
||||
// It works both with and without a running engine. anonymizeLevel is "default"
|
||||
// or "strict"; strict also anonymizes internal IP ranges, peer names, and
|
||||
@@ -552,11 +525,7 @@ func (c *Client) OnUpdatedHostDNS(list *DNSList) error {
|
||||
|
||||
// SetConnectionListener set the network connection listener
|
||||
func (c *Client) SetConnectionListener(listener ConnectionListener) {
|
||||
if listener == nil {
|
||||
c.recorder.RemoveConnectionListener()
|
||||
return
|
||||
}
|
||||
c.recorder.SetConnectionListener(connectionListenerAdapter{listener})
|
||||
c.recorder.SetConnectionListener(listener)
|
||||
}
|
||||
|
||||
// RemoveConnectionListener remove connection listener
|
||||
|
||||
@@ -1,41 +0,0 @@
|
||||
//go:build android
|
||||
|
||||
package android
|
||||
|
||||
import (
|
||||
"github.com/netbirdio/netbird/client/internal/peer"
|
||||
)
|
||||
|
||||
// Client state values delivered via ConnectionListener.OnStateChanged,
|
||||
// re-exported as basic constants so gomobile emits them into the generated
|
||||
// Java bindings. They mirror peer.ClientState*: append-only, never reorder.
|
||||
const (
|
||||
ClientStateDisconnected = int(peer.ClientStateDisconnected)
|
||||
ClientStateConnected = int(peer.ClientStateConnected)
|
||||
ClientStateConnecting = int(peer.ClientStateConnecting)
|
||||
ClientStateDisconnecting = int(peer.ClientStateDisconnecting)
|
||||
ClientStateNoNetwork = int(peer.ClientStateNoNetwork)
|
||||
)
|
||||
|
||||
// ConnectionListener export internal Listener for mobile. It mirrors
|
||||
// peer.Listener with OnStateChanged taking a plain int (one of the
|
||||
// ClientState* constants), because gomobile cannot bind named types.
|
||||
type ConnectionListener interface {
|
||||
OnStateChanged(state int)
|
||||
OnConnected()
|
||||
OnDisconnected()
|
||||
OnConnecting()
|
||||
OnDisconnecting()
|
||||
OnAddressChanged(string, string)
|
||||
OnPeersListChanged(int)
|
||||
}
|
||||
|
||||
// connectionListenerAdapter adapts the gomobile-facing ConnectionListener to
|
||||
// peer.Listener, converting the typed state to the int the binding carries.
|
||||
type connectionListenerAdapter struct {
|
||||
ConnectionListener
|
||||
}
|
||||
|
||||
func (a connectionListenerAdapter) OnStateChanged(state peer.ClientState) {
|
||||
a.ConnectionListener.OnStateChanged(int(state))
|
||||
}
|
||||
@@ -191,49 +191,40 @@ func (a *Auth) login(urlOpener URLOpener, isAndroidTV bool) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// loginHintSetter is implemented by both concrete flows (PKCE and device code)
|
||||
// but absent from the OAuthFlow interface, hence the assertion below — the same
|
||||
// way internal/auth wires it in authenticateWithPKCEFlow.
|
||||
type loginHintSetter interface {
|
||||
SetLoginHint(hint string)
|
||||
}
|
||||
|
||||
func (a *Auth) foregroundGetTokenInfo(authClient *auth.Auth, urlOpener URLOpener, isAndroidTV bool) (*auth.TokenInfo, error) {
|
||||
oAuthFlow, err := authClient.GetOAuthFlow(a.ctx, isAndroidTV, profileLoginHint(a.cfgPath))
|
||||
oAuthFlow, err := authClient.GetOAuthFlow(a.ctx, isAndroidTV)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get OAuth flow: %v", err)
|
||||
}
|
||||
|
||||
return runOAuthFlow(a.ctx, oAuthFlow, urlOpener, nil)
|
||||
}
|
||||
|
||||
// profileLoginHint returns the stored account email for the profile at cfgPath.
|
||||
// An empty hint is deliberate, not a fallback: a fresh profile leaves the
|
||||
// choice to the IdP. Switching accounts is done by switching or removing
|
||||
// profiles, not by logging out — logout keeps the email.
|
||||
func profileLoginHint(cfgPath string) string {
|
||||
if cfgPath == "" {
|
||||
return ""
|
||||
// An empty hint is deliberate, not a fallback: a fresh profile leaves the
|
||||
// choice to the IdP. Switching accounts is done by switching or removing
|
||||
// profiles, not by logging out — logout keeps the email.
|
||||
if a.cfgPath != "" {
|
||||
if hint := readProfileEmail(a.cfgPath); hint != "" {
|
||||
if setter, ok := oAuthFlow.(loginHintSetter); ok {
|
||||
setter.SetLoginHint(hint)
|
||||
}
|
||||
}
|
||||
}
|
||||
return readProfileEmail(cfgPath)
|
||||
}
|
||||
|
||||
// runOAuthFlow drives an already acquired OAuth flow to a token: requests the
|
||||
// flow info, presents the verification URL through the opener and waits for
|
||||
// the browser round-trip. Open is called synchronously — it is what marks the
|
||||
// surface as opened on the client side, and a fast token's OnLoginSuccess is
|
||||
// a no-op until it has, so the dismissal would be dropped rather than
|
||||
// delayed. Openers must therefore not block: they post their UI work and
|
||||
// return. onWaiting, when set, runs after the URL is shown, right before the
|
||||
// blocking wait.
|
||||
func runOAuthFlow(ctx context.Context, flow auth.OAuthFlow, urlOpener URLOpener, onWaiting func()) (*auth.TokenInfo, error) {
|
||||
flowInfo, err := flow.RequestAuthInfo(ctx)
|
||||
flowInfo, err := oAuthFlow.RequestAuthInfo(context.TODO())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("request auth info: %w", err)
|
||||
return nil, fmt.Errorf("getting a request OAuth flow info failed: %v", err)
|
||||
}
|
||||
|
||||
urlOpener.Open(flowInfo.VerificationURIComplete, flowInfo.UserCode)
|
||||
go urlOpener.Open(flowInfo.VerificationURIComplete, flowInfo.UserCode)
|
||||
|
||||
if onWaiting != nil {
|
||||
onWaiting()
|
||||
}
|
||||
|
||||
tokenInfo, err := flow.WaitToken(ctx, flowInfo)
|
||||
tokenInfo, err := oAuthFlow.WaitToken(a.ctx, flowInfo)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("wait for token: %w", err)
|
||||
return nil, fmt.Errorf("waiting for browser login failed: %v", err)
|
||||
}
|
||||
|
||||
return &tokenInfo, nil
|
||||
|
||||
@@ -1,38 +0,0 @@
|
||||
//go:build android
|
||||
|
||||
package android
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/netbirdio/netbird/client/internal/profilemanager"
|
||||
)
|
||||
|
||||
type prefsStore interface {
|
||||
Get(namespace string, v any) (bool, error)
|
||||
Put(namespace string, v any) error
|
||||
}
|
||||
|
||||
type profilePrefs struct {
|
||||
prefs *profilemanager.Prefs
|
||||
}
|
||||
|
||||
func newProfilePrefs(configDir, profileID string) (*profilePrefs, error) {
|
||||
if configDir == "" || profileID == "" {
|
||||
return nil, fmt.Errorf("profile prefs require a config dir and profile ID")
|
||||
}
|
||||
pm := NewProfileManager(configDir)
|
||||
prefs, err := pm.serviceMgr.ProfilePrefs(profilemanager.ID(profileID), androidUsername)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("resolve profile prefs: %w", err)
|
||||
}
|
||||
return &profilePrefs{prefs: prefs}, nil
|
||||
}
|
||||
|
||||
func (p *profilePrefs) Get(namespace string, v any) (bool, error) {
|
||||
return p.prefs.Get(namespace, v)
|
||||
}
|
||||
|
||||
func (p *profilePrefs) Put(namespace string, v any) error {
|
||||
return p.prefs.Put(namespace, v)
|
||||
}
|
||||
@@ -1,649 +0,0 @@
|
||||
//go:build android
|
||||
|
||||
package android
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
gossh "golang.org/x/crypto/ssh"
|
||||
|
||||
"github.com/netbirdio/netbird/client/internal"
|
||||
"github.com/netbirdio/netbird/client/internal/auth"
|
||||
"github.com/netbirdio/netbird/client/internal/profilemanager"
|
||||
nbssh "github.com/netbirdio/netbird/client/ssh"
|
||||
"github.com/netbirdio/netbird/client/ssh/detection"
|
||||
)
|
||||
|
||||
const (
|
||||
sshDialTimeout = 30 * time.Second
|
||||
sshDetectionTimeout = 5 * time.Second
|
||||
)
|
||||
|
||||
// PasswordRequiredMarker tells Java to prompt for a password and retry. It is
|
||||
// a string because gomobile flattens errors to their message, so a sentinel
|
||||
// value would not survive the binding.
|
||||
const PasswordRequiredMarker = "netbird-ssh-password-required"
|
||||
|
||||
// HostKeyUnknownMarker tells Java to show the fingerprint and, on confirmation,
|
||||
// retry with TrustHostKey set. The presented fingerprint is appended after the
|
||||
// marker so the prompt can display it and the retry can guard against a key
|
||||
// that changed between the two connects. Only regular (non-NetBird) servers
|
||||
// reach this: NetBird peers verify against the registry.
|
||||
const HostKeyUnknownMarker = "netbird-ssh-hostkey-unknown"
|
||||
|
||||
var (
|
||||
errPasswordRequired = errors.New(PasswordRequiredMarker)
|
||||
errClientClosed = errors.New("ssh client closed")
|
||||
)
|
||||
|
||||
// errHostKeyUnknown carries the presented fingerprint so Connect can build the
|
||||
// marker message the Java side parses.
|
||||
type errHostKeyUnknown struct {
|
||||
fingerprint string
|
||||
}
|
||||
|
||||
func (e *errHostKeyUnknown) Error() string {
|
||||
return HostKeyUnknownMarker + ":" + e.fingerprint
|
||||
}
|
||||
|
||||
// SSHTerminalListener receives SSH session events. It is implemented in Java.
|
||||
//
|
||||
// All callbacks are invoked from goroutines and may run concurrently with each
|
||||
// other; the implementation must be safe to call from any thread.
|
||||
type SSHTerminalListener interface {
|
||||
OnConnected()
|
||||
OnData(data []byte)
|
||||
OnClose(reason string)
|
||||
OnError(message string)
|
||||
}
|
||||
|
||||
// SSHClient is a NetBird-aware SSH client exposed to Java via gomobile.
|
||||
//
|
||||
// It dials through the running NetBird tunnel and runs a standard SSH session
|
||||
// on top with PTY enabled. Host-key verification uses the NetBird-provided
|
||||
// peer SSH host keys, identical to the desktop client.
|
||||
type SSHClient struct {
|
||||
nb *Client
|
||||
mu sync.Mutex
|
||||
listener SSHTerminalListener
|
||||
urlOpener URLOpener
|
||||
|
||||
sshClient *gossh.Client
|
||||
session *gossh.Session
|
||||
stdin io.WriteCloser
|
||||
closed bool
|
||||
|
||||
// gen identifies the current connection attempt. Connect and Close bump it,
|
||||
// so an in-flight dial or a reader left over from a previous connection
|
||||
// finds itself stale and stays silent instead of publishing OnConnected or
|
||||
// OnClose for a connection the caller already abandoned.
|
||||
gen uint64
|
||||
dialCancel context.CancelFunc
|
||||
|
||||
// knownHostsConfigDir and knownHostsProfile locate the TOFU store for
|
||||
// regular SSH servers in the profile's preferences. Java supplies them,
|
||||
// since an overlay IP is a different host under a different profile. Empty
|
||||
// until set: without them a regular server cannot be verified and Connect
|
||||
// refuses one.
|
||||
knownHostsConfigDir string
|
||||
knownHostsProfile string
|
||||
// trustHostKey carries the fingerprint the user confirmed on a previous
|
||||
// attempt, so the retry accepts exactly that key and persists it.
|
||||
trustHostKey string
|
||||
}
|
||||
|
||||
// NewSSHClient creates a new SSH client bound to the running NetBird Client.
|
||||
func NewSSHClient(c *Client) *SSHClient {
|
||||
return &SSHClient{nb: c}
|
||||
}
|
||||
|
||||
// SetListener registers the Java listener. Must be called before Connect to
|
||||
// receive any events.
|
||||
func (s *SSHClient) SetListener(l SSHTerminalListener) {
|
||||
s.mu.Lock()
|
||||
s.listener = l
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// SetURLOpener registers the Java URL opener used to display the device-code
|
||||
// authorization page in a Custom Tabs window when the target peer requires
|
||||
// JWT authentication. Must be set before Connect to be effective.
|
||||
func (s *SSHClient) SetURLOpener(opener URLOpener) {
|
||||
s.mu.Lock()
|
||||
s.urlOpener = opener
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// SetKnownHostsStore points the TOFU host-key store at a profile's preferences.
|
||||
// Must be set before connecting to a regular SSH server; without it such a
|
||||
// server cannot be verified and Connect refuses one.
|
||||
func (s *SSHClient) SetKnownHostsStore(configDir, profileID string) {
|
||||
s.mu.Lock()
|
||||
s.knownHostsConfigDir = configDir
|
||||
s.knownHostsProfile = profileID
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// TrustHostKey records the fingerprint the user confirmed for a regular server,
|
||||
// so the next Connect accepts that exact key and adds it to the known-hosts
|
||||
// store. Passing a fingerprint that no longer matches makes the connect fail
|
||||
// rather than trust a key that changed since the prompt.
|
||||
func (s *SSHClient) TrustHostKey(fingerprint string) {
|
||||
s.mu.Lock()
|
||||
s.trustHostKey = fingerprint
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// Connect dials the SSH server through the NetBird tunnel and performs the
|
||||
// SSH handshake. It auto-detects the server type via SSH banner inspection
|
||||
// and selects the appropriate authentication path:
|
||||
//
|
||||
// - NetBird-SSH server requiring JWT: launches the OAuth 2.0 device-code
|
||||
// flow, opens the verification URL through the registered URLOpener, and
|
||||
// uses the resulting token as the SSH password. Host-key verification
|
||||
// uses the NetBird peer registry.
|
||||
// - NetBird-SSH server without JWT: authenticates with the NetBird SSH
|
||||
// private key. Host-key verification uses the NetBird peer registry.
|
||||
// - Regular SSH server (e.g. OpenSSH): authenticates with the NetBird key
|
||||
// first (so a user-installed NetBird public key works), then falls back
|
||||
// to the supplied password if non-empty. Host-key verification is
|
||||
// trust-on-first-use against the per-profile known-hosts store.
|
||||
//
|
||||
// The password parameter is only consulted for regular SSH servers.
|
||||
func (s *SSHClient) Connect(host string, port int, user, password string) error {
|
||||
if port < 1 || port > 65535 {
|
||||
return fmt.Errorf("invalid port: %d", port)
|
||||
}
|
||||
|
||||
cfg, cfgPath, cc := s.nb.authSnapshot()
|
||||
if cc == nil {
|
||||
return errors.New("netbird client not running")
|
||||
}
|
||||
if cfg == nil {
|
||||
return errors.New("netbird config not loaded")
|
||||
}
|
||||
engine := cc.Engine()
|
||||
if engine == nil {
|
||||
return errors.New("netbird engine not available")
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
s.gen++
|
||||
gen := s.gen
|
||||
s.mu.Unlock()
|
||||
|
||||
serverType := detectServerType(host, port)
|
||||
log.Debugf("SSH server type: %s", serverType)
|
||||
|
||||
authMethods, hostKeyCallback, err := s.buildAuth(cfg, cfgPath, engine, serverType, password)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
clientConfig := &gossh.ClientConfig{
|
||||
User: user,
|
||||
Auth: authMethods,
|
||||
HostKeyCallback: hostKeyCallback,
|
||||
Timeout: sshDialTimeout,
|
||||
}
|
||||
err = s.dialAndHandshake(gen, host, port, clientConfig)
|
||||
|
||||
// An unknown host key is a prompt, not a failure: return the marker intact
|
||||
// (rootCause would unwrap it) so Java can show the fingerprint and retry.
|
||||
var unknownHost *errHostKeyUnknown
|
||||
if errors.As(err, &unknownHost) {
|
||||
return errors.New(unknownHost.Error())
|
||||
}
|
||||
|
||||
// A regular server may still accept a password, so let the caller ask for
|
||||
// one instead of failing. NetBird servers never use a password, so a
|
||||
// failure there is genuine.
|
||||
if err != nil && serverType != detection.ServerTypeNetBirdJWT &&
|
||||
serverType != detection.ServerTypeNetBirdNoJWT && isAuthFailure(err) &&
|
||||
passwordCouldHelp(err, password != "") {
|
||||
return errPasswordRequired
|
||||
}
|
||||
if err != nil {
|
||||
return rootCause(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// StartSession requests a PTY and starts an interactive shell. Output from
|
||||
// the session is forwarded to the listener via OnData.
|
||||
func (s *SSHClient) StartSession(cols, rows int) error {
|
||||
err := s.startSession(cols, rows)
|
||||
if err != nil {
|
||||
log.Infof("SSH: start session failed: %v", err)
|
||||
return rootCause(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Write sends data to the SSH session stdin.
|
||||
func (s *SSHClient) Write(data []byte) error {
|
||||
s.mu.Lock()
|
||||
stdin := s.stdin
|
||||
s.mu.Unlock()
|
||||
if stdin == nil {
|
||||
return errors.New("ssh session not started")
|
||||
}
|
||||
if _, err := stdin.Write(data); err != nil {
|
||||
return fmt.Errorf("write stdin: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Resize updates the PTY window size.
|
||||
func (s *SSHClient) Resize(cols, rows int) error {
|
||||
s.mu.Lock()
|
||||
session := s.session
|
||||
s.mu.Unlock()
|
||||
if session == nil {
|
||||
return errors.New("ssh session not started")
|
||||
}
|
||||
return session.WindowChange(rows, cols)
|
||||
}
|
||||
|
||||
// Reset makes a closed client usable for another Connect: Close leaves the
|
||||
// one-shot guard set, and clearing it lets the same client back a reconnect.
|
||||
func (s *SSHClient) Reset() {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.closed = false
|
||||
}
|
||||
|
||||
// Close terminates the SSH session and underlying connection. Safe to call
|
||||
// multiple times.
|
||||
func (s *SSHClient) Close() error {
|
||||
s.mu.Lock()
|
||||
s.gen++
|
||||
if s.dialCancel != nil {
|
||||
s.dialCancel()
|
||||
s.dialCancel = nil
|
||||
}
|
||||
sshClient := s.sshClient
|
||||
session := s.session
|
||||
stdin := s.stdin
|
||||
s.sshClient = nil
|
||||
s.session = nil
|
||||
s.stdin = nil
|
||||
notify := !s.closed
|
||||
s.closed = true
|
||||
listener := s.listener
|
||||
s.mu.Unlock()
|
||||
|
||||
if stdin != nil {
|
||||
if err := stdin.Close(); err != nil {
|
||||
log.Debugf("ssh: stdin close: %v", err)
|
||||
}
|
||||
}
|
||||
if session != nil {
|
||||
if err := session.Close(); err != nil && !errors.Is(err, io.EOF) {
|
||||
log.Debugf("ssh: session close: %v", err)
|
||||
}
|
||||
}
|
||||
var firstErr error
|
||||
if sshClient != nil {
|
||||
if err := sshClient.Close(); err != nil {
|
||||
firstErr = err
|
||||
}
|
||||
}
|
||||
if notify && listener != nil {
|
||||
listener.OnClose("closed by client")
|
||||
}
|
||||
return firstErr
|
||||
}
|
||||
|
||||
func (s *SSHClient) startSession(cols, rows int) error {
|
||||
log.Debugf("SSH: starting session %dx%d", cols, rows)
|
||||
s.mu.Lock()
|
||||
sshClient := s.sshClient
|
||||
gen := s.gen
|
||||
s.mu.Unlock()
|
||||
|
||||
if sshClient == nil {
|
||||
return errors.New("ssh client not connected")
|
||||
}
|
||||
|
||||
pty, err := nbssh.StartPTYSession(sshClient, cols, rows)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
if gen != s.gen {
|
||||
s.mu.Unlock()
|
||||
closeQuiet(pty.Session, "stale session")
|
||||
return errClientClosed
|
||||
}
|
||||
s.session = pty.Session
|
||||
s.stdin = pty.Stdin
|
||||
s.mu.Unlock()
|
||||
|
||||
readerDone := make(chan string, 2)
|
||||
go func() { readerDone <- s.readLoop(pty.Stdout, "stdout") }()
|
||||
go func() { readerDone <- s.readLoop(pty.Stderr, "stderr") }()
|
||||
go func() {
|
||||
reason := <-readerDone
|
||||
if second := <-readerDone; reason == "" {
|
||||
reason = second
|
||||
}
|
||||
s.notifyClose(gen, reason)
|
||||
}()
|
||||
log.Debug("SSH: session started, shell running")
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *SSHClient) buildAuth(cfg *profilemanager.Config, cfgPath string, engine *internal.Engine,
|
||||
serverType detection.ServerType, password string) ([]gossh.AuthMethod, gossh.HostKeyCallback, error) {
|
||||
|
||||
switch serverType {
|
||||
case detection.ServerTypeNetBirdJWT:
|
||||
token, err := s.requestJWTToken(cfg, cfgPath)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("jwt: %w", err)
|
||||
}
|
||||
auths := []gossh.AuthMethod{gossh.Password(token)}
|
||||
return auths, nbssh.CreateHostKeyCallback(nbssh.PeerKeyLookup(engine.GetPeerSSHKey)), nil
|
||||
|
||||
case detection.ServerTypeNetBirdNoJWT:
|
||||
if cfg.SSHKey == "" {
|
||||
return nil, nil, errors.New("no NetBird SSH key available")
|
||||
}
|
||||
signer, err := gossh.ParsePrivateKey([]byte(cfg.SSHKey))
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("parse netbird ssh key: %w", err)
|
||||
}
|
||||
auths := []gossh.AuthMethod{gossh.PublicKeys(signer)}
|
||||
return auths, nbssh.CreateHostKeyCallback(nbssh.PeerKeyLookup(engine.GetPeerSSHKey)), nil
|
||||
|
||||
case detection.ServerTypeRegular:
|
||||
var auths []gossh.AuthMethod
|
||||
if cfg.SSHKey != "" {
|
||||
if signer, err := gossh.ParsePrivateKey([]byte(cfg.SSHKey)); err == nil {
|
||||
auths = append(auths, gossh.PublicKeys(signer))
|
||||
} else {
|
||||
log.Debugf("ssh: parse netbird key for regular auth: %v", err)
|
||||
}
|
||||
}
|
||||
if password != "" {
|
||||
pw := password
|
||||
auths = append(auths, gossh.Password(pw))
|
||||
auths = append(auths, gossh.KeyboardInteractive(func(_, _ string, questions []string, _ []bool) ([]string, error) {
|
||||
answers := make([]string, len(questions))
|
||||
for i := range questions {
|
||||
answers[i] = pw
|
||||
}
|
||||
return answers, nil
|
||||
}))
|
||||
}
|
||||
if len(auths) == 0 {
|
||||
// Nothing to offer at all: ask for a password rather than failing,
|
||||
// so the caller can retry once the user supplies one.
|
||||
return nil, nil, errPasswordRequired
|
||||
}
|
||||
callback, err := s.tofuHostKeyCallback()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return auths, callback, nil
|
||||
|
||||
default:
|
||||
return nil, nil, fmt.Errorf("unsupported SSH server type: %v", serverType)
|
||||
}
|
||||
}
|
||||
|
||||
// tofuHostKeyCallback verifies a regular server's host key against the
|
||||
// per-profile known-hosts store. An unknown host returns errHostKeyUnknown so
|
||||
// Java can show the fingerprint and, once confirmed, retry with the key
|
||||
// trusted; a changed key is rejected outright, as OpenSSH does. When the user
|
||||
// has confirmed a fingerprint, the callback accepts exactly that key and
|
||||
// appends it to the store.
|
||||
func (s *SSHClient) tofuHostKeyCallback() (gossh.HostKeyCallback, error) {
|
||||
s.mu.Lock()
|
||||
configDir := s.knownHostsConfigDir
|
||||
profileID := s.knownHostsProfile
|
||||
trusted := s.trustHostKey
|
||||
s.mu.Unlock()
|
||||
|
||||
if configDir == "" || profileID == "" {
|
||||
return nil, errors.New("no known-hosts store configured for regular SSH")
|
||||
}
|
||||
|
||||
store, err := openKnownHostsStore(configDir, profileID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("load known-hosts store: %w", err)
|
||||
}
|
||||
|
||||
return func(hostname string, remote net.Addr, key gossh.PublicKey) error {
|
||||
verdict, err := store.verify(hostname, remote, key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if verdict == hostKeyMatched {
|
||||
return nil
|
||||
}
|
||||
if verdict == hostKeyChanged {
|
||||
return fmt.Errorf("SSH host key changed for %s (possible attack)", hostname)
|
||||
}
|
||||
|
||||
fingerprint := gossh.FingerprintSHA256(key)
|
||||
if trusted == "" {
|
||||
return &errHostKeyUnknown{fingerprint: fingerprint}
|
||||
}
|
||||
if trusted != fingerprint {
|
||||
return fmt.Errorf("SSH host key changed since it was confirmed for %s", hostname)
|
||||
}
|
||||
if err := store.append(hostname, remote, key); err != nil {
|
||||
return fmt.Errorf("persist trusted host key: %w", err)
|
||||
}
|
||||
// The confirmation is spent: now that the key is stored, a later
|
||||
// reconnect must verify against the file, not re-accept this fingerprint.
|
||||
s.mu.Lock()
|
||||
s.trustHostKey = ""
|
||||
s.mu.Unlock()
|
||||
return nil
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *SSHClient) requestJWTToken(cfg *profilemanager.Config, cfgPath string) (string, error) {
|
||||
s.mu.Lock()
|
||||
urlOpener := s.urlOpener
|
||||
s.mu.Unlock()
|
||||
if urlOpener == nil {
|
||||
return "", errors.New("URL opener not configured for JWT auth")
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
flow, err := auth.NewOAuthFlow(ctx, cfg, false, true, profileLoginHint(cfgPath))
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("create oauth flow: %w", err)
|
||||
}
|
||||
|
||||
// The status callback covers the browser round-trip, which would
|
||||
// otherwise leave the terminal blank.
|
||||
tokenInfo, err := runOAuthFlow(ctx, flow, urlOpener, func() {
|
||||
s.notifyStatus("Waiting for browser authentication...")
|
||||
})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
token := tokenInfo.GetTokenToUse()
|
||||
if token == "" {
|
||||
return "", errors.New("empty token returned by IdP")
|
||||
}
|
||||
|
||||
// Tells the client the browser round-trip is over so it can dismiss the
|
||||
// surface it opened, the same way the login and session-extend flows do.
|
||||
// Without it the Custom Tab stays in front of the terminal even though the
|
||||
// token has already been collected.
|
||||
urlOpener.OnLoginSuccess()
|
||||
|
||||
return token, nil
|
||||
}
|
||||
|
||||
func (s *SSHClient) dialAndHandshake(gen uint64, host string, port int, clientConfig *gossh.ClientConfig) error {
|
||||
addr := net.JoinHostPort(host, strconv.Itoa(port))
|
||||
ctx, cancel := context.WithTimeout(context.Background(), sshDialTimeout)
|
||||
defer cancel()
|
||||
|
||||
s.mu.Lock()
|
||||
if gen != s.gen {
|
||||
s.mu.Unlock()
|
||||
return errClientClosed
|
||||
}
|
||||
s.dialCancel = cancel
|
||||
s.mu.Unlock()
|
||||
|
||||
var dialer net.Dialer
|
||||
conn, err := dialer.DialContext(ctx, "tcp", addr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("dial %s: %w", addr, err)
|
||||
}
|
||||
|
||||
client, err := nbssh.Handshake(ctx, conn, addr, clientConfig)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
if gen != s.gen {
|
||||
s.mu.Unlock()
|
||||
closeQuiet(client, "stale ssh client")
|
||||
return errClientClosed
|
||||
}
|
||||
s.sshClient = client
|
||||
listener := s.listener
|
||||
s.mu.Unlock()
|
||||
|
||||
if listener != nil {
|
||||
listener.OnConnected()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *SSHClient) readLoop(r io.Reader, name string) string {
|
||||
buf := make([]byte, 4096)
|
||||
for {
|
||||
n, err := r.Read(buf)
|
||||
if n > 0 {
|
||||
s.mu.Lock()
|
||||
listener := s.listener
|
||||
s.mu.Unlock()
|
||||
if listener != nil {
|
||||
chunk := make([]byte, n)
|
||||
copy(chunk, buf[:n])
|
||||
listener.OnData(chunk)
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
// EOF is a normal shell exit, so report it without a reason.
|
||||
if errors.Is(err, io.EOF) {
|
||||
return ""
|
||||
}
|
||||
log.Debugf("ssh %s read: %v", name, err)
|
||||
return rootCause(err).Error()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// notifyStatus writes a progress line to the terminal through the normal
|
||||
// output path, so long steps are visible while nothing else is arriving.
|
||||
func (s *SSHClient) notifyStatus(text string) {
|
||||
s.mu.Lock()
|
||||
listener := s.listener
|
||||
s.mu.Unlock()
|
||||
if listener != nil {
|
||||
listener.OnData([]byte("\r\n\x1b[33m" + text + "\x1b[0m\r\n"))
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SSHClient) notifyClose(gen uint64, reason string) {
|
||||
s.mu.Lock()
|
||||
if gen != s.gen || s.closed {
|
||||
s.mu.Unlock()
|
||||
return
|
||||
}
|
||||
s.closed = true
|
||||
listener := s.listener
|
||||
s.mu.Unlock()
|
||||
if listener != nil {
|
||||
listener.OnClose(reason)
|
||||
}
|
||||
}
|
||||
|
||||
func closeQuiet(c io.Closer, label string) {
|
||||
if c == nil {
|
||||
return
|
||||
}
|
||||
if err := c.Close(); err != nil && !errors.Is(err, io.EOF) {
|
||||
log.Debugf("ssh: close %s: %v", label, err)
|
||||
}
|
||||
}
|
||||
|
||||
func detectServerType(host string, port int) detection.ServerType {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), sshDetectionTimeout)
|
||||
defer cancel()
|
||||
|
||||
dialer := &net.Dialer{}
|
||||
serverType, err := detection.DetectSSHServerType(ctx, dialer, host, port)
|
||||
if err != nil {
|
||||
log.Debugf("ssh: server detection failed: %v (assuming regular SSH)", err)
|
||||
return detection.ServerTypeRegular
|
||||
}
|
||||
return serverType
|
||||
}
|
||||
|
||||
// rootCause returns the innermost error of a %w chain, so the terminal shows
|
||||
// "i/o timeout" rather than every layer that added context on the way up.
|
||||
func rootCause(err error) error {
|
||||
for {
|
||||
// A joined error has no single root, so keep it as-is.
|
||||
if _, ok := err.(interface{ Unwrap() []error }); ok {
|
||||
return err
|
||||
}
|
||||
next := errors.Unwrap(err)
|
||||
if next == nil {
|
||||
return err
|
||||
}
|
||||
err = next
|
||||
}
|
||||
}
|
||||
|
||||
// isAuthFailure distinguishes credential rejection from dial, timeout and
|
||||
// host-key errors, which retrying with a password would not fix.
|
||||
func isAuthFailure(err error) bool {
|
||||
if errors.Is(err, errPasswordRequired) {
|
||||
return true
|
||||
}
|
||||
var partial *gossh.PartialSuccessError
|
||||
if errors.As(err, &partial) {
|
||||
return true
|
||||
}
|
||||
return strings.Contains(err.Error(), "unable to authenticate")
|
||||
}
|
||||
|
||||
// passwordCouldHelp reports whether prompting for a password again can change
|
||||
// the outcome. gossh lists a method under "attempted methods" only when the
|
||||
// server offered it, so a supplied password that was never attempted means the
|
||||
// server does not accept passwords and the real error should surface instead.
|
||||
func passwordCouldHelp(err error, passwordOffered bool) bool {
|
||||
if !passwordOffered {
|
||||
return true
|
||||
}
|
||||
msg := err.Error()
|
||||
return strings.Contains(msg, "password") || strings.Contains(msg, "keyboard-interactive")
|
||||
}
|
||||
@@ -1,168 +0,0 @@
|
||||
//go:build android
|
||||
|
||||
package android
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"net"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
gossh "golang.org/x/crypto/ssh"
|
||||
"golang.org/x/crypto/ssh/knownhosts"
|
||||
)
|
||||
|
||||
const knownHostsNamespace = "ssh"
|
||||
|
||||
const (
|
||||
hostKeyUnknown hostKeyVerdict = iota
|
||||
hostKeyMatched
|
||||
hostKeyChanged
|
||||
)
|
||||
|
||||
var knownHostsMu sync.Mutex
|
||||
|
||||
type hostKeyVerdict uint8
|
||||
|
||||
type knownHostsSection struct {
|
||||
KnownHosts []string `json:"knownHosts"`
|
||||
}
|
||||
|
||||
type knownHostsStore struct {
|
||||
prefs prefsStore
|
||||
}
|
||||
|
||||
// RemoveKnownHost deletes every known-hosts entry for host:port from the
|
||||
// profile's store, so a host trusted for a session that is being deleted does
|
||||
// not linger. Java calls this only once no session targets that host, so a
|
||||
// shared host stays trusted. A missing entry is not an error: the goal state
|
||||
// is "absent".
|
||||
func RemoveKnownHost(configDir, profileID, host string, port int) error {
|
||||
store, err := openKnownHostsStore(configDir, profileID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return store.removeHost(host, port)
|
||||
}
|
||||
|
||||
func openKnownHostsStore(configDir, profileID string) (*knownHostsStore, error) {
|
||||
prefs, err := newProfilePrefs(configDir, profileID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &knownHostsStore{prefs: prefs}, nil
|
||||
}
|
||||
|
||||
func (st *knownHostsStore) verify(hostname string, remote net.Addr, key gossh.PublicKey) (hostKeyVerdict, error) {
|
||||
lines, err := st.lines()
|
||||
if err != nil {
|
||||
return hostKeyUnknown, err
|
||||
}
|
||||
targets := knownHostsTargets(hostname, remote)
|
||||
|
||||
verdict := hostKeyUnknown
|
||||
for _, line := range lines {
|
||||
pubKey, ok := knownHostsLineKey(line, targets)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if pubKey.Type() == key.Type() && bytes.Equal(pubKey.Marshal(), key.Marshal()) {
|
||||
return hostKeyMatched, nil
|
||||
}
|
||||
verdict = hostKeyChanged
|
||||
}
|
||||
return verdict, nil
|
||||
}
|
||||
|
||||
func (st *knownHostsStore) append(hostname string, remote net.Addr, key gossh.PublicKey) error {
|
||||
line := knownhosts.Line(knownHostsTargets(hostname, remote), key)
|
||||
|
||||
knownHostsMu.Lock()
|
||||
defer knownHostsMu.Unlock()
|
||||
|
||||
lines, err := st.lines()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return st.prefs.Put(knownHostsNamespace, knownHostsSection{KnownHosts: append(lines, line)})
|
||||
}
|
||||
|
||||
func (st *knownHostsStore) removeHost(host string, port int) error {
|
||||
target := knownhosts.Normalize(net.JoinHostPort(host, strconv.Itoa(port)))
|
||||
|
||||
knownHostsMu.Lock()
|
||||
defer knownHostsMu.Unlock()
|
||||
|
||||
lines, err := st.lines()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
kept := make([]string, 0, len(lines))
|
||||
for _, line := range lines {
|
||||
if knownHostsLineMatches(line, target) {
|
||||
continue
|
||||
}
|
||||
kept = append(kept, line)
|
||||
}
|
||||
if len(kept) == len(lines) {
|
||||
return nil
|
||||
}
|
||||
return st.prefs.Put(knownHostsNamespace, knownHostsSection{KnownHosts: kept})
|
||||
}
|
||||
|
||||
func (st *knownHostsStore) lines() ([]string, error) {
|
||||
var section knownHostsSection
|
||||
if _, err := st.prefs.Get(knownHostsNamespace, §ion); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return section.KnownHosts, nil
|
||||
}
|
||||
|
||||
func knownHostsTargets(hostname string, remote net.Addr) []string {
|
||||
targets := []string{knownhosts.Normalize(hostname)}
|
||||
if remote != nil {
|
||||
if normalized := knownhosts.Normalize(remote.String()); normalized != targets[0] {
|
||||
targets = append(targets, normalized)
|
||||
}
|
||||
}
|
||||
return targets
|
||||
}
|
||||
|
||||
func knownHostsLineKey(line string, targets []string) (gossh.PublicKey, bool) {
|
||||
trimmed := strings.TrimSpace(line)
|
||||
if trimmed == "" || strings.HasPrefix(trimmed, "#") {
|
||||
return nil, false
|
||||
}
|
||||
_, hosts, pubKey, _, _, err := gossh.ParseKnownHosts([]byte(trimmed))
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
for _, host := range hosts {
|
||||
for _, target := range targets {
|
||||
if host == target {
|
||||
return pubKey, true
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// knownHostsLineMatches reports whether a known-hosts line's address list
|
||||
// contains the normalized target. Comment and blank lines never match.
|
||||
func knownHostsLineMatches(line, target string) bool {
|
||||
trimmed := strings.TrimSpace(line)
|
||||
if trimmed == "" || strings.HasPrefix(trimmed, "#") {
|
||||
return false
|
||||
}
|
||||
fields := strings.Fields(trimmed)
|
||||
if len(fields) == 0 {
|
||||
return false
|
||||
}
|
||||
for _, addr := range strings.Split(fields[0], ",") {
|
||||
if addr == target {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -1,104 +0,0 @@
|
||||
//go:build android
|
||||
|
||||
package android
|
||||
|
||||
const (
|
||||
sshSessionsNamespace = "ssh-sessions"
|
||||
maxStoredSSHSessions = 50
|
||||
)
|
||||
|
||||
type sshSessionRecord struct {
|
||||
ID string `json:"id"`
|
||||
Host string `json:"host"`
|
||||
Port int `json:"port"`
|
||||
User string `json:"user"`
|
||||
}
|
||||
|
||||
type sshSessionsSection struct {
|
||||
Sessions []sshSessionRecord `json:"sessions"`
|
||||
}
|
||||
|
||||
// SSHSessionEntry is one stored SSH session, without any credential.
|
||||
type SSHSessionEntry struct {
|
||||
ID string
|
||||
Host string
|
||||
Port int
|
||||
User string
|
||||
}
|
||||
|
||||
// SSHSessionArray wraps stored SSH sessions for gomobile compatibility.
|
||||
type SSHSessionArray struct {
|
||||
items []*SSHSessionEntry
|
||||
}
|
||||
|
||||
// NewSSHSessionArray creates an empty session array to fill via Add.
|
||||
func NewSSHSessionArray() *SSHSessionArray {
|
||||
return &SSHSessionArray{}
|
||||
}
|
||||
|
||||
// Add appends a session entry, oldest first.
|
||||
func (a *SSHSessionArray) Add(id, host string, port int, user string) {
|
||||
a.items = append(a.items, &SSHSessionEntry{ID: id, Host: host, Port: port, User: user})
|
||||
}
|
||||
|
||||
// Length returns the number of entries.
|
||||
func (a *SSHSessionArray) Length() int {
|
||||
return len(a.items)
|
||||
}
|
||||
|
||||
// Get returns the entry at index i, or nil when out of range.
|
||||
func (a *SSHSessionArray) Get(i int) *SSHSessionEntry {
|
||||
if i < 0 || i >= len(a.items) {
|
||||
return nil
|
||||
}
|
||||
return a.items[i]
|
||||
}
|
||||
|
||||
// SSHSessionStore reads and writes a profile's stored SSH sessions.
|
||||
type SSHSessionStore struct {
|
||||
prefs prefsStore
|
||||
}
|
||||
|
||||
// NewSSHSessionStore opens the session store of the given profile.
|
||||
func NewSSHSessionStore(configDir, profileID string) (*SSHSessionStore, error) {
|
||||
prefs, err := newProfilePrefs(configDir, profileID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &SSHSessionStore{prefs: prefs}, nil
|
||||
}
|
||||
|
||||
// Load returns the stored sessions, oldest first.
|
||||
func (s *SSHSessionStore) Load() (*SSHSessionArray, error) {
|
||||
var section sshSessionsSection
|
||||
if _, err := s.prefs.Get(sshSessionsNamespace, §ion); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
out := NewSSHSessionArray()
|
||||
for _, record := range section.Sessions {
|
||||
if record.ID == "" || record.Host == "" {
|
||||
continue
|
||||
}
|
||||
out.Add(record.ID, record.Host, record.Port, record.User)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Save replaces the stored sessions, keeping only the newest entries when the
|
||||
// list exceeds the storage cap.
|
||||
func (s *SSHSessionStore) Save(sessions *SSHSessionArray) error {
|
||||
var items []*SSHSessionEntry
|
||||
if sessions != nil {
|
||||
items = sessions.items
|
||||
}
|
||||
if len(items) > maxStoredSSHSessions {
|
||||
items = items[len(items)-maxStoredSSHSessions:]
|
||||
}
|
||||
|
||||
records := make([]sshSessionRecord, 0, len(items))
|
||||
for _, item := range items {
|
||||
records = append(records, sshSessionRecord{ID: item.ID, Host: item.Host, Port: item.Port, User: item.User})
|
||||
}
|
||||
return s.prefs.Put(sshSessionsNamespace, sshSessionsSection{Sessions: records})
|
||||
}
|
||||
@@ -45,8 +45,8 @@ func daemonServerOptions(network string) []grpc.ServerOption {
|
||||
return nil
|
||||
}
|
||||
|
||||
creds := ipcauth.NewTransportCredentials() //nolint:staticcheck
|
||||
if creds == nil { //nolint:staticcheck // nil only on platforms without a peer-identity primitive
|
||||
creds := ipcauth.NewTransportCredentials()
|
||||
if creds == nil {
|
||||
log.Warnf("daemon IPC has no peer-identity primitive on %s: privileged operations will be denied", runtime.GOOS)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -27,8 +27,8 @@ func listenOnAddress(addr string) (*socketListener, error) {
|
||||
}
|
||||
|
||||
if network == "npipe" {
|
||||
listener, path, err := listenNamedPipe(address) //nolint:staticcheck
|
||||
if err != nil { //nolint:staticcheck // always errors on non-Windows builds
|
||||
listener, path, err := listenNamedPipe(address)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &socketListener{Listener: listener, network: network, address: path}, nil
|
||||
|
||||
@@ -6,7 +6,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go.uber.org/mock/gomock"
|
||||
"github.com/golang/mock/gomock"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.opentelemetry.io/otel"
|
||||
"google.golang.org/grpc"
|
||||
|
||||
@@ -21,7 +21,7 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/auth"
|
||||
"github.com/netbirdio/netbird/client/internal/peer"
|
||||
"github.com/netbirdio/netbird/client/internal/profilemanager"
|
||||
nbssh "github.com/netbirdio/netbird/client/ssh"
|
||||
sshcommon "github.com/netbirdio/netbird/client/ssh"
|
||||
"github.com/netbirdio/netbird/client/system"
|
||||
"github.com/netbirdio/netbird/shared/management/domain"
|
||||
mgmProto "github.com/netbirdio/netbird/shared/management/proto"
|
||||
@@ -91,13 +91,6 @@ type Options struct {
|
||||
// when the embedded client must never act as a stepping stone into
|
||||
// the host's local network (e.g. the proxy's overlay peer).
|
||||
BlockLANAccess bool
|
||||
// LazyConnectionEnabled is a tri-state local override for lazy connections,
|
||||
// mirroring the NB_LAZY_CONN env var. Nil defers to the management feature
|
||||
// flag; a set value overrides it in both directions. A short-lived client
|
||||
// that reaches only a few known peers can set this to false, so its peers
|
||||
// connect eagerly and the first request does not wait for the connection to
|
||||
// be established.
|
||||
LazyConnectionEnabled *bool
|
||||
// WireguardPort is the port for the tunnel interface. Use 0 for a random port.
|
||||
WireguardPort *int
|
||||
// MTU is the MTU for the tunnel interface.
|
||||
@@ -227,15 +220,6 @@ func New(opts Options) (*Client, error) {
|
||||
config.PrivateKey = opts.PrivateKey
|
||||
}
|
||||
|
||||
if opts.LazyConnectionEnabled != nil {
|
||||
// Runtime-only override, read back through lazyconn.ParseState; a set value
|
||||
// wins over the management feature flag in both directions.
|
||||
config.LazyConnection = "off"
|
||||
if *opts.LazyConnectionEnabled {
|
||||
config.LazyConnection = "on"
|
||||
}
|
||||
}
|
||||
|
||||
if opts.Performance.PreallocatedBuffersPerPool != nil {
|
||||
wgdevice.SetPreallocatedBuffersPerPool(*opts.Performance.PreallocatedBuffersPerPool)
|
||||
}
|
||||
@@ -537,7 +521,12 @@ func (c *Client) VerifySSHHostKey(peerAddress string, key []byte) error {
|
||||
return err
|
||||
}
|
||||
|
||||
return nbssh.PeerKeyLookup(engine.GetPeerSSHKey).VerifySSHHostKey(peerAddress, key)
|
||||
storedKey, found := engine.GetPeerSSHKey(peerAddress)
|
||||
if !found {
|
||||
return sshcommon.ErrPeerNotFound
|
||||
}
|
||||
|
||||
return sshcommon.VerifyHostKey(storedKey, key, peerAddress)
|
||||
}
|
||||
|
||||
// SetPerformance retunes a running Client. Only PreallocatedBuffersPerPool
|
||||
|
||||
@@ -6,7 +6,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go.uber.org/mock/gomock"
|
||||
"github.com/golang/mock/gomock"
|
||||
"github.com/stretchr/testify/require"
|
||||
"google.golang.org/grpc"
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@ import (
|
||||
"net/netip"
|
||||
"testing"
|
||||
|
||||
"go.uber.org/mock/gomock"
|
||||
"github.com/golang/mock/gomock"
|
||||
"github.com/google/gopacket"
|
||||
"github.com/google/gopacket/layers"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
@@ -4,7 +4,7 @@ import (
|
||||
"net/netip"
|
||||
"testing"
|
||||
|
||||
"go.uber.org/mock/gomock"
|
||||
"github.com/golang/mock/gomock"
|
||||
"github.com/google/gopacket/layers"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
@@ -16,47 +16,28 @@ import (
|
||||
"google.golang.org/grpc"
|
||||
|
||||
nbnet "github.com/netbirdio/netbird/client/net"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
)
|
||||
|
||||
func WithCustomDialer(_ bool, _ string) grpc.DialOption {
|
||||
return grpc.WithContextDialer(dialContext)
|
||||
}
|
||||
|
||||
// WithSweeper dials like WithCustomDialer but registers connections and
|
||||
// dials with the sweeper. Append it after WithCustomDialer: gRPC applies
|
||||
// dial options in order, so the later context dialer wins.
|
||||
func WithSweeper(sweeper *netsweep.Sweeper) grpc.DialOption {
|
||||
return grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
|
||||
dial := sweeper.StartDial(ctx)
|
||||
defer dial.Release()
|
||||
if runtime.GOOS == "linux" {
|
||||
currentUser, err := user.Current()
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.FailedPrecondition, "failed to get current user: %v", err)
|
||||
}
|
||||
|
||||
conn, err := dialContext(dial.Ctx(), addr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
// the custom dialer requires root permissions which are not required for use cases run as non-root
|
||||
if currentUser.Uid != "0" {
|
||||
log.Debug("Not running as root, using standard dialer")
|
||||
dialer := &net.Dialer{}
|
||||
return dialer.DialContext(ctx, "tcp", addr)
|
||||
}
|
||||
}
|
||||
return dial.WrapConn(conn)
|
||||
|
||||
conn, err := nbnet.NewDialer().DialContext(ctx, "tcp", addr)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("nbnet.NewDialer().DialContext: %w", err)
|
||||
}
|
||||
return conn, nil
|
||||
})
|
||||
}
|
||||
|
||||
func dialContext(ctx context.Context, addr string) (net.Conn, error) {
|
||||
if runtime.GOOS == "linux" {
|
||||
currentUser, err := user.Current()
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.FailedPrecondition, "failed to get current user: %v", err)
|
||||
}
|
||||
|
||||
// the custom dialer requires root permissions which are not required for use cases run as non-root
|
||||
if currentUser.Uid != "0" {
|
||||
log.Debug("Not running as root, using standard dialer")
|
||||
dialer := &net.Dialer{}
|
||||
return dialer.DialContext(ctx, "tcp", addr)
|
||||
}
|
||||
}
|
||||
|
||||
conn, err := nbnet.NewDialer().DialContext(ctx, "tcp", addr)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("nbnet.NewDialer().DialContext: %w", err)
|
||||
}
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
@@ -3,7 +3,6 @@ package grpc
|
||||
import (
|
||||
"google.golang.org/grpc"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
"github.com/netbirdio/netbird/util/wsproxy/client"
|
||||
)
|
||||
|
||||
@@ -12,8 +11,3 @@ import (
|
||||
func WithCustomDialer(tlsEnabled bool, component string) grpc.DialOption {
|
||||
return client.WithWebSocketDialer(tlsEnabled, component)
|
||||
}
|
||||
|
||||
// WithSweeper is a no-op on WASM/JS: there is no network change signal.
|
||||
func WithSweeper(_ *netsweep.Sweeper) grpc.DialOption {
|
||||
return grpc.EmptyDialOption{}
|
||||
}
|
||||
|
||||
@@ -1,49 +0,0 @@
|
||||
package grpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
// Retry mirrors backoff.Retry, but the sleep between attempts also wakes on
|
||||
// OS network availability transitions: an operation cut down by a network
|
||||
// change retries the moment the network settles instead of sleeping through
|
||||
// the recovery. A nil netState never fires, leaving plain backoff.Retry
|
||||
// behavior.
|
||||
func Retry(ctx context.Context, operation backoff.Operation, bo backoff.BackOff, netState *netstate.State) error {
|
||||
bo.Reset()
|
||||
for {
|
||||
err := operation()
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
var permanent *backoff.PermanentError
|
||||
if errors.As(err, &permanent) {
|
||||
return permanent.Err
|
||||
}
|
||||
|
||||
next := bo.NextBackOff()
|
||||
if next == backoff.Stop {
|
||||
if cerr := ctx.Err(); cerr != nil {
|
||||
return cerr
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
timer := time.NewTimer(next)
|
||||
select {
|
||||
case <-timer.C:
|
||||
case <-netState.Changed():
|
||||
timer.Stop()
|
||||
case <-ctx.Done():
|
||||
timer.Stop()
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,91 +0,0 @@
|
||||
package grpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
func TestRetryWakesOnNetworkChange(t *testing.T) {
|
||||
ns := netstate.New()
|
||||
attempts := 0
|
||||
operation := func() error {
|
||||
attempts++
|
||||
if attempts == 1 {
|
||||
return errors.New("cut by network change")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
go func() {
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
ns.Set(false)
|
||||
}()
|
||||
|
||||
start := time.Now()
|
||||
err := Retry(context.Background(), operation, backoff.NewConstantBackOff(time.Minute), ns)
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 2, attempts, "network change must cause one immediate retry")
|
||||
assert.Less(t, time.Since(start), time.Second, "the transition must cut the minute-long sleep short")
|
||||
}
|
||||
|
||||
func TestRetryPermanentError(t *testing.T) {
|
||||
sentinel := errors.New("permission denied")
|
||||
operation := func() error {
|
||||
return backoff.Permanent(sentinel)
|
||||
}
|
||||
|
||||
err := Retry(context.Background(), operation, backoff.NewConstantBackOff(time.Millisecond), nil)
|
||||
assert.ErrorIs(t, err, sentinel, "permanent errors must stop retries")
|
||||
}
|
||||
|
||||
func TestRetryNilNetState(t *testing.T) {
|
||||
attempts := 0
|
||||
operation := func() error {
|
||||
attempts++
|
||||
if attempts < 3 {
|
||||
return errors.New("transient")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
err := Retry(context.Background(), operation, backoff.NewConstantBackOff(time.Millisecond), nil)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 3, attempts, "nil network state must preserve timed retries")
|
||||
}
|
||||
|
||||
func TestRetryStops(t *testing.T) {
|
||||
failure := errors.New("still failing")
|
||||
operation := func() error {
|
||||
return failure
|
||||
}
|
||||
|
||||
err := Retry(context.Background(), operation, &backoff.StopBackOff{}, nil)
|
||||
assert.ErrorIs(t, err, failure, "stop backoff must return the operation error")
|
||||
}
|
||||
|
||||
func TestRetryCtxCancelDuringSleep(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
operation := func() error {
|
||||
return errors.New("failing")
|
||||
}
|
||||
|
||||
go func() {
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
cancel()
|
||||
}()
|
||||
|
||||
start := time.Now()
|
||||
err := Retry(ctx, operation, backoff.NewConstantBackOff(time.Minute), netstate.New())
|
||||
|
||||
assert.ErrorIs(t, err, context.Canceled, "context cancellation must stop the retry loop")
|
||||
assert.Less(t, time.Since(start), time.Second, "context cancellation must interrupt backoff sleep")
|
||||
}
|
||||
@@ -4,7 +4,7 @@ import (
|
||||
"net"
|
||||
"testing"
|
||||
|
||||
"go.uber.org/mock/gomock"
|
||||
"github.com/golang/mock/gomock"
|
||||
"github.com/google/gopacket"
|
||||
"github.com/google/gopacket/layers"
|
||||
|
||||
|
||||
@@ -8,7 +8,7 @@ import (
|
||||
"net/netip"
|
||||
reflect "reflect"
|
||||
|
||||
gomock "go.uber.org/mock/gomock"
|
||||
gomock "github.com/golang/mock/gomock"
|
||||
)
|
||||
|
||||
// MockPacketFilter is a mock of PacketFilter interface.
|
||||
|
||||
@@ -8,7 +8,7 @@ import (
|
||||
os "os"
|
||||
reflect "reflect"
|
||||
|
||||
gomock "go.uber.org/mock/gomock"
|
||||
gomock "github.com/golang/mock/gomock"
|
||||
tun "golang.zx2c4.com/wireguard/tun"
|
||||
)
|
||||
|
||||
|
||||
@@ -53,15 +53,15 @@ func NewProxyBind(bind Bind, mtu uint16) *ProxyBind {
|
||||
return p
|
||||
}
|
||||
|
||||
// AddRelayedConn adds a new connection to the bind.
|
||||
// AddTurnConn adds a new connection to the bind.
|
||||
// endpoint is the NetBird address of the remote peer. The SetEndpoint return with the address what will be used in the
|
||||
// WireGuard configuration.
|
||||
//
|
||||
// Parameters:
|
||||
// - ctx: Context is used for proxyToLocal to avoid unnecessary error messages
|
||||
// - nbAddr: The NetBird UDP address of the remote peer, it required to generate fake address
|
||||
// - remoteConn: The established relayed connection to the remote peer
|
||||
func (p *ProxyBind) AddRelayedConn(ctx context.Context, nbAddr *net.UDPAddr, remoteConn net.Conn) error {
|
||||
// - remoteConn: The established TURN connection to the remote peer
|
||||
func (p *ProxyBind) AddTurnConn(ctx context.Context, nbAddr *net.UDPAddr, remoteConn net.Conn) error {
|
||||
fakeNetIP, err := fakeAddress(nbAddr)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@@ -30,9 +30,9 @@ type WGEBPFProxy struct {
|
||||
proxyPort int
|
||||
mtu uint16
|
||||
|
||||
ebpfManager ebpfMgr.Manager
|
||||
relayedConnStore map[uint16]net.Conn
|
||||
relayedConnMutex sync.Mutex
|
||||
ebpfManager ebpfMgr.Manager
|
||||
turnConnStore map[uint16]net.Conn
|
||||
turnConnMutex sync.Mutex
|
||||
|
||||
lastUsedPort uint16
|
||||
rawConnIPv4 net.PacketConn
|
||||
@@ -50,7 +50,7 @@ func NewWGEBPFProxy(wgPort int, mtu uint16) *WGEBPFProxy {
|
||||
localWGListenPort: wgPort,
|
||||
mtu: mtu,
|
||||
ebpfManager: ebpf.GetEbpfManagerInstance(),
|
||||
relayedConnStore: make(map[uint16]net.Conn),
|
||||
turnConnStore: make(map[uint16]net.Conn),
|
||||
}
|
||||
return wgProxy
|
||||
}
|
||||
@@ -110,14 +110,14 @@ func (p *WGEBPFProxy) Listen() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// AddRelayedConn add new relayed connection for the proxy
|
||||
func (p *WGEBPFProxy) AddRelayedConn(relayedConn net.Conn) (*net.UDPAddr, error) {
|
||||
wgEndpointPort, err := p.storeRelayedConn(relayedConn)
|
||||
// AddTurnConn add new turn connection for the proxy
|
||||
func (p *WGEBPFProxy) AddTurnConn(turnConn net.Conn) (*net.UDPAddr, error) {
|
||||
wgEndpointPort, err := p.storeTurnConn(turnConn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
log.Infof("relayed conn added to wg proxy store: %s, endpoint port: :%d", relayedConn.RemoteAddr(), wgEndpointPort)
|
||||
log.Infof("turn conn added to wg proxy store: %s, endpoint port: :%d", turnConn.RemoteAddr(), wgEndpointPort)
|
||||
|
||||
wgEndpoint := &net.UDPAddr{
|
||||
IP: net.ParseIP(loopbackAddr),
|
||||
@@ -186,48 +186,48 @@ func (p *WGEBPFProxy) readAndForwardPacket(buf []byte) error {
|
||||
return fmt.Errorf("failed to read UDP packet from WG: %w", err)
|
||||
}
|
||||
|
||||
p.relayedConnMutex.Lock()
|
||||
conn, ok := p.relayedConnStore[uint16(addr.Port)]
|
||||
p.relayedConnMutex.Unlock()
|
||||
p.turnConnMutex.Lock()
|
||||
conn, ok := p.turnConnStore[uint16(addr.Port)]
|
||||
p.turnConnMutex.Unlock()
|
||||
if !ok {
|
||||
if p.ctx.Err() == nil {
|
||||
log.Debugf("relayed conn not found by port because conn already has been closed: %d", addr.Port)
|
||||
log.Debugf("turn conn not found by port because conn already has been closed: %d", addr.Port)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
if _, err := conn.Write(buf[:n]); err != nil {
|
||||
return fmt.Errorf("forward local WG packet (%d) to remote relayed conn: %w", addr.Port, err)
|
||||
return fmt.Errorf("failed to forward local WG packet (%d) to remote turn conn: %w", addr.Port, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *WGEBPFProxy) storeRelayedConn(relayedConn net.Conn) (uint16, error) {
|
||||
p.relayedConnMutex.Lock()
|
||||
defer p.relayedConnMutex.Unlock()
|
||||
func (p *WGEBPFProxy) storeTurnConn(turnConn net.Conn) (uint16, error) {
|
||||
p.turnConnMutex.Lock()
|
||||
defer p.turnConnMutex.Unlock()
|
||||
|
||||
np, err := p.nextFreePort()
|
||||
if err != nil {
|
||||
return np, err
|
||||
}
|
||||
p.relayedConnStore[np] = relayedConn
|
||||
p.turnConnStore[np] = turnConn
|
||||
return np, nil
|
||||
}
|
||||
|
||||
func (p *WGEBPFProxy) removeRelayedConn(relayedConnID uint16) {
|
||||
p.relayedConnMutex.Lock()
|
||||
defer p.relayedConnMutex.Unlock()
|
||||
func (p *WGEBPFProxy) removeTurnConn(turnConnID uint16) {
|
||||
p.turnConnMutex.Lock()
|
||||
defer p.turnConnMutex.Unlock()
|
||||
|
||||
_, ok := p.relayedConnStore[relayedConnID]
|
||||
_, ok := p.turnConnStore[turnConnID]
|
||||
if ok {
|
||||
log.Debugf("remove relayed conn from store by port: %d", relayedConnID)
|
||||
log.Debugf("remove turn conn from store by port: %d", turnConnID)
|
||||
}
|
||||
delete(p.relayedConnStore, relayedConnID)
|
||||
delete(p.turnConnStore, turnConnID)
|
||||
}
|
||||
|
||||
func (p *WGEBPFProxy) nextFreePort() (uint16, error) {
|
||||
if len(p.relayedConnStore) == 65535 {
|
||||
return 0, fmt.Errorf("reached maximum relayed connection numbers")
|
||||
if len(p.turnConnStore) == 65535 {
|
||||
return 0, fmt.Errorf("reached maximum turn connection numbers")
|
||||
}
|
||||
generatePort:
|
||||
if p.lastUsedPort == 65535 {
|
||||
@@ -236,7 +236,7 @@ generatePort:
|
||||
p.lastUsedPort++
|
||||
}
|
||||
|
||||
if _, ok := p.relayedConnStore[p.lastUsedPort]; ok {
|
||||
if _, ok := p.turnConnStore[p.lastUsedPort]; ok {
|
||||
goto generatePort
|
||||
}
|
||||
return p.lastUsedPort, nil
|
||||
|
||||
@@ -9,32 +9,32 @@ import (
|
||||
func TestWGEBPFProxy_connStore(t *testing.T) {
|
||||
wgProxy := NewWGEBPFProxy(1, 1280)
|
||||
|
||||
p, _ := wgProxy.storeRelayedConn(nil)
|
||||
p, _ := wgProxy.storeTurnConn(nil)
|
||||
if p != 1 {
|
||||
t.Errorf("invalid initial port: %d", wgProxy.lastUsedPort)
|
||||
}
|
||||
|
||||
numOfConns := 10
|
||||
for i := 0; i < numOfConns; i++ {
|
||||
p, _ = wgProxy.storeRelayedConn(nil)
|
||||
p, _ = wgProxy.storeTurnConn(nil)
|
||||
}
|
||||
if p != uint16(numOfConns)+1 {
|
||||
t.Errorf("invalid last used port: %d, expected: %d", p, numOfConns+1)
|
||||
}
|
||||
if len(wgProxy.relayedConnStore) != numOfConns+1 {
|
||||
t.Errorf("invalid store size: %d, expected: %d", len(wgProxy.relayedConnStore), numOfConns+1)
|
||||
if len(wgProxy.turnConnStore) != numOfConns+1 {
|
||||
t.Errorf("invalid store size: %d, expected: %d", len(wgProxy.turnConnStore), numOfConns+1)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWGEBPFProxy_portCalculation_overflow(t *testing.T) {
|
||||
wgProxy := NewWGEBPFProxy(1, 1280)
|
||||
|
||||
_, _ = wgProxy.storeRelayedConn(nil)
|
||||
_, _ = wgProxy.storeTurnConn(nil)
|
||||
wgProxy.lastUsedPort = 65535
|
||||
p, _ := wgProxy.storeRelayedConn(nil)
|
||||
p, _ := wgProxy.storeTurnConn(nil)
|
||||
|
||||
if len(wgProxy.relayedConnStore) != 2 {
|
||||
t.Errorf("invalid store size: %d, expected: %d", len(wgProxy.relayedConnStore), 2)
|
||||
if len(wgProxy.turnConnStore) != 2 {
|
||||
t.Errorf("invalid store size: %d, expected: %d", len(wgProxy.turnConnStore), 2)
|
||||
}
|
||||
|
||||
if p != 2 {
|
||||
@@ -46,11 +46,11 @@ func TestWGEBPFProxy_portCalculation_maxConn(t *testing.T) {
|
||||
wgProxy := NewWGEBPFProxy(1, 1280)
|
||||
|
||||
for i := 0; i < 65535; i++ {
|
||||
_, _ = wgProxy.storeRelayedConn(nil)
|
||||
_, _ = wgProxy.storeTurnConn(nil)
|
||||
}
|
||||
|
||||
_, err := wgProxy.storeRelayedConn(nil)
|
||||
_, err := wgProxy.storeTurnConn(nil)
|
||||
if err == nil {
|
||||
t.Errorf("invalid relayed conn store calculation")
|
||||
t.Errorf("invalid turn conn store calculation")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -121,10 +121,10 @@ func NewProxyWrapper(proxy *WGEBPFProxy) *ProxyWrapper {
|
||||
}
|
||||
}
|
||||
|
||||
func (p *ProxyWrapper) AddRelayedConn(ctx context.Context, _ *net.UDPAddr, remoteConn net.Conn) error {
|
||||
addr, err := p.wgeBPFProxy.AddRelayedConn(remoteConn)
|
||||
func (p *ProxyWrapper) AddTurnConn(ctx context.Context, _ *net.UDPAddr, remoteConn net.Conn) error {
|
||||
addr, err := p.wgeBPFProxy.AddTurnConn(remoteConn)
|
||||
if err != nil {
|
||||
return fmt.Errorf("add relayed conn: %w", err)
|
||||
return fmt.Errorf("add turn conn: %w", err)
|
||||
}
|
||||
|
||||
headers, err := NewPacketHeaders(p.wgeBPFProxy.localWGListenPort, addr)
|
||||
@@ -252,7 +252,7 @@ func (p *ProxyWrapper) CloseConn() error {
|
||||
}
|
||||
|
||||
func (p *ProxyWrapper) proxyToLocal(ctx context.Context) {
|
||||
defer p.wgeBPFProxy.removeRelayedConn(uint16(p.wgRelayedEndpointAddr.Port))
|
||||
defer p.wgeBPFProxy.removeTurnConn(uint16(p.wgRelayedEndpointAddr.Port))
|
||||
|
||||
buf := make([]byte, p.wgeBPFProxy.mtu+bufsize.WGBufferOverhead)
|
||||
for {
|
||||
@@ -273,7 +273,7 @@ func (p *ProxyWrapper) proxyToLocal(ctx context.Context) {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
log.Errorf("failed to write out relayed pkg to local conn: %v", err)
|
||||
log.Errorf("failed to write out turn pkg to local conn: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -286,7 +286,7 @@ func (p *ProxyWrapper) readFromRemote(ctx context.Context, buf []byte) (int, err
|
||||
}
|
||||
p.closeListener.Notify()
|
||||
if !errors.Is(err, io.EOF) {
|
||||
log.Errorf("failed to read from relayed conn (endpoint: :%d): %s", p.wgRelayedEndpointAddr.Port, err)
|
||||
log.Errorf("failed to read from turn conn (endpoint: :%d): %s", p.wgRelayedEndpointAddr.Port, err)
|
||||
}
|
||||
return 0, err
|
||||
}
|
||||
|
||||
@@ -7,7 +7,7 @@ import (
|
||||
|
||||
// Proxy is a transfer layer between the relayed connection and the WireGuard
|
||||
type Proxy interface {
|
||||
AddRelayedConn(ctx context.Context, endpoint *net.UDPAddr, remoteConn net.Conn) error
|
||||
AddTurnConn(ctx context.Context, endpoint *net.UDPAddr, remoteConn net.Conn) error
|
||||
EndpointAddr() *net.UDPAddr // EndpointAddr returns the address of the WireGuard peer endpoint
|
||||
Work() // Work start or resume the proxy
|
||||
Pause() // Pause to forward the packages from remote connection to WireGuard. The opposite way still works.
|
||||
|
||||
@@ -95,7 +95,7 @@ func TestProxyCloseByRemoteConn(t *testing.T) {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
addr, _ := net.ResolveUDPAddr("udp", "100.108.135.221:51892")
|
||||
relayedConn := newMockConn()
|
||||
err := tt.proxy.AddRelayedConn(ctx, addr, relayedConn)
|
||||
err := tt.proxy.AddTurnConn(ctx, addr, relayedConn)
|
||||
if err != nil {
|
||||
t.Errorf("error: %v", err)
|
||||
}
|
||||
@@ -157,7 +157,7 @@ func redirectTraffic(t *testing.T, proxy Proxy, wgPort int, endPointAddr *net.UD
|
||||
_ = relayedServer.Close()
|
||||
}()
|
||||
|
||||
if err := proxy.AddRelayedConn(context.Background(), endPointAddr, relayedConn); err != nil {
|
||||
if err := proxy.AddTurnConn(context.Background(), endPointAddr, relayedConn); err != nil {
|
||||
t.Errorf("error: %v", err)
|
||||
}
|
||||
defer func() {
|
||||
|
||||
@@ -119,9 +119,9 @@ func testRedirectAs(t *testing.T, proxy Proxy, wgPort int, nbAddr, p2pEndpoint *
|
||||
}
|
||||
defer relayConn.Close()
|
||||
|
||||
// Add relayed connection to proxy
|
||||
if err := proxy.AddRelayedConn(ctx, nbAddr, relayConn); err != nil {
|
||||
t.Fatalf("failed to add relayed connection: %v", err)
|
||||
// Add TURN connection to proxy
|
||||
if err := proxy.AddTurnConn(ctx, nbAddr, relayConn); err != nil {
|
||||
t.Fatalf("failed to add TURN connection: %v", err)
|
||||
}
|
||||
defer func() {
|
||||
if err := proxy.CloseConn(); err != nil {
|
||||
@@ -304,8 +304,8 @@ func TestRedirectAs_Multiple_Switches(t *testing.T) {
|
||||
Port: 38746,
|
||||
}
|
||||
|
||||
if err := proxy.AddRelayedConn(ctx, nbAddr, relayConn); err != nil {
|
||||
t.Fatalf("failed to add relayed connection: %v", err)
|
||||
if err := proxy.AddTurnConn(ctx, nbAddr, relayConn); err != nil {
|
||||
t.Fatalf("failed to add TURN connection: %v", err)
|
||||
}
|
||||
defer func() {
|
||||
if err := proxy.CloseConn(); err != nil {
|
||||
|
||||
@@ -51,12 +51,12 @@ func NewWGUDPProxy(wgPort int, mtu uint16) *WGUDPProxy {
|
||||
return p
|
||||
}
|
||||
|
||||
// AddRelayedConn dials the local WireGuard port and stores the relayed connection.
|
||||
// AddTurnConn
|
||||
// The provided Context must be non-nil. If the context expires before
|
||||
// the connection is complete, an error is returned. Once successfully
|
||||
// connected, any expiration of the context will not affect the
|
||||
// connection.
|
||||
func (p *WGUDPProxy) AddRelayedConn(ctx context.Context, _ *net.UDPAddr, remoteConn net.Conn) error {
|
||||
func (p *WGUDPProxy) AddTurnConn(ctx context.Context, _ *net.UDPAddr, remoteConn net.Conn) error {
|
||||
dialer := net.Dialer{}
|
||||
localConn, err := dialer.DialContext(ctx, "udp", fmt.Sprintf(":%d", p.localWGListenPort))
|
||||
if err != nil {
|
||||
|
||||
@@ -116,11 +116,11 @@ func (d *DefaultManager) ApplyFiltering(networkMap *mgmProto.NetworkMap, dnsRout
|
||||
// firewall state, so an identical hash means an identical resulting ruleset.
|
||||
func (d *DefaultManager) firewallConfigHash(networkMap *mgmProto.NetworkMap, dnsRouteFeatureFlag bool) (uint64, error) {
|
||||
return hashstructure.Hash(struct {
|
||||
PeerRules []*mgmProto.FirewallRule
|
||||
PeerRulesIsEmpty bool
|
||||
RouteRules []*mgmProto.RouteFirewallRule
|
||||
RouteRulesIsEmpty bool
|
||||
DNSRouteFeatureFlag bool
|
||||
PeerRules []*mgmProto.FirewallRule
|
||||
PeerRulesIsEmpty bool
|
||||
RouteRules []*mgmProto.RouteFirewallRule
|
||||
RouteRulesIsEmpty bool
|
||||
DNSRouteFeatureFlag bool
|
||||
}{
|
||||
PeerRules: networkMap.GetFirewallRules(),
|
||||
PeerRulesIsEmpty: networkMap.GetFirewallRulesIsEmpty(),
|
||||
@@ -144,13 +144,13 @@ func (d *DefaultManager) applyPeerACLs(networkMap *mgmProto.NetworkMap) {
|
||||
log.Warn("this peer is connected to a NetBird Management service with an older version. Allowing all traffic from connected peers")
|
||||
rules = append(rules,
|
||||
&mgmProto.FirewallRule{
|
||||
PeerIP: "0.0.0.0", //nolint:staticcheck
|
||||
PeerIP: "0.0.0.0",
|
||||
Direction: mgmProto.RuleDirection_IN,
|
||||
Action: mgmProto.RuleAction_ACCEPT,
|
||||
Protocol: mgmProto.RuleProtocol_ALL,
|
||||
},
|
||||
&mgmProto.FirewallRule{
|
||||
PeerIP: "0.0.0.0", //nolint:staticcheck
|
||||
PeerIP: "0.0.0.0",
|
||||
Direction: mgmProto.RuleDirection_OUT,
|
||||
Action: mgmProto.RuleAction_ACCEPT,
|
||||
Protocol: mgmProto.RuleProtocol_ALL,
|
||||
@@ -407,6 +407,7 @@ func (d *DefaultManager) getRuleGroupingSelector(rule *mgmProto.FirewallRule) st
|
||||
return fmt.Sprintf("%v:%v:%v:%s:%v", strconv.Itoa(int(rule.Direction)), rule.Action, rule.Protocol, rule.Port, rule.PortInfo)
|
||||
}
|
||||
|
||||
|
||||
// extractRuleIP extracts the peer IP from a firewall rule.
|
||||
// If sourcePrefixes is populated (new management), decode the first entry and use its address.
|
||||
// Otherwise fall back to the deprecated PeerIP string field (old management).
|
||||
|
||||
@@ -5,9 +5,9 @@ import (
|
||||
"net/netip"
|
||||
"testing"
|
||||
|
||||
"github.com/golang/mock/gomock"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/mock/gomock"
|
||||
|
||||
"github.com/netbirdio/netbird/client/firewall"
|
||||
"github.com/netbirdio/netbird/client/iface"
|
||||
@@ -87,7 +87,7 @@ func TestDefaultManager(t *testing.T) {
|
||||
networkMap.FirewallRules = append(
|
||||
networkMap.FirewallRules,
|
||||
&mgmProto.FirewallRule{
|
||||
PeerIP: "10.93.0.3", //nolint:staticcheck
|
||||
PeerIP: "10.93.0.3",
|
||||
Direction: mgmProto.RuleDirection_IN,
|
||||
Action: mgmProto.RuleAction_DROP,
|
||||
Protocol: mgmProto.RuleProtocol_ICMP,
|
||||
@@ -556,12 +556,12 @@ func TestApplyFilteringSkipsUnchangedConfig(t *testing.T) {
|
||||
|
||||
func buildNetworkMap(peerRules, routeRules int) *mgmProto.NetworkMap {
|
||||
nm := &mgmProto.NetworkMap{
|
||||
FirewallRulesIsEmpty: peerRules == 0,
|
||||
FirewallRulesIsEmpty: peerRules == 0,
|
||||
RoutesFirewallRulesIsEmpty: routeRules == 0,
|
||||
}
|
||||
for i := range peerRules {
|
||||
nm.FirewallRules = append(nm.FirewallRules, &mgmProto.FirewallRule{
|
||||
PeerIP: fmt.Sprintf("10.%d.%d.%d", i>>16&0xff, i>>8&0xff, i&0xff), //nolint:staticcheck
|
||||
PeerIP: fmt.Sprintf("10.%d.%d.%d", i>>16&0xff, i>>8&0xff, i&0xff),
|
||||
Direction: mgmProto.RuleDirection_IN,
|
||||
Action: mgmProto.RuleAction_ACCEPT,
|
||||
Protocol: mgmProto.RuleProtocol_TCP,
|
||||
|
||||
@@ -7,7 +7,7 @@ package mocks
|
||||
import (
|
||||
reflect "reflect"
|
||||
|
||||
gomock "go.uber.org/mock/gomock"
|
||||
gomock "github.com/golang/mock/gomock"
|
||||
wgdevice "golang.zx2c4.com/wireguard/device"
|
||||
|
||||
"github.com/netbirdio/netbird/client/iface/device"
|
||||
|
||||
@@ -138,37 +138,26 @@ func (a *Auth) IsSSOSupported(ctx context.Context) (bool, error) {
|
||||
|
||||
// GetOAuthFlow returns an OAuth flow (PKCE or Device) using the existing management connection
|
||||
// This avoids creating a new connection to the management server
|
||||
func (a *Auth) GetOAuthFlow(ctx context.Context, forceDeviceAuth bool, hint string) (OAuthFlow, error) {
|
||||
func (a *Auth) GetOAuthFlow(ctx context.Context, forceDeviceAuth bool) (OAuthFlow, error) {
|
||||
var flow OAuthFlow
|
||||
var err error
|
||||
|
||||
err := a.withRetry(ctx, func(client *mgm.GrpcClient) error {
|
||||
err = a.withRetry(ctx, func(client *mgm.GrpcClient) error {
|
||||
if forceDeviceAuth {
|
||||
deviceFlow, err := a.getDeviceFlow(client)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
deviceFlow.SetLoginHint(hint)
|
||||
flow = deviceFlow
|
||||
return nil
|
||||
flow, err = a.getDeviceFlow(client)
|
||||
return err
|
||||
}
|
||||
|
||||
// Try PKCE flow first
|
||||
pkceFlow, err := a.getPKCEFlow(client)
|
||||
flow, err = a.getPKCEFlow(client)
|
||||
if err != nil {
|
||||
// If PKCE not supported, try Device flow
|
||||
if s, ok := status.FromError(err); ok && (s.Code() == codes.NotFound || s.Code() == codes.Unimplemented) {
|
||||
deviceFlow, err := a.getDeviceFlow(client)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
deviceFlow.SetLoginHint(hint)
|
||||
flow = deviceFlow
|
||||
return nil
|
||||
flow, err = a.getDeviceFlow(client)
|
||||
return err
|
||||
}
|
||||
return err
|
||||
}
|
||||
pkceFlow.SetLoginHint(hint)
|
||||
flow = pkceFlow
|
||||
return nil
|
||||
})
|
||||
|
||||
|
||||
@@ -97,7 +97,9 @@ func authenticateWithPKCEFlow(ctx context.Context, config *profilemanager.Config
|
||||
return nil, fmt.Errorf("getting pkce authorization flow info failed with error: %v", err)
|
||||
}
|
||||
|
||||
pkceFlowInfo.SetLoginHint(hint)
|
||||
if hint != "" {
|
||||
pkceFlowInfo.SetLoginHint(hint)
|
||||
}
|
||||
|
||||
return pkceFlowInfo, nil
|
||||
}
|
||||
@@ -125,7 +127,9 @@ func authenticateWithDeviceCodeFlow(ctx context.Context, config *profilemanager.
|
||||
}
|
||||
}
|
||||
|
||||
deviceFlowInfo.SetLoginHint(hint)
|
||||
if hint != "" {
|
||||
deviceFlowInfo.SetLoginHint(hint)
|
||||
}
|
||||
|
||||
return deviceFlowInfo, nil
|
||||
}
|
||||
|
||||
@@ -38,8 +38,6 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/updater"
|
||||
"github.com/netbirdio/netbird/client/internal/updater/installer"
|
||||
nbnet "github.com/netbirdio/netbird/client/net"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
cProto "github.com/netbirdio/netbird/client/proto"
|
||||
"github.com/netbirdio/netbird/client/ssh"
|
||||
sshconfig "github.com/netbirdio/netbird/client/ssh/config"
|
||||
@@ -72,42 +70,18 @@ type ConnectClient struct {
|
||||
updateManager *updater.Manager
|
||||
|
||||
persistSyncResponse bool
|
||||
|
||||
// netState gates every reconnection loop on OS-reported network
|
||||
// availability. Nil (the default) disables gating; mobile platforms
|
||||
// inject it via WithNetworkState.
|
||||
netState *netstate.State
|
||||
|
||||
// sweeper cuts the management, signal and relay connections on network
|
||||
// change; nil disables it.
|
||||
sweeper *netsweep.Sweeper
|
||||
}
|
||||
|
||||
// ConnectClientOption configures optional ConnectClient behavior.
|
||||
type ConnectClientOption func(*ConnectClient)
|
||||
|
||||
// WithNetworkState injects the OS network availability state that gates every
|
||||
// reconnection loop; without it gating is disabled.
|
||||
func WithNetworkState(netState *netstate.State) ConnectClientOption {
|
||||
return func(c *ConnectClient) { c.netState = netState }
|
||||
}
|
||||
|
||||
// WithSweeper injects the network change sweeper.
|
||||
func WithSweeper(sweeper *netsweep.Sweeper) ConnectClientOption {
|
||||
return func(c *ConnectClient) { c.sweeper = sweeper }
|
||||
}
|
||||
|
||||
func NewConnectClient(
|
||||
ctx context.Context,
|
||||
config *profilemanager.Config,
|
||||
statusRecorder *peer.Status,
|
||||
opts ...ConnectClientOption,
|
||||
) *ConnectClient {
|
||||
// Derive the run context here so Stop owns the cancel that unblocks the run
|
||||
// loop. runCancel is set once at construction, so Stop can call it without
|
||||
// racing the run loop's startup. Callers therefore need not cancel before Stop.
|
||||
runCtx, runCancel := context.WithCancel(ctx)
|
||||
c := &ConnectClient{
|
||||
return &ConnectClient{
|
||||
ctx: runCtx,
|
||||
runCancel: runCancel,
|
||||
runExited: make(chan struct{}),
|
||||
@@ -115,10 +89,6 @@ func NewConnectClient(
|
||||
statusRecorder: statusRecorder,
|
||||
engineMutex: sync.Mutex{},
|
||||
}
|
||||
for _, opt := range opts {
|
||||
opt(c)
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
func (c *ConnectClient) SetUpdateManager(um *updater.Manager) {
|
||||
@@ -304,13 +274,6 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
return nil
|
||||
}
|
||||
|
||||
// suspend connection attempts while the OS reports no usable network
|
||||
if waited, err := c.netState.Wait(c.ctx); err != nil {
|
||||
return nil
|
||||
} else if waited {
|
||||
backOff.Reset()
|
||||
}
|
||||
|
||||
state.Set(StatusConnecting)
|
||||
|
||||
engineCtx, cancel := context.WithCancel(c.ctx)
|
||||
@@ -322,8 +285,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
}()
|
||||
|
||||
log.Debugf("connecting to the Management service %s", c.config.ManagementURL.Host)
|
||||
mgmClient, err := mgm.NewClient(engineCtx, c.config.ManagementURL.Host, myPrivateKey, mgmTlsEnabled,
|
||||
mgm.WithNetworkState(c.netState), mgm.WithSweeper(c.sweeper))
|
||||
mgmClient, err := mgm.NewClient(engineCtx, c.config.ManagementURL.Host, myPrivateKey, mgmTlsEnabled)
|
||||
if err != nil {
|
||||
// On daemon shutdown / Down() the parent context is cancelled
|
||||
// and the dial fails with "context canceled". Wrapping that
|
||||
@@ -398,7 +360,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
}()
|
||||
|
||||
// with the global Netbird config in hand connect (just a connection, no stream yet) Signal
|
||||
signalClient, err := connectToSignal(engineCtx, loginResp.GetNetbirdConfig(), myPrivateKey, c.netState, c.sweeper)
|
||||
signalClient, err := connectToSignal(engineCtx, loginResp.GetNetbirdConfig(), myPrivateKey)
|
||||
if err != nil {
|
||||
log.Error(err)
|
||||
return wrapErr(err)
|
||||
@@ -434,8 +396,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
engineConfig.StateDir = filepath.Dir(path)
|
||||
}
|
||||
|
||||
relayManager := relayClient.NewManager(engineCtx, relayURLs, myPrivateKey.PublicKey().String(), engineConfig.MTU,
|
||||
relayClient.WithNetworkState(c.netState), relayClient.WithSweeper(c.sweeper))
|
||||
relayManager := relayClient.NewManager(engineCtx, relayURLs, myPrivateKey.PublicKey().String(), engineConfig.MTU)
|
||||
c.statusRecorder.SetRelayMgr(relayManager)
|
||||
if len(relayURLs) > 0 {
|
||||
if token != nil {
|
||||
@@ -463,7 +424,6 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
UpdateManager: c.updateManager,
|
||||
ClientMetrics: c.clientMetrics,
|
||||
MetricsCtx: c.ctx,
|
||||
NetState: c.netState,
|
||||
}, mobileDependency)
|
||||
engine.SetSyncResponsePersistence(c.persistSyncResponse)
|
||||
c.engine = engine
|
||||
@@ -520,16 +480,6 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
// status stream stuck at Connecting.
|
||||
err = backoff.Retry(operation, backoff.WithContext(backOff, c.ctx))
|
||||
if err != nil {
|
||||
// Once the client context is cancelled backoff.WithContext surfaces the
|
||||
// bare context error, and any attempt torn down mid-flight reports the
|
||||
// same. That cancellation is the caller asking us to stop (Stop, Down or
|
||||
// an engine restart), so exit cleanly instead of handing back a failure
|
||||
// the caller would have to distinguish from a real one.
|
||||
if c.ctx.Err() != nil && errors.Is(err, context.Canceled) {
|
||||
log.Info("exiting client retry loop, context cancelled")
|
||||
return nil
|
||||
}
|
||||
|
||||
log.Debugf("exiting client retry loop due to unrecoverable error: %s", err)
|
||||
if s, ok := gstatus.FromError(err); ok && (s.Code() == codes.PermissionDenied) {
|
||||
state.Set(StatusNeedsLogin)
|
||||
@@ -723,7 +673,7 @@ func selectMTU(localMTU uint16, peerMTU int32) uint16 {
|
||||
}
|
||||
|
||||
// connectToSignal creates Signal Service client and established a connection
|
||||
func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourPrivateKey wgtypes.Key, netState *netstate.State, sweeper *netsweep.Sweeper) (*signal.GrpcClient, error) {
|
||||
func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourPrivateKey wgtypes.Key) (*signal.GrpcClient, error) {
|
||||
var sigTLSEnabled bool
|
||||
if wtConfig.Signal.Protocol == mgmProto.HostConfig_HTTPS {
|
||||
sigTLSEnabled = true
|
||||
@@ -731,8 +681,7 @@ func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourP
|
||||
sigTLSEnabled = false
|
||||
}
|
||||
|
||||
signalClient, err := signal.NewClient(ctx, wtConfig.Signal.Uri, ourPrivateKey, sigTLSEnabled,
|
||||
signal.WithNetworkState(netState), signal.WithSweeper(sweeper))
|
||||
signalClient, err := signal.NewClient(ctx, wtConfig.Signal.Uri, ourPrivateKey, sigTLSEnabled)
|
||||
if err != nil {
|
||||
log.Errorf("error while connecting to the Signal Exchange Service %s: %s", wtConfig.Signal.Uri, err)
|
||||
return nil, gstatus.Errorf(codes.FailedPrecondition, "failed connecting to Signal Service : %s", err)
|
||||
|
||||
@@ -459,7 +459,7 @@ func (r *registryConfigurator) flushDNSCache() {
|
||||
|
||||
ret, _, err := dnsFlushResolverCacheFn.Call()
|
||||
if ret == 0 {
|
||||
if !errors.Is(err, syscall.Errno(0)) {
|
||||
if err != nil && !errors.Is(err, syscall.Errno(0)) {
|
||||
log.Errorf("DnsFlushResolverCache failed: %v", err)
|
||||
return
|
||||
}
|
||||
@@ -627,7 +627,7 @@ func refreshGroupPolicy() error {
|
||||
)
|
||||
|
||||
if ret == 0 {
|
||||
if !errors.Is(err, syscall.Errno(0)) {
|
||||
if err != nil && !errors.Is(err, syscall.Errno(0)) {
|
||||
return fmt.Errorf("RefreshPolicyEx failed: %w", err)
|
||||
}
|
||||
return fmt.Errorf("RefreshPolicyEx failed")
|
||||
|
||||
@@ -4,7 +4,7 @@ import (
|
||||
"net"
|
||||
"testing"
|
||||
|
||||
"go.uber.org/mock/gomock"
|
||||
"github.com/golang/mock/gomock"
|
||||
"github.com/google/gopacket"
|
||||
"github.com/google/gopacket/layers"
|
||||
"github.com/miekg/dns"
|
||||
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"go.uber.org/mock/gomock"
|
||||
"github.com/golang/mock/gomock"
|
||||
"github.com/miekg/dns"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
|
||||
|
||||
@@ -101,7 +101,7 @@ func (m *Manager) Start(fwdEntries []*ForwarderEntry) error {
|
||||
m.dnsForwarder = NewDNSForwarder(listenAddress, dnsTTL, m.firewall, m.statusRecorder, m.wgIface)
|
||||
|
||||
go func() {
|
||||
if err := m.dnsForwarder.Listen(fwdEntries); err != nil { //nolint:staticcheck
|
||||
if err := m.dnsForwarder.Listen(fwdEntries); err != nil {
|
||||
// todo handle close error if it is exists
|
||||
log.Errorf("failed to start DNS forwarder, err: %v", err)
|
||||
}
|
||||
|
||||
@@ -59,7 +59,6 @@ import (
|
||||
"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/netstate"
|
||||
cProto "github.com/netbirdio/netbird/client/proto"
|
||||
"github.com/netbirdio/netbird/client/system"
|
||||
nbdns "github.com/netbirdio/netbird/dns"
|
||||
@@ -182,9 +181,6 @@ type EngineServices struct {
|
||||
UpdateManager *updater.Manager
|
||||
ClientMetrics *metrics.ClientMetrics
|
||||
MetricsCtx context.Context
|
||||
// NetState gates the reconnection loops on OS-reported network
|
||||
// availability; nil disables gating.
|
||||
NetState *netstate.State
|
||||
}
|
||||
|
||||
// Engine is a mechanism responsible for reacting on Signal and Management stream events and managing connections to the remote peers.
|
||||
@@ -208,10 +204,6 @@ type Engine struct {
|
||||
config *EngineConfig
|
||||
mobileDep MobileDependency
|
||||
|
||||
// netState gates the peer reconnection guards on OS-reported network
|
||||
// availability; nil disables gating.
|
||||
netState *netstate.State
|
||||
|
||||
// STUNs is a list of STUN servers used by ICE
|
||||
STUNs []*stun.URI
|
||||
// TURNs is a list of STUN servers used by ICE
|
||||
@@ -345,7 +337,6 @@ func NewEngine(
|
||||
syncMsgMux: &sync.Mutex{},
|
||||
config: config,
|
||||
mobileDep: mobileDep,
|
||||
netState: services.NetState,
|
||||
STUNs: []*stun.URI{},
|
||||
TURNs: []*stun.URI{},
|
||||
networkSerial: 0,
|
||||
@@ -1902,8 +1893,7 @@ func (e *Engine) createPeerConn(pubKey string, allowedIPs []netip.Prefix, agentV
|
||||
Addr: e.getRosenpassAddr(),
|
||||
PermissiveMode: e.config.RosenpassPermissive,
|
||||
},
|
||||
ICEConfig: e.createICEConfig(),
|
||||
NetworkState: e.netState,
|
||||
ICEConfig: e.createICEConfig(),
|
||||
}
|
||||
|
||||
serviceDependencies := peer.ServiceDependencies{
|
||||
@@ -2572,7 +2562,7 @@ func (e *Engine) SetCapture(pc device.PacketCapture) error {
|
||||
}
|
||||
|
||||
afc := capture.NewAFPacketCapture(intf.Name(), sess)
|
||||
if err := afc.Start(); err != nil { //nolint:staticcheck // always errors on non-Linux builds
|
||||
if err := afc.Start(); err != nil {
|
||||
return fmt.Errorf("start AF_PACKET capture on %s: %w", intf.Name(), err)
|
||||
}
|
||||
e.afpacketCapture = afc
|
||||
|
||||
@@ -12,7 +12,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go.uber.org/mock/gomock"
|
||||
"github.com/golang/mock/gomock"
|
||||
"github.com/google/uuid"
|
||||
log "github.com/sirupsen/logrus"
|
||||
"github.com/stretchr/testify/assert"
|
||||
|
||||
@@ -26,7 +26,6 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/portforward"
|
||||
"github.com/netbirdio/netbird/client/internal/rosenpass"
|
||||
"github.com/netbirdio/netbird/client/internal/stdnet"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/route"
|
||||
relayClient "github.com/netbirdio/netbird/shared/relay/client"
|
||||
)
|
||||
@@ -94,10 +93,6 @@ type ConnConfig struct {
|
||||
|
||||
// ICEConfig ICE protocol configuration
|
||||
ICEConfig icemaker.Config
|
||||
|
||||
// NetworkState gates the reconnection guard on OS-reported network
|
||||
// availability; nil disables gating.
|
||||
NetworkState *netstate.State
|
||||
}
|
||||
|
||||
type Conn struct {
|
||||
@@ -259,7 +254,7 @@ func (conn *Conn) open(engineCtx context.Context, firstPacket []byte) error {
|
||||
conn.handshaker.AddICEListener(conn.workerICE.OnNewOffer)
|
||||
}
|
||||
|
||||
conn.guard = guard.NewGuard(conn.Log, conn.isConnectedOnAllWay, conn.config.Timeout, conn.srWatcher, conn.config.NetworkState)
|
||||
conn.guard = guard.NewGuard(conn.Log, conn.isConnectedOnAllWay, conn.config.Timeout, conn.srWatcher)
|
||||
|
||||
conn.wg.Add(1)
|
||||
go func() {
|
||||
@@ -445,7 +440,7 @@ func (conn *Conn) onICEConnectionIsReady(priority conntype.ConnPriority, iceConn
|
||||
conn.dumpState.NewLocalProxy()
|
||||
wgProxy, err = conn.newProxy(iceConnInfo.RemoteConn)
|
||||
if err != nil {
|
||||
conn.Log.Errorf("failed to add relayed net.Conn to local proxy: %v", err)
|
||||
conn.Log.Errorf("failed to add turn net.Conn to local proxy: %v", err)
|
||||
return
|
||||
}
|
||||
ep = wgProxy.EndpointAddr()
|
||||
@@ -883,8 +878,9 @@ func (conn *Conn) newProxy(remoteConn net.Conn) (wgproxy.Proxy, error) {
|
||||
}
|
||||
|
||||
wgProxy := conn.config.WgConfig.WgInterface.GetProxy()
|
||||
if err := wgProxy.AddRelayedConn(conn.ctx, udpAddr, remoteConn); err != nil {
|
||||
return nil, fmt.Errorf("add relayed conn to proxy: %w", err)
|
||||
if err := wgProxy.AddTurnConn(conn.ctx, udpAddr, remoteConn); err != nil {
|
||||
conn.Log.Errorf("failed to add turn net.Conn to local proxy: %v", err)
|
||||
return nil, err
|
||||
}
|
||||
return wgProxy, nil
|
||||
}
|
||||
|
||||
@@ -6,8 +6,6 @@ import (
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
// ConnStatus represents the connection state as seen by the guard.
|
||||
@@ -33,26 +31,20 @@ type connStatusFunc func() ConnStatus
|
||||
// - Relayed connection disconnected
|
||||
// - ICE candidate changes
|
||||
type Guard struct {
|
||||
log *log.Entry
|
||||
isConnectedOnAllWay connStatusFunc
|
||||
timeout time.Duration
|
||||
srWatcher *SRWatcher
|
||||
// netState gates reconnect attempts on OS-reported network availability;
|
||||
// nil disables gating.
|
||||
netState *netstate.State
|
||||
log *log.Entry
|
||||
isConnectedOnAllWay connStatusFunc
|
||||
timeout time.Duration
|
||||
srWatcher *SRWatcher
|
||||
relayedConnDisconnected chan struct{}
|
||||
iCEConnDisconnected chan struct{}
|
||||
}
|
||||
|
||||
// NewGuard creates a reconnection guard for a peer connection. A nil netState
|
||||
// disables network availability gating.
|
||||
func NewGuard(log *log.Entry, isConnectedFn connStatusFunc, timeout time.Duration, srWatcher *SRWatcher, netState *netstate.State) *Guard {
|
||||
func NewGuard(log *log.Entry, isConnectedFn connStatusFunc, timeout time.Duration, srWatcher *SRWatcher) *Guard {
|
||||
return &Guard{
|
||||
log: log,
|
||||
isConnectedOnAllWay: isConnectedFn,
|
||||
timeout: timeout,
|
||||
srWatcher: srWatcher,
|
||||
netState: netState,
|
||||
relayedConnDisconnected: make(chan struct{}, 1),
|
||||
iCEConnDisconnected: make(chan struct{}, 1),
|
||||
}
|
||||
@@ -104,16 +96,9 @@ func (g *Guard) reconnectLoopWithRetry(ctx context.Context, callback func()) {
|
||||
iceState := &iceRetryState{log: g.log}
|
||||
defer iceState.reset()
|
||||
|
||||
netChanged := g.netState.Changed()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-tickerChannel:
|
||||
// skip attempts while the OS reports no usable network; the
|
||||
// netChanged case below resumes the loop once it returns
|
||||
if !g.netState.IsOnline() {
|
||||
continue
|
||||
}
|
||||
switch g.isConnectedOnAllWay() {
|
||||
case ConnStatusConnected:
|
||||
// all good, nothing to do
|
||||
@@ -150,23 +135,6 @@ func (g *Guard) reconnectLoopWithRetry(ctx context.Context, callback func()) {
|
||||
tickerChannel = ticker.C
|
||||
iceState.reset()
|
||||
|
||||
case <-netChanged:
|
||||
// Re-arm for the next transition before acting on this one.
|
||||
netChanged = g.netState.Changed()
|
||||
if !g.netState.IsOnline() {
|
||||
continue
|
||||
}
|
||||
// Ticks skipped while offline drove the backoff towards its
|
||||
// maximum without ever attempting, and left the ICE budget
|
||||
// frozen — possibly in hourly mode. Recover on our own so the
|
||||
// peer does not depend on a signal or relay event that never
|
||||
// comes when both stayed up across the outage.
|
||||
g.log.Debugf("network is back, reset reconnection ticker")
|
||||
ticker.Stop()
|
||||
ticker = g.newReconnectTicker(ctx)
|
||||
tickerChannel = ticker.C
|
||||
iceState.reset()
|
||||
|
||||
case <-ctx.Done():
|
||||
g.log.Debugf("context is done, stop reconnect loop")
|
||||
return
|
||||
|
||||
@@ -15,7 +15,7 @@ import (
|
||||
|
||||
func newTestGuard(status connStatusFunc) *Guard {
|
||||
srw := NewSRWatcher(nil, nil, nil, ice.Config{})
|
||||
return NewGuard(log.WithField("test", "guard"), status, 50*time.Millisecond, srw, nil)
|
||||
return NewGuard(log.WithField("test", "guard"), status, 50*time.Millisecond, srw)
|
||||
}
|
||||
|
||||
// countBackoffTickerGoroutines returns how many goroutines are currently sitting
|
||||
|
||||
@@ -1,107 +0,0 @@
|
||||
package guard
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/internal/peer/ice"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
// newTestGuardWithNetState builds a guard with a realistic MaxInterval: the
|
||||
// backoff must be able to grow well past the outage, as it does in production
|
||||
// where the timeout is seconds to minutes.
|
||||
func newTestGuardWithNetState(status connStatusFunc, netState *netstate.State) *Guard {
|
||||
srw := NewSRWatcher(nil, nil, nil, ice.Config{})
|
||||
return NewGuard(log.WithField("test", "guard"), status, 30*time.Second, srw, netState)
|
||||
}
|
||||
|
||||
// TestGuard_RecoversAfterOfflineToOnline covers a peer that stays disconnected
|
||||
// across a network outage while neither signal nor relay reports an event —
|
||||
// both stayed up, as on a short airplane mode toggle over Wi-Fi.
|
||||
//
|
||||
// Every tick taken while offline is skipped, but it still advances the
|
||||
// exponential backoff, so by the time the network returns the next tick can be
|
||||
// tens of seconds away. Without an explicit reaction to the transition the
|
||||
// peer waits out that interval for a recovery that could start immediately.
|
||||
func TestGuard_RecoversAfterOfflineToOnline(t *testing.T) {
|
||||
netState := netstate.New()
|
||||
|
||||
var attempts atomic.Int32
|
||||
g := newTestGuardWithNetState(func() ConnStatus { return ConnStatusDisconnected }, netState)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
// Start from the reconnect ticker (800ms initial interval), the state a
|
||||
// peer is in after it loses its connection.
|
||||
go g.Start(ctx, func() { attempts.Add(1) })
|
||||
g.SetRelayedConnDisconnected()
|
||||
|
||||
// Let the backoff climb: 0.8s, 1.6s, 3.2s, 6.4s ... every tick is skipped
|
||||
// while offline, but each one doubles the wait for the next.
|
||||
netState.Set(false)
|
||||
time.Sleep(8 * time.Second)
|
||||
|
||||
offlineAttempts := attempts.Load()
|
||||
if offlineAttempts != 0 {
|
||||
t.Fatalf("callback ran %d times while offline, want 0", offlineAttempts)
|
||||
}
|
||||
|
||||
netState.Set(true)
|
||||
|
||||
// The next organic tick is now several seconds out, so anything within
|
||||
// this window can only come from reacting to the transition itself.
|
||||
pollCtx, stopPolling := context.WithTimeout(ctx, 2*time.Second)
|
||||
defer stopPolling()
|
||||
|
||||
select {
|
||||
case <-pollCtx.Done():
|
||||
t.Fatal("peer was not retried within 2s of the network coming back, " +
|
||||
"with neither a signal nor a relay event to fall back on")
|
||||
case <-pollUntil(pollCtx, func() bool { return attempts.Load() > 0 }):
|
||||
}
|
||||
}
|
||||
|
||||
// TestGuard_OfflineTransitionDoesNotRetry checks the other direction: going
|
||||
// offline must not itself trigger an attempt.
|
||||
func TestGuard_OfflineTransitionDoesNotRetry(t *testing.T) {
|
||||
netState := netstate.New()
|
||||
|
||||
var attempts atomic.Int32
|
||||
g := newTestGuardWithNetState(func() ConnStatus { return ConnStatusDisconnected }, netState)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
go g.Start(ctx, func() { attempts.Add(1) })
|
||||
|
||||
netState.Set(false)
|
||||
time.Sleep(5 * time.Second)
|
||||
|
||||
if got := attempts.Load(); got != 0 {
|
||||
t.Fatalf("callback ran %d times after going offline, want 0", got)
|
||||
}
|
||||
}
|
||||
|
||||
// pollUntil closes the returned channel once cond holds. It gives up when ctx
|
||||
// is done, so the polling goroutine never outlives the test that started it.
|
||||
func pollUntil(ctx context.Context, cond func() bool) <-chan struct{} {
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
for {
|
||||
if cond() {
|
||||
close(done)
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(10 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
}()
|
||||
return done
|
||||
}
|
||||
@@ -1,40 +1,11 @@
|
||||
package peer
|
||||
|
||||
// ClientState identifies the client connection state delivered via
|
||||
// Listener.OnStateChanged.
|
||||
type ClientState int
|
||||
|
||||
// Client states. The numeric values cross the gomobile boundary (the mobile
|
||||
// bindings re-export them as integer constants), so they are a wire format:
|
||||
// append new states at the end, never reorder or insert.
|
||||
const (
|
||||
ClientStateDisconnected ClientState = iota
|
||||
ClientStateConnected
|
||||
ClientStateConnecting
|
||||
ClientStateDisconnecting
|
||||
// ClientStateNoNetwork is an overlay state: it is never stored as the
|
||||
// last notification, only derived from ClientStateConnecting while the
|
||||
// OS reports no usable network (see notifier.effectiveState).
|
||||
ClientStateNoNetwork
|
||||
)
|
||||
|
||||
// Listener is a callback type about the NetBird network connection state
|
||||
type Listener interface {
|
||||
// OnStateChanged reports every client state transition. New states are
|
||||
// delivered only through this callback; the per-state callbacks below
|
||||
// are kept for compatibility and will be removed once all consumers
|
||||
// have migrated.
|
||||
OnStateChanged(state ClientState)
|
||||
|
||||
// Deprecated: consume OnStateChanged instead.
|
||||
OnConnected()
|
||||
// Deprecated: consume OnStateChanged instead.
|
||||
OnDisconnected()
|
||||
// Deprecated: consume OnStateChanged instead.
|
||||
OnConnecting()
|
||||
// Deprecated: consume OnStateChanged instead.
|
||||
OnDisconnecting()
|
||||
|
||||
OnAddressChanged(string, string)
|
||||
OnPeersListChanged(int)
|
||||
}
|
||||
|
||||
@@ -4,64 +4,31 @@ import (
|
||||
"sync"
|
||||
)
|
||||
|
||||
const (
|
||||
stateDisconnected = iota
|
||||
stateConnected
|
||||
stateConnecting
|
||||
stateDisconnecting
|
||||
)
|
||||
|
||||
type notifier struct {
|
||||
// publishLock orders state publication: it is held across computing the
|
||||
// effective state and handing it to the listener, so a transition cannot
|
||||
// overtake a newer one and leave the listener on a stale state.
|
||||
publishLock sync.Mutex
|
||||
serverStateLock sync.Mutex
|
||||
listenersLock sync.Mutex
|
||||
listener Listener
|
||||
currentClientState bool
|
||||
lastNotification ClientState
|
||||
lastNotification int
|
||||
lastNumberOfPeers int
|
||||
lastFqdnAddress string
|
||||
lastIPAddress string
|
||||
networkAvailable bool
|
||||
}
|
||||
|
||||
func newNotifier() *notifier {
|
||||
return ¬ifier{
|
||||
networkAvailable: true,
|
||||
}
|
||||
}
|
||||
|
||||
// effectiveState maps the computed state to what listeners should see:
|
||||
// while the OS reports no usable network, "Connecting" would be a lie —
|
||||
// connection attempts are suspended — so it is reported as NoNetwork.
|
||||
// Caller must hold serverStateLock.
|
||||
func (n *notifier) effectiveState(state ClientState) ClientState {
|
||||
if !n.networkAvailable && state == ClientStateConnecting {
|
||||
return ClientStateNoNetwork
|
||||
}
|
||||
return state
|
||||
}
|
||||
|
||||
// setNetworkAvailable records the OS network availability and re-notifies
|
||||
// the listener when the flag flips the effective state (Connecting <->
|
||||
// NoNetwork).
|
||||
func (n *notifier) setNetworkAvailable(available bool) {
|
||||
n.publishLock.Lock()
|
||||
defer n.publishLock.Unlock()
|
||||
|
||||
n.serverStateLock.Lock()
|
||||
if n.networkAvailable == available {
|
||||
n.serverStateLock.Unlock()
|
||||
return
|
||||
}
|
||||
previous := n.effectiveState(n.lastNotification)
|
||||
n.networkAvailable = available
|
||||
current := n.effectiveState(n.lastNotification)
|
||||
n.serverStateLock.Unlock()
|
||||
|
||||
if previous != current {
|
||||
n.notify(current)
|
||||
}
|
||||
return ¬ifier{}
|
||||
}
|
||||
|
||||
func (n *notifier) setListener(listener Listener) {
|
||||
n.serverStateLock.Lock()
|
||||
lastNotification := n.effectiveState(n.lastNotification)
|
||||
lastNotification := n.lastNotification
|
||||
numOfPeers := n.lastNumberOfPeers
|
||||
fqdnAddress := n.lastFqdnAddress
|
||||
address := n.lastIPAddress
|
||||
@@ -85,9 +52,6 @@ func (n *notifier) removeListener() {
|
||||
}
|
||||
|
||||
func (n *notifier) updateServerStates(mgmState bool, signalState bool) {
|
||||
n.publishLock.Lock()
|
||||
defer n.publishLock.Unlock()
|
||||
|
||||
n.serverStateLock.Lock()
|
||||
calculatedState := n.calculateState(mgmState, signalState)
|
||||
|
||||
@@ -97,54 +61,43 @@ func (n *notifier) updateServerStates(mgmState bool, signalState bool) {
|
||||
}
|
||||
|
||||
n.lastNotification = calculatedState
|
||||
effective := n.effectiveState(calculatedState)
|
||||
n.serverStateLock.Unlock()
|
||||
|
||||
n.notify(effective)
|
||||
n.notify(calculatedState)
|
||||
}
|
||||
|
||||
func (n *notifier) clientStart() {
|
||||
n.publishLock.Lock()
|
||||
defer n.publishLock.Unlock()
|
||||
|
||||
n.serverStateLock.Lock()
|
||||
n.currentClientState = true
|
||||
n.lastNotification = ClientStateConnecting
|
||||
effective := n.effectiveState(ClientStateConnecting)
|
||||
n.lastNotification = stateConnecting
|
||||
n.serverStateLock.Unlock()
|
||||
|
||||
n.notify(effective)
|
||||
n.notify(stateConnecting)
|
||||
}
|
||||
|
||||
func (n *notifier) clientStop() {
|
||||
n.publishLock.Lock()
|
||||
defer n.publishLock.Unlock()
|
||||
|
||||
n.serverStateLock.Lock()
|
||||
n.currentClientState = false
|
||||
n.lastNotification = ClientStateDisconnected
|
||||
n.lastNotification = stateDisconnected
|
||||
n.serverStateLock.Unlock()
|
||||
|
||||
n.notify(ClientStateDisconnected)
|
||||
n.notify(stateDisconnected)
|
||||
}
|
||||
|
||||
func (n *notifier) clientTearDown() {
|
||||
n.publishLock.Lock()
|
||||
defer n.publishLock.Unlock()
|
||||
|
||||
n.serverStateLock.Lock()
|
||||
n.currentClientState = false
|
||||
n.lastNotification = ClientStateDisconnecting
|
||||
n.lastNotification = stateDisconnecting
|
||||
n.serverStateLock.Unlock()
|
||||
|
||||
n.notify(ClientStateDisconnecting)
|
||||
n.notify(stateDisconnecting)
|
||||
}
|
||||
|
||||
func (n *notifier) isServerStateChanged(newState ClientState) bool {
|
||||
func (n *notifier) isServerStateChanged(newState int) bool {
|
||||
return n.lastNotification != newState
|
||||
}
|
||||
|
||||
func (n *notifier) notify(state ClientState) {
|
||||
func (n *notifier) notify(state int) {
|
||||
n.listenersLock.Lock()
|
||||
listener := n.listener
|
||||
n.listenersLock.Unlock()
|
||||
@@ -156,20 +109,20 @@ func (n *notifier) notify(state ClientState) {
|
||||
notifyListener(listener, state)
|
||||
}
|
||||
|
||||
func (n *notifier) calculateState(managementConn, signalConn bool) ClientState {
|
||||
func (n *notifier) calculateState(managementConn, signalConn bool) int {
|
||||
if managementConn && signalConn {
|
||||
return ClientStateConnected
|
||||
return stateConnected
|
||||
}
|
||||
|
||||
if !managementConn && !signalConn && !n.currentClientState {
|
||||
return ClientStateDisconnected
|
||||
return stateDisconnected
|
||||
}
|
||||
|
||||
if n.lastNotification == ClientStateDisconnecting {
|
||||
return ClientStateDisconnecting
|
||||
if n.lastNotification == stateDisconnecting {
|
||||
return stateDisconnecting
|
||||
}
|
||||
|
||||
return ClientStateConnecting
|
||||
return stateConnecting
|
||||
}
|
||||
|
||||
func (n *notifier) peerListChanged(numOfPeers int) {
|
||||
@@ -206,19 +159,15 @@ func (n *notifier) localAddressChanged(fqdn, address string) {
|
||||
listener.OnAddressChanged(fqdn, address)
|
||||
}
|
||||
|
||||
func notifyListener(l Listener, state ClientState) {
|
||||
// legacy per-state callbacks; NoNetwork is delivered only via
|
||||
// OnStateChanged below
|
||||
func notifyListener(l Listener, state int) {
|
||||
switch state {
|
||||
case ClientStateDisconnected:
|
||||
case stateDisconnected:
|
||||
l.OnDisconnected()
|
||||
case ClientStateConnected:
|
||||
case stateConnected:
|
||||
l.OnConnected()
|
||||
case ClientStateConnecting:
|
||||
case stateConnecting:
|
||||
l.OnConnecting()
|
||||
case ClientStateDisconnecting:
|
||||
case stateDisconnecting:
|
||||
l.OnDisconnecting()
|
||||
}
|
||||
|
||||
l.OnStateChanged(state)
|
||||
}
|
||||
|
||||
@@ -1,108 +0,0 @@
|
||||
package peer
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type recordingListener struct {
|
||||
mu sync.Mutex
|
||||
states []ClientState
|
||||
onState func(ClientState)
|
||||
}
|
||||
|
||||
func (l *recordingListener) OnStateChanged(state ClientState) {
|
||||
l.mu.Lock()
|
||||
l.states = append(l.states, state)
|
||||
hook := l.onState
|
||||
l.mu.Unlock()
|
||||
|
||||
if hook != nil {
|
||||
hook(state)
|
||||
}
|
||||
}
|
||||
|
||||
func (l *recordingListener) last() (ClientState, bool) {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
if len(l.states) == 0 {
|
||||
return 0, false
|
||||
}
|
||||
return l.states[len(l.states)-1], true
|
||||
}
|
||||
|
||||
func (l *recordingListener) snapshot() []ClientState {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
return append([]ClientState(nil), l.states...)
|
||||
}
|
||||
|
||||
func (l *recordingListener) OnConnected() {}
|
||||
func (l *recordingListener) OnDisconnected() {}
|
||||
func (l *recordingListener) OnConnecting() {}
|
||||
func (l *recordingListener) OnDisconnecting() {}
|
||||
func (l *recordingListener) OnAddressChanged(string, string) {}
|
||||
func (l *recordingListener) OnPeersListChanged(int) {}
|
||||
|
||||
// TestNotifier_ConcurrentAvailabilityFlipOrdersPublication holds the first
|
||||
// transition inside the listener callback and flips availability again from
|
||||
// another goroutine while it is parked. The second flip must not publish
|
||||
// ahead of the one in flight, otherwise the listener ends up on a state the
|
||||
// notifier already superseded.
|
||||
func TestNotifier_ConcurrentAvailabilityFlipOrdersPublication(t *testing.T) {
|
||||
n := newNotifier()
|
||||
n.currentClientState = true
|
||||
n.lastNotification = ClientStateConnecting
|
||||
|
||||
entered := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
|
||||
l := &recordingListener{}
|
||||
l.onState = func(state ClientState) {
|
||||
if state != ClientStateNoNetwork {
|
||||
return
|
||||
}
|
||||
l.mu.Lock()
|
||||
l.onState = nil
|
||||
l.mu.Unlock()
|
||||
close(entered)
|
||||
<-release
|
||||
}
|
||||
n.listener = l
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
n.setNetworkAvailable(false)
|
||||
}()
|
||||
|
||||
<-entered
|
||||
|
||||
flipped := make(chan struct{})
|
||||
go func() {
|
||||
defer close(flipped)
|
||||
n.setNetworkAvailable(true)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-flipped:
|
||||
t.Fatal("the online transition published while the offline one was " +
|
||||
"still in flight; publication is not serialized")
|
||||
case <-time.After(200 * time.Millisecond):
|
||||
}
|
||||
|
||||
close(release)
|
||||
<-flipped
|
||||
wg.Wait()
|
||||
|
||||
got, ok := l.last()
|
||||
if !ok {
|
||||
t.Fatal("listener never observed a state")
|
||||
}
|
||||
if got != ClientStateConnecting {
|
||||
t.Fatalf("listener holds %v after the network came back, want Connecting; sequence: %v",
|
||||
got, l.snapshot())
|
||||
}
|
||||
}
|
||||
@@ -6,32 +6,29 @@ import (
|
||||
)
|
||||
|
||||
type mocListener struct {
|
||||
lastState ClientState
|
||||
lastState int
|
||||
wg sync.WaitGroup
|
||||
peersWg sync.WaitGroup
|
||||
peers int
|
||||
}
|
||||
|
||||
func (l *mocListener) OnConnected() {
|
||||
l.lastState = ClientStateConnected
|
||||
l.lastState = stateConnected
|
||||
l.wg.Done()
|
||||
}
|
||||
func (l *mocListener) OnDisconnected() {
|
||||
l.lastState = ClientStateDisconnected
|
||||
l.lastState = stateDisconnected
|
||||
l.wg.Done()
|
||||
}
|
||||
func (l *mocListener) OnConnecting() {
|
||||
l.lastState = ClientStateConnecting
|
||||
l.lastState = stateConnecting
|
||||
l.wg.Done()
|
||||
}
|
||||
func (l *mocListener) OnDisconnecting() {
|
||||
l.lastState = ClientStateDisconnecting
|
||||
l.lastState = stateDisconnecting
|
||||
l.wg.Done()
|
||||
}
|
||||
|
||||
func (l *mocListener) OnStateChanged(state ClientState) {
|
||||
|
||||
}
|
||||
func (l *mocListener) OnAddressChanged(host, addr string) {
|
||||
|
||||
}
|
||||
@@ -60,15 +57,15 @@ func Test_notifier_serverState(t *testing.T) {
|
||||
|
||||
type scenario struct {
|
||||
name string
|
||||
expected ClientState
|
||||
expected int
|
||||
mgmState bool
|
||||
signalState bool
|
||||
}
|
||||
scenarios := []scenario{
|
||||
{"connected", ClientStateConnected, true, true},
|
||||
{"mgm down", ClientStateConnecting, false, true},
|
||||
{"signal down", ClientStateConnecting, true, false},
|
||||
{"disconnected", ClientStateDisconnected, false, false},
|
||||
{"connected", stateConnected, true, true},
|
||||
{"mgm down", stateConnecting, false, true},
|
||||
{"signal down", stateConnecting, true, false},
|
||||
{"disconnected", stateDisconnected, false, false},
|
||||
}
|
||||
|
||||
for _, tt := range scenarios {
|
||||
@@ -88,7 +85,7 @@ func Test_notifier_SetListener(t *testing.T) {
|
||||
listener.setPeersWaiter()
|
||||
|
||||
n := newNotifier()
|
||||
n.lastNotification = ClientStateConnecting
|
||||
n.lastNotification = stateConnecting
|
||||
n.setListener(listener)
|
||||
listener.wait()
|
||||
listener.waitPeers()
|
||||
@@ -102,7 +99,7 @@ func Test_notifier_RemoveListener(t *testing.T) {
|
||||
listener.setWaiter()
|
||||
listener.setPeersWaiter()
|
||||
n := newNotifier()
|
||||
n.lastNotification = ClientStateConnecting
|
||||
n.lastNotification = stateConnecting
|
||||
n.setListener(listener)
|
||||
// setListener replays cached state on a goroutine; wait for both the state
|
||||
// and peers callbacks to finish so we don't race on listener.peers.
|
||||
|
||||
@@ -1211,12 +1211,6 @@ func (d *Status) ClientTeardown() {
|
||||
d.notifyStateChange()
|
||||
}
|
||||
|
||||
// SetNetworkAvailable records the OS-reported network availability; while
|
||||
// unavailable, listeners see NoNetwork instead of Connecting.
|
||||
func (d *Status) SetNetworkAvailable(available bool) {
|
||||
d.notifier.setNetworkAvailable(available)
|
||||
}
|
||||
|
||||
// SetConnectionListener set a listener to the notifier
|
||||
func (d *Status) SetConnectionListener(listener Listener) {
|
||||
d.notifier.setListener(listener)
|
||||
|
||||
@@ -255,8 +255,8 @@ func (w *WorkerICE) connect(ctx context.Context, agent *icemaker.ThreadSafeAgent
|
||||
return
|
||||
}
|
||||
|
||||
w.log.Debugf("agent dial")
|
||||
remoteConn, err := w.agentDial(ctx, agent, remoteOfferAnswer)
|
||||
w.log.Debugf("turn agent dial")
|
||||
remoteConn, err := w.turnAgentDial(ctx, agent, remoteOfferAnswer)
|
||||
if err != nil {
|
||||
w.log.Debugf("failed to dial the remote peer: %s", err)
|
||||
w.closeAgent(agent, w.agentDialerCancel)
|
||||
@@ -389,17 +389,6 @@ func (w *WorkerICE) injectPortForwardedCandidate(srflxCandidate ice.Candidate) {
|
||||
return
|
||||
}
|
||||
|
||||
// A forwarded candidate only makes sense for an IPv4 mapping, which
|
||||
// translates a port on the gateway's address. An IPv6 pinhole translates
|
||||
// nothing: it unblocks the address ICE already gathers as a host candidate,
|
||||
// so there is no second address to advertise. Injecting one here would also
|
||||
// paste an IPv6 address onto whichever server-reflexive candidate arrived
|
||||
// first, which is usually IPv4.
|
||||
if mapping.ExternalIP != nil && mapping.ExternalIP.To4() == nil {
|
||||
w.log.Debugf("skipping port-forwarded candidate: %s mapping is IPv6-only", mapping.NATType)
|
||||
return
|
||||
}
|
||||
|
||||
w.muxAgent.Lock()
|
||||
if w.portForwardAttempted {
|
||||
w.muxAgent.Unlock()
|
||||
@@ -528,8 +517,8 @@ func (w *WorkerICE) onConnectionStateChange(agent *icemaker.ThreadSafeAgent, dia
|
||||
w.logSuccessfulPaths(agent)
|
||||
return
|
||||
case ice.ConnectionStateFailed, ice.ConnectionStateDisconnected, ice.ConnectionStateClosed:
|
||||
// ice.ConnectionStateClosed happens when we recreate the agent. The P2P to relay switch requires
|
||||
// notifying conn.onICEStateDisconnected so it can update the currently used priority.
|
||||
// ice.ConnectionStateClosed happens when we recreate the agent. For the P2P to TURN switch important to
|
||||
// notify the conn.onICEStateDisconnected changes to update the current used priority
|
||||
|
||||
sessionChanged := w.closeAgent(agent, dialerCancel)
|
||||
|
||||
@@ -543,7 +532,7 @@ func (w *WorkerICE) onConnectionStateChange(agent *icemaker.ThreadSafeAgent, dia
|
||||
}
|
||||
}
|
||||
|
||||
func (w *WorkerICE) agentDial(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *OfferAnswer) (*ice.Conn, error) {
|
||||
func (w *WorkerICE) turnAgentDial(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *OfferAnswer) (*ice.Conn, error) {
|
||||
if isController(w.config) {
|
||||
return agent.Dial(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd)
|
||||
} else {
|
||||
|
||||
@@ -10,8 +10,10 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/netbirdio/go-nat"
|
||||
"github.com/libp2p/go-nat"
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/internal/portforward/pcp"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -166,11 +168,6 @@ func (m *Manager) setup(ctx context.Context) (nat.NAT, *Mapping, error) {
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("create port mapping: %w", err)
|
||||
}
|
||||
|
||||
// Only meaningful once a mapping has been attempted: that is what opens the
|
||||
// pinhole and records its outcome.
|
||||
logIPv6Pinhole(gateway)
|
||||
|
||||
return gateway, mapping, nil
|
||||
}
|
||||
|
||||
@@ -268,9 +265,7 @@ func (m *Manager) checkHealthAndRecreate(ctx context.Context, gateway nat.NAT) b
|
||||
return false
|
||||
}
|
||||
|
||||
// Assert on the interface, not on a concrete type: a dual-stack gateway is
|
||||
// a wrapper around the IPv4 NAT, so a type assertion misses it.
|
||||
checker, ok := gateway.(nat.HealthChecker)
|
||||
pcpNAT, ok := gateway.(*pcp.NAT)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
@@ -278,7 +273,7 @@ func (m *Manager) checkHealthAndRecreate(ctx context.Context, gateway nat.NAT) b
|
||||
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
epoch, serverRestarted, err := checker.CheckServerHealth(ctx)
|
||||
epoch, serverRestarted, err := pcpNAT.CheckServerHealth(ctx)
|
||||
if err != nil {
|
||||
log.Debugf("PCP health check failed: %v", err)
|
||||
return false
|
||||
@@ -345,18 +340,3 @@ func (m *Manager) startTearDown(ctx context.Context) {
|
||||
func isPermanentLeaseRequired(err error) bool {
|
||||
return err != nil && upnpErrPermanentLeaseOnly.MatchString(err.Error())
|
||||
}
|
||||
|
||||
// logIPv6Pinhole reports the outcome of the IPv6 pinhole. Pinholes are best
|
||||
// effort and never fail a mapping on their own, so this is the only way to see
|
||||
// whether one was actually opened.
|
||||
func logIPv6Pinhole(gateway nat.NAT) {
|
||||
reporter, ok := gateway.(nat.IPv6PinholeReporter)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if err := reporter.IPv6PinholeError(); err != nil {
|
||||
log.Warnf("IPv6 pinhole: %v", err)
|
||||
return
|
||||
}
|
||||
log.Infof("IPv6 pinhole open")
|
||||
}
|
||||
|
||||
408
client/internal/portforward/pcp/client.go
Normal file
408
client/internal/portforward/pcp/client.go
Normal file
@@ -0,0 +1,408 @@
|
||||
package pcp
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/netip"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultTimeout = 3 * time.Second
|
||||
responseBufferSize = 128
|
||||
|
||||
// RFC 6887 Section 8.1.1 retry timing
|
||||
initialRetryDelay = 3 * time.Second
|
||||
maxRetryDelay = 1024 * time.Second
|
||||
maxRetries = 4 // 3s + 6s + 12s + 24s = 45s total worst case
|
||||
)
|
||||
|
||||
// Client is a PCP protocol client.
|
||||
// All methods are safe for concurrent use.
|
||||
type Client struct {
|
||||
gateway netip.Addr
|
||||
timeout time.Duration
|
||||
|
||||
mu sync.Mutex
|
||||
// localIP caches the resolved local IP address.
|
||||
localIP netip.Addr
|
||||
// lastEpoch is the last observed server epoch value.
|
||||
lastEpoch uint32
|
||||
// epochTime tracks when lastEpoch was received for state loss detection.
|
||||
epochTime time.Time
|
||||
// externalIP caches the external IP from the last successful MAP response.
|
||||
externalIP netip.Addr
|
||||
// epochStateLost is set when epoch indicates server restart.
|
||||
epochStateLost bool
|
||||
}
|
||||
|
||||
// NewClient creates a new PCP client for the gateway at the given IP.
|
||||
func NewClient(gateway net.IP) *Client {
|
||||
addr, ok := netip.AddrFromSlice(gateway)
|
||||
if !ok {
|
||||
log.Debugf("invalid gateway IP: %v", gateway)
|
||||
}
|
||||
return &Client{
|
||||
gateway: addr.Unmap(),
|
||||
timeout: defaultTimeout,
|
||||
}
|
||||
}
|
||||
|
||||
// NewClientWithTimeout creates a new PCP client with a custom timeout.
|
||||
func NewClientWithTimeout(gateway net.IP, timeout time.Duration) *Client {
|
||||
addr, ok := netip.AddrFromSlice(gateway)
|
||||
if !ok {
|
||||
log.Debugf("invalid gateway IP: %v", gateway)
|
||||
}
|
||||
return &Client{
|
||||
gateway: addr.Unmap(),
|
||||
timeout: timeout,
|
||||
}
|
||||
}
|
||||
|
||||
// SetLocalIP sets the local IP address to use in PCP requests.
|
||||
func (c *Client) SetLocalIP(ip net.IP) {
|
||||
addr, ok := netip.AddrFromSlice(ip)
|
||||
if !ok {
|
||||
log.Debugf("invalid local IP: %v", ip)
|
||||
}
|
||||
c.mu.Lock()
|
||||
c.localIP = addr.Unmap()
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
// Gateway returns the gateway IP address.
|
||||
func (c *Client) Gateway() net.IP {
|
||||
return c.gateway.AsSlice()
|
||||
}
|
||||
|
||||
// Announce sends a PCP ANNOUNCE request to discover PCP support.
|
||||
// Returns the server's epoch time on success.
|
||||
func (c *Client) Announce(ctx context.Context) (epoch uint32, err error) {
|
||||
localIP, err := c.getLocalIP()
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("get local IP: %w", err)
|
||||
}
|
||||
|
||||
req := buildAnnounceRequest(localIP)
|
||||
resp, err := c.sendRequest(ctx, req)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("send announce: %w", err)
|
||||
}
|
||||
|
||||
parsed, err := parseResponse(resp)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("parse announce response: %w", err)
|
||||
}
|
||||
|
||||
if parsed.ResultCode != ResultSuccess {
|
||||
return 0, fmt.Errorf("PCP ANNOUNCE failed: %s", ResultCodeString(parsed.ResultCode))
|
||||
}
|
||||
|
||||
c.mu.Lock()
|
||||
if c.updateEpochLocked(parsed.Epoch) {
|
||||
log.Warnf("PCP server epoch indicates state loss - mappings may need refresh")
|
||||
}
|
||||
c.mu.Unlock()
|
||||
return parsed.Epoch, nil
|
||||
}
|
||||
|
||||
// AddPortMapping requests a port mapping from the PCP server.
|
||||
func (c *Client) AddPortMapping(ctx context.Context, protocol string, internalPort int, lifetime time.Duration) (*MapResponse, error) {
|
||||
return c.addPortMappingWithHint(ctx, protocol, internalPort, internalPort, netip.Addr{}, lifetime)
|
||||
}
|
||||
|
||||
// AddPortMappingWithHint requests a port mapping with suggested external port and IP.
|
||||
// Use lifetime <= 0 to delete a mapping.
|
||||
func (c *Client) AddPortMappingWithHint(ctx context.Context, protocol string, internalPort, suggestedExtPort int, suggestedExtIP net.IP, lifetime time.Duration) (*MapResponse, error) {
|
||||
var extIP netip.Addr
|
||||
if suggestedExtIP != nil {
|
||||
var ok bool
|
||||
extIP, ok = netip.AddrFromSlice(suggestedExtIP)
|
||||
if !ok {
|
||||
log.Debugf("invalid suggested external IP: %v", suggestedExtIP)
|
||||
}
|
||||
extIP = extIP.Unmap()
|
||||
}
|
||||
return c.addPortMappingWithHint(ctx, protocol, internalPort, suggestedExtPort, extIP, lifetime)
|
||||
}
|
||||
|
||||
func (c *Client) addPortMappingWithHint(ctx context.Context, protocol string, internalPort, suggestedExtPort int, suggestedExtIP netip.Addr, lifetime time.Duration) (*MapResponse, error) {
|
||||
localIP, err := c.getLocalIP()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get local IP: %w", err)
|
||||
}
|
||||
|
||||
proto, err := protocolNumber(protocol)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse protocol: %w", err)
|
||||
}
|
||||
|
||||
var nonce [12]byte
|
||||
if _, err := rand.Read(nonce[:]); err != nil {
|
||||
return nil, fmt.Errorf("generate nonce: %w", err)
|
||||
}
|
||||
|
||||
// Convert lifetime to seconds. Lifetime 0 means delete, so only apply
|
||||
// default for positive durations that round to 0 seconds.
|
||||
var lifetimeSec uint32
|
||||
if lifetime > 0 {
|
||||
lifetimeSec = uint32(lifetime.Seconds())
|
||||
if lifetimeSec == 0 {
|
||||
lifetimeSec = DefaultLifetime
|
||||
}
|
||||
}
|
||||
|
||||
req := buildMapRequest(localIP, nonce, proto, uint16(internalPort), uint16(suggestedExtPort), suggestedExtIP, lifetimeSec)
|
||||
|
||||
resp, err := c.sendRequest(ctx, req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("send map request: %w", err)
|
||||
}
|
||||
|
||||
mapResp, err := parseMapResponse(resp)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse map response: %w", err)
|
||||
}
|
||||
|
||||
if mapResp.Nonce != nonce {
|
||||
return nil, fmt.Errorf("nonce mismatch in response")
|
||||
}
|
||||
|
||||
if mapResp.Protocol != proto {
|
||||
return nil, fmt.Errorf("protocol mismatch: requested %d, got %d", proto, mapResp.Protocol)
|
||||
}
|
||||
if mapResp.InternalPort != uint16(internalPort) {
|
||||
return nil, fmt.Errorf("internal port mismatch: requested %d, got %d", internalPort, mapResp.InternalPort)
|
||||
}
|
||||
|
||||
if mapResp.ResultCode != ResultSuccess {
|
||||
return nil, &Error{
|
||||
Code: mapResp.ResultCode,
|
||||
Message: ResultCodeString(mapResp.ResultCode),
|
||||
}
|
||||
}
|
||||
|
||||
c.mu.Lock()
|
||||
if c.updateEpochLocked(mapResp.Epoch) {
|
||||
log.Warnf("PCP server epoch indicates state loss - mappings may need refresh")
|
||||
}
|
||||
c.cacheExternalIPLocked(mapResp.ExternalIP)
|
||||
c.mu.Unlock()
|
||||
return mapResp, nil
|
||||
}
|
||||
|
||||
// DeletePortMapping removes a port mapping by requesting zero lifetime.
|
||||
func (c *Client) DeletePortMapping(ctx context.Context, protocol string, internalPort int) error {
|
||||
if _, err := c.addPortMappingWithHint(ctx, protocol, internalPort, 0, netip.Addr{}, 0); err != nil {
|
||||
var pcpErr *Error
|
||||
if errors.As(err, &pcpErr) && pcpErr.Code == ResultNotAuthorized {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("delete mapping: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetExternalAddress returns the external IP address.
|
||||
// First checks for a cached value from previous MAP responses.
|
||||
// If not cached, creates a short-lived mapping to discover the external IP.
|
||||
func (c *Client) GetExternalAddress(ctx context.Context) (net.IP, error) {
|
||||
c.mu.Lock()
|
||||
if c.externalIP.IsValid() {
|
||||
ip := c.externalIP.AsSlice()
|
||||
c.mu.Unlock()
|
||||
return ip, nil
|
||||
}
|
||||
c.mu.Unlock()
|
||||
|
||||
// Use an ephemeral port in the dynamic range (49152-65535).
|
||||
// Port 0 is not valid with UDP/TCP protocols per RFC 6887.
|
||||
ephemeralPort := 49152 + int(uint16(time.Now().UnixNano()))%(65535-49152)
|
||||
|
||||
// Use minimal lifetime (1 second) for discovery.
|
||||
resp, err := c.AddPortMapping(ctx, "udp", ephemeralPort, time.Second)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create temporary mapping: %w", err)
|
||||
}
|
||||
|
||||
if err := c.DeletePortMapping(ctx, "udp", ephemeralPort); err != nil {
|
||||
log.Debugf("cleanup temporary PCP mapping: %v", err)
|
||||
}
|
||||
|
||||
return resp.ExternalIP.AsSlice(), nil
|
||||
}
|
||||
|
||||
// LastEpoch returns the last observed server epoch value.
|
||||
// A decrease in epoch indicates the server may have restarted and mappings may be lost.
|
||||
func (c *Client) LastEpoch() uint32 {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return c.lastEpoch
|
||||
}
|
||||
|
||||
// EpochStateLost returns true if epoch state loss was detected and clears the flag.
|
||||
func (c *Client) EpochStateLost() bool {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
lost := c.epochStateLost
|
||||
c.epochStateLost = false
|
||||
return lost
|
||||
}
|
||||
|
||||
// updateEpoch updates the epoch tracking and detects potential state loss.
|
||||
// Returns true if state loss was detected (server likely restarted).
|
||||
// Caller must hold c.mu.
|
||||
func (c *Client) updateEpochLocked(newEpoch uint32) bool {
|
||||
now := time.Now()
|
||||
stateLost := false
|
||||
|
||||
// RFC 6887 Section 8.5: Detect invalid epoch indicating server state loss.
|
||||
// client_delta = time since last response
|
||||
// server_delta = epoch change since last response
|
||||
// Invalid if: client_delta+2 < server_delta - server_delta/16
|
||||
// OR: server_delta+2 < client_delta - client_delta/16
|
||||
// The +2 handles quantization, /16 (6.25%) handles clock drift.
|
||||
if !c.epochTime.IsZero() && c.lastEpoch > 0 {
|
||||
clientDelta := uint32(now.Sub(c.epochTime).Seconds())
|
||||
serverDelta := newEpoch - c.lastEpoch
|
||||
|
||||
// Check for epoch going backwards or jumping unexpectedly.
|
||||
// Subtraction is safe: serverDelta/16 is always <= serverDelta.
|
||||
if clientDelta+2 < serverDelta-(serverDelta/16) ||
|
||||
serverDelta+2 < clientDelta-(clientDelta/16) {
|
||||
stateLost = true
|
||||
c.epochStateLost = true
|
||||
}
|
||||
}
|
||||
|
||||
c.lastEpoch = newEpoch
|
||||
c.epochTime = now
|
||||
return stateLost
|
||||
}
|
||||
|
||||
// cacheExternalIP stores the external IP from a successful MAP response.
|
||||
// Caller must hold c.mu.
|
||||
func (c *Client) cacheExternalIPLocked(ip netip.Addr) {
|
||||
if ip.IsValid() && !ip.IsUnspecified() {
|
||||
c.externalIP = ip
|
||||
}
|
||||
}
|
||||
|
||||
// sendRequest sends a PCP request with retries per RFC 6887 Section 8.1.1.
|
||||
func (c *Client) sendRequest(ctx context.Context, req []byte) ([]byte, error) {
|
||||
addr := &net.UDPAddr{IP: c.gateway.AsSlice(), Port: Port}
|
||||
|
||||
var lastErr error
|
||||
delay := initialRetryDelay
|
||||
|
||||
for range maxRetries {
|
||||
resp, err := c.sendOnce(ctx, addr, req)
|
||||
if err == nil {
|
||||
return resp, nil
|
||||
}
|
||||
lastErr = err
|
||||
|
||||
if ctx.Err() != nil {
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
|
||||
// RFC 6887 Section 8.1.1: RT = (1 + RAND) * MIN(2 * RTprev, MRT)
|
||||
// RAND is random between -0.1 and +0.1
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
case <-time.After(retryDelayWithJitter(delay)):
|
||||
}
|
||||
delay = min(delay*2, maxRetryDelay)
|
||||
}
|
||||
|
||||
return nil, fmt.Errorf("PCP request failed after %d retries: %w", maxRetries, lastErr)
|
||||
}
|
||||
|
||||
// retryDelayWithJitter applies RFC 6887 jitter: multiply by (1 + RAND) where RAND is [-0.1, +0.1].
|
||||
func retryDelayWithJitter(d time.Duration) time.Duration {
|
||||
var b [1]byte
|
||||
_, _ = rand.Read(b[:])
|
||||
// Convert byte to range [-0.1, +0.1]: (b/255 * 0.2) - 0.1
|
||||
jitter := (float64(b[0])/255.0)*0.2 - 0.1
|
||||
return time.Duration(float64(d) * (1 + jitter))
|
||||
}
|
||||
|
||||
func (c *Client) sendOnce(ctx context.Context, addr *net.UDPAddr, req []byte) ([]byte, error) {
|
||||
// Use ListenUDP instead of DialUDP to validate response source address per RFC 6887 §8.3.
|
||||
conn, err := net.ListenUDP("udp", nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("listen: %w", err)
|
||||
}
|
||||
defer func() {
|
||||
if err := conn.Close(); err != nil {
|
||||
log.Debugf("close UDP connection: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
timeout := c.timeout
|
||||
if deadline, ok := ctx.Deadline(); ok {
|
||||
if remaining := time.Until(deadline); remaining < timeout {
|
||||
timeout = remaining
|
||||
}
|
||||
}
|
||||
|
||||
if err := conn.SetDeadline(time.Now().Add(timeout)); err != nil {
|
||||
return nil, fmt.Errorf("set deadline: %w", err)
|
||||
}
|
||||
|
||||
if _, err := conn.WriteToUDP(req, addr); err != nil {
|
||||
return nil, fmt.Errorf("write: %w", err)
|
||||
}
|
||||
|
||||
resp := make([]byte, responseBufferSize)
|
||||
n, from, err := conn.ReadFromUDP(resp)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read: %w", err)
|
||||
}
|
||||
|
||||
// RFC 6887 §8.3: Validate response came from expected PCP server.
|
||||
if !from.IP.Equal(addr.IP) {
|
||||
return nil, fmt.Errorf("response from unexpected source %s (expected %s)", from.IP, addr.IP)
|
||||
}
|
||||
|
||||
return resp[:n], nil
|
||||
}
|
||||
|
||||
func (c *Client) getLocalIP() (netip.Addr, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
if !c.localIP.IsValid() {
|
||||
return netip.Addr{}, fmt.Errorf("local IP not set for gateway %s", c.gateway)
|
||||
}
|
||||
return c.localIP, nil
|
||||
}
|
||||
|
||||
func protocolNumber(protocol string) (uint8, error) {
|
||||
switch protocol {
|
||||
case "udp", "UDP":
|
||||
return ProtoUDP, nil
|
||||
case "tcp", "TCP":
|
||||
return ProtoTCP, nil
|
||||
default:
|
||||
return 0, fmt.Errorf("unsupported protocol: %s", protocol)
|
||||
}
|
||||
}
|
||||
|
||||
// Error represents a PCP error response.
|
||||
type Error struct {
|
||||
Code uint8
|
||||
Message string
|
||||
}
|
||||
|
||||
func (e *Error) Error() string {
|
||||
return fmt.Sprintf("PCP error: %s (%d)", e.Message, e.Code)
|
||||
}
|
||||
187
client/internal/portforward/pcp/client_test.go
Normal file
187
client/internal/portforward/pcp/client_test.go
Normal file
@@ -0,0 +1,187 @@
|
||||
package pcp
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"net/netip"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestAddrConversion(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
addr netip.Addr
|
||||
}{
|
||||
{"IPv4", netip.MustParseAddr("192.168.1.100")},
|
||||
{"IPv4 loopback", netip.MustParseAddr("127.0.0.1")},
|
||||
{"IPv6", netip.MustParseAddr("2001:db8::1")},
|
||||
{"IPv6 loopback", netip.MustParseAddr("::1")},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
b16 := addrTo16(tt.addr)
|
||||
|
||||
recovered := addrFrom16(b16)
|
||||
assert.Equal(t, tt.addr, recovered, "address should round-trip")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildAnnounceRequest(t *testing.T) {
|
||||
clientIP := netip.MustParseAddr("192.168.1.100")
|
||||
req := buildAnnounceRequest(clientIP)
|
||||
|
||||
require.Len(t, req, headerSize)
|
||||
assert.Equal(t, byte(Version), req[0], "version")
|
||||
assert.Equal(t, byte(OpAnnounce), req[1], "opcode")
|
||||
|
||||
// Check client IP is properly encoded as IPv4-mapped IPv6
|
||||
assert.Equal(t, byte(0xff), req[18], "IPv4-mapped prefix byte 10")
|
||||
assert.Equal(t, byte(0xff), req[19], "IPv4-mapped prefix byte 11")
|
||||
assert.Equal(t, byte(192), req[20], "IP octet 1")
|
||||
assert.Equal(t, byte(168), req[21], "IP octet 2")
|
||||
assert.Equal(t, byte(1), req[22], "IP octet 3")
|
||||
assert.Equal(t, byte(100), req[23], "IP octet 4")
|
||||
}
|
||||
|
||||
func TestBuildMapRequest(t *testing.T) {
|
||||
clientIP := netip.MustParseAddr("192.168.1.100")
|
||||
nonce := [12]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12}
|
||||
req := buildMapRequest(clientIP, nonce, ProtoUDP, 51820, 51820, netip.Addr{}, 3600)
|
||||
|
||||
require.Len(t, req, mapRequestSize)
|
||||
assert.Equal(t, byte(Version), req[0], "version")
|
||||
assert.Equal(t, byte(OpMap), req[1], "opcode")
|
||||
|
||||
// Lifetime at bytes 4-7
|
||||
assert.Equal(t, uint32(3600), (uint32(req[4])<<24)|(uint32(req[5])<<16)|(uint32(req[6])<<8)|uint32(req[7]), "lifetime")
|
||||
|
||||
// Nonce at bytes 24-35
|
||||
assert.Equal(t, nonce[:], req[24:36], "nonce")
|
||||
|
||||
// Protocol at byte 36
|
||||
assert.Equal(t, byte(ProtoUDP), req[36], "protocol")
|
||||
|
||||
// Internal port at bytes 40-41
|
||||
assert.Equal(t, uint16(51820), (uint16(req[40])<<8)|uint16(req[41]), "internal port")
|
||||
|
||||
// External port at bytes 42-43
|
||||
assert.Equal(t, uint16(51820), (uint16(req[42])<<8)|uint16(req[43]), "external port")
|
||||
}
|
||||
|
||||
func TestParseResponse(t *testing.T) {
|
||||
// Construct a valid ANNOUNCE response
|
||||
resp := make([]byte, headerSize)
|
||||
resp[0] = Version
|
||||
resp[1] = OpAnnounce | OpReply
|
||||
// Result code = 0 (success)
|
||||
// Lifetime = 0
|
||||
// Epoch = 12345
|
||||
resp[8] = 0
|
||||
resp[9] = 0
|
||||
resp[10] = 0x30
|
||||
resp[11] = 0x39
|
||||
|
||||
parsed, err := parseResponse(resp)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, uint8(Version), parsed.Version)
|
||||
assert.Equal(t, uint8(OpAnnounce|OpReply), parsed.Opcode)
|
||||
assert.Equal(t, uint8(ResultSuccess), parsed.ResultCode)
|
||||
assert.Equal(t, uint32(12345), parsed.Epoch)
|
||||
}
|
||||
|
||||
func TestParseResponseErrors(t *testing.T) {
|
||||
t.Run("too short", func(t *testing.T) {
|
||||
_, err := parseResponse([]byte{1, 2, 3})
|
||||
assert.Error(t, err)
|
||||
})
|
||||
|
||||
t.Run("wrong version", func(t *testing.T) {
|
||||
resp := make([]byte, headerSize)
|
||||
resp[0] = 1 // Wrong version
|
||||
resp[1] = OpReply
|
||||
_, err := parseResponse(resp)
|
||||
assert.Error(t, err)
|
||||
})
|
||||
|
||||
t.Run("missing reply bit", func(t *testing.T) {
|
||||
resp := make([]byte, headerSize)
|
||||
resp[0] = Version
|
||||
resp[1] = OpAnnounce // Missing OpReply bit
|
||||
_, err := parseResponse(resp)
|
||||
assert.Error(t, err)
|
||||
})
|
||||
}
|
||||
|
||||
func TestResultCodeString(t *testing.T) {
|
||||
assert.Equal(t, "SUCCESS", ResultCodeString(ResultSuccess))
|
||||
assert.Equal(t, "NOT_AUTHORIZED", ResultCodeString(ResultNotAuthorized))
|
||||
assert.Equal(t, "ADDRESS_MISMATCH", ResultCodeString(ResultAddressMismatch))
|
||||
assert.Contains(t, ResultCodeString(255), "UNKNOWN")
|
||||
}
|
||||
|
||||
func TestProtocolNumber(t *testing.T) {
|
||||
proto, err := protocolNumber("udp")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, uint8(ProtoUDP), proto)
|
||||
|
||||
proto, err = protocolNumber("tcp")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, uint8(ProtoTCP), proto)
|
||||
|
||||
proto, err = protocolNumber("UDP")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, uint8(ProtoUDP), proto)
|
||||
|
||||
_, err = protocolNumber("icmp")
|
||||
assert.Error(t, err)
|
||||
}
|
||||
|
||||
func TestClientCreation(t *testing.T) {
|
||||
gateway := netip.MustParseAddr("192.168.1.1").AsSlice()
|
||||
|
||||
client := NewClient(gateway)
|
||||
assert.Equal(t, net.IP(gateway), client.Gateway())
|
||||
assert.Equal(t, defaultTimeout, client.timeout)
|
||||
|
||||
clientWithTimeout := NewClientWithTimeout(gateway, 5*time.Second)
|
||||
assert.Equal(t, 5*time.Second, clientWithTimeout.timeout)
|
||||
}
|
||||
|
||||
func TestNATType(t *testing.T) {
|
||||
n := NewNAT(netip.MustParseAddr("192.168.1.1").AsSlice(), netip.MustParseAddr("192.168.1.100").AsSlice())
|
||||
assert.Equal(t, "PCP", n.Type())
|
||||
}
|
||||
|
||||
// Integration test - skipped unless PCP_TEST_GATEWAY env is set
|
||||
func TestClientIntegration(t *testing.T) {
|
||||
t.Skip("Integration test - run manually with PCP_TEST_GATEWAY=<gateway-ip>")
|
||||
|
||||
gateway := netip.MustParseAddr("10.0.1.1").AsSlice() // Change to your test gateway
|
||||
localIP := netip.MustParseAddr("10.0.1.100").AsSlice() // Change to your local IP
|
||||
|
||||
client := NewClient(gateway)
|
||||
client.SetLocalIP(localIP)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Test ANNOUNCE
|
||||
epoch, err := client.Announce(ctx)
|
||||
require.NoError(t, err)
|
||||
t.Logf("Server epoch: %d", epoch)
|
||||
|
||||
// Test MAP
|
||||
resp, err := client.AddPortMapping(ctx, "udp", 51820, 1*time.Hour)
|
||||
require.NoError(t, err)
|
||||
t.Logf("Mapping: internal=%d external=%d externalIP=%s",
|
||||
resp.InternalPort, resp.ExternalPort, resp.ExternalIP)
|
||||
|
||||
// Cleanup
|
||||
err = client.DeletePortMapping(ctx, "udp", 51820)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
222
client/internal/portforward/pcp/nat.go
Normal file
222
client/internal/portforward/pcp/nat.go
Normal file
@@ -0,0 +1,222 @@
|
||||
package pcp
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/netip"
|
||||
"runtime"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/libp2p/go-nat"
|
||||
"github.com/libp2p/go-netroute"
|
||||
)
|
||||
|
||||
var _ nat.NAT = (*NAT)(nil)
|
||||
|
||||
// NAT implements the go-nat NAT interface using PCP.
|
||||
// Supports dual-stack (IPv4 and IPv6) when available.
|
||||
// All methods are safe for concurrent use.
|
||||
//
|
||||
// TODO: IPv6 pinholes use the local IPv6 address. If the address changes
|
||||
// (e.g., due to SLAAC rotation or network change), the pinhole becomes stale
|
||||
// and needs to be recreated with the new address.
|
||||
type NAT struct {
|
||||
client *Client
|
||||
|
||||
mu sync.RWMutex
|
||||
// client6 is the IPv6 PCP client, nil if IPv6 is unavailable.
|
||||
client6 *Client
|
||||
// localIP6 caches the local IPv6 address used for PCP requests.
|
||||
localIP6 netip.Addr
|
||||
}
|
||||
|
||||
// NewNAT creates a new NAT instance backed by PCP.
|
||||
func NewNAT(gateway, localIP net.IP) *NAT {
|
||||
client := NewClient(gateway)
|
||||
client.SetLocalIP(localIP)
|
||||
return &NAT{
|
||||
client: client,
|
||||
}
|
||||
}
|
||||
|
||||
// Type returns "PCP" as the NAT type.
|
||||
func (n *NAT) Type() string {
|
||||
return "PCP"
|
||||
}
|
||||
|
||||
// GetDeviceAddress returns the gateway IP address.
|
||||
func (n *NAT) GetDeviceAddress() (net.IP, error) {
|
||||
return n.client.Gateway(), nil
|
||||
}
|
||||
|
||||
// GetExternalAddress returns the external IP address.
|
||||
func (n *NAT) GetExternalAddress() (net.IP, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
return n.client.GetExternalAddress(ctx)
|
||||
}
|
||||
|
||||
// GetInternalAddress returns the local IP address used to communicate with the gateway.
|
||||
func (n *NAT) GetInternalAddress() (net.IP, error) {
|
||||
addr, err := n.client.getLocalIP()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return addr.AsSlice(), nil
|
||||
}
|
||||
|
||||
// AddPortMapping creates a port mapping on both IPv4 and IPv6 (if available).
|
||||
func (n *NAT) AddPortMapping(ctx context.Context, protocol string, internalPort int, _ string, timeout time.Duration) (int, error) {
|
||||
resp, err := n.client.AddPortMapping(ctx, protocol, internalPort, timeout)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("add mapping: %w", err)
|
||||
}
|
||||
|
||||
n.mu.RLock()
|
||||
client6 := n.client6
|
||||
localIP6 := n.localIP6
|
||||
n.mu.RUnlock()
|
||||
|
||||
if client6 == nil {
|
||||
return int(resp.ExternalPort), nil
|
||||
}
|
||||
|
||||
if _, err := client6.AddPortMapping(ctx, protocol, internalPort, timeout); err != nil {
|
||||
log.Warnf("IPv6 PCP mapping failed (continuing with IPv4): %v", err)
|
||||
return int(resp.ExternalPort), nil
|
||||
}
|
||||
|
||||
log.Infof("created IPv6 PCP pinhole: %s:%d", localIP6, internalPort)
|
||||
return int(resp.ExternalPort), nil
|
||||
}
|
||||
|
||||
// DeletePortMapping removes a port mapping from both IPv4 and IPv6.
|
||||
func (n *NAT) DeletePortMapping(ctx context.Context, protocol string, internalPort int) error {
|
||||
err := n.client.DeletePortMapping(ctx, protocol, internalPort)
|
||||
|
||||
n.mu.RLock()
|
||||
client6 := n.client6
|
||||
n.mu.RUnlock()
|
||||
|
||||
if client6 != nil {
|
||||
if err6 := client6.DeletePortMapping(ctx, protocol, internalPort); err6 != nil {
|
||||
log.Warnf("IPv6 PCP delete mapping failed: %v", err6)
|
||||
}
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return fmt.Errorf("delete mapping: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// CheckServerHealth sends an ANNOUNCE to verify the server is still responsive.
|
||||
// Returns the current epoch and whether the server may have restarted (epoch state loss detected).
|
||||
func (n *NAT) CheckServerHealth(ctx context.Context) (epoch uint32, serverRestarted bool, err error) {
|
||||
epoch, err = n.client.Announce(ctx)
|
||||
if err != nil {
|
||||
return 0, false, fmt.Errorf("announce: %w", err)
|
||||
}
|
||||
return epoch, n.client.EpochStateLost(), nil
|
||||
}
|
||||
|
||||
// DiscoverPCP attempts to discover a PCP-capable gateway.
|
||||
// Returns a NAT interface if PCP is supported, or an error otherwise.
|
||||
// Discovers both IPv4 and IPv6 gateways when available.
|
||||
func DiscoverPCP(ctx context.Context) (nat.NAT, error) {
|
||||
gateway, localIP, err := getDefaultGateway()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get default gateway: %w", err)
|
||||
}
|
||||
|
||||
client := NewClient(gateway)
|
||||
client.SetLocalIP(localIP)
|
||||
if _, err := client.Announce(ctx); err != nil {
|
||||
return nil, fmt.Errorf("PCP announce: %w", err)
|
||||
}
|
||||
|
||||
result := &NAT{client: client}
|
||||
discoverIPv6(ctx, result)
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func discoverIPv6(ctx context.Context, result *NAT) {
|
||||
gateway6, localIP6, err := getDefaultGateway6()
|
||||
if err != nil {
|
||||
log.Debugf("IPv6 gateway discovery failed: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
client6 := NewClient(gateway6)
|
||||
client6.SetLocalIP(localIP6)
|
||||
if _, err := client6.Announce(ctx); err != nil {
|
||||
log.Debugf("PCP IPv6 announce failed: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
addr, ok := netip.AddrFromSlice(localIP6)
|
||||
if !ok {
|
||||
log.Debugf("invalid IPv6 local IP: %v", localIP6)
|
||||
return
|
||||
}
|
||||
result.mu.Lock()
|
||||
result.client6 = client6
|
||||
result.localIP6 = addr
|
||||
result.mu.Unlock()
|
||||
log.Debugf("PCP IPv6 gateway discovered: %s (local: %s)", gateway6, localIP6)
|
||||
}
|
||||
|
||||
// getDefaultGateway returns the default IPv4 gateway and local IP using the system routing table.
|
||||
func getDefaultGateway() (gateway net.IP, localIP net.IP, err error) {
|
||||
router, err := netroute.New()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
dst := net.IPv4zero
|
||||
if runtime.GOOS == "linux" || runtime.GOOS == "android" {
|
||||
// go-netroute v0.4.0 rejects unspecified destinations client-side on Linux/Android.
|
||||
// TODO: on android/ios, use platform APIs (ConnectivityManager.getLinkProperties /
|
||||
// NWPathMonitor) when netlink-based lookup is restricted or unavailable.
|
||||
dst = net.IPv4(0, 0, 0, 1)
|
||||
}
|
||||
_, gateway, localIP, err = router.Route(dst)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
if gateway == nil {
|
||||
return nil, nil, nat.ErrNoNATFound
|
||||
}
|
||||
|
||||
return gateway, localIP, nil
|
||||
}
|
||||
|
||||
// getDefaultGateway6 returns the default IPv6 gateway IP address using the system routing table.
|
||||
func getDefaultGateway6() (gateway net.IP, localIP net.IP, err error) {
|
||||
router, err := netroute.New()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
dst := net.IPv6zero
|
||||
if runtime.GOOS == "linux" || runtime.GOOS == "android" {
|
||||
// ::2
|
||||
dst = net.IP{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2}
|
||||
}
|
||||
_, gateway, localIP, err = router.Route(dst)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
if gateway == nil {
|
||||
return nil, nil, nat.ErrNoNATFound
|
||||
}
|
||||
|
||||
return gateway, localIP, nil
|
||||
}
|
||||
225
client/internal/portforward/pcp/protocol.go
Normal file
225
client/internal/portforward/pcp/protocol.go
Normal file
@@ -0,0 +1,225 @@
|
||||
// Package pcp implements the Port Control Protocol (RFC 6887).
|
||||
//
|
||||
// # Implemented Features
|
||||
//
|
||||
// - ANNOUNCE opcode: Discovers PCP server support
|
||||
// - MAP opcode: Creates/deletes port mappings (IPv4 NAT) and firewall pinholes (IPv6)
|
||||
// - Dual-stack: Simultaneous IPv4 and IPv6 support via separate clients
|
||||
// - Nonce validation: Prevents response spoofing
|
||||
// - Epoch tracking: Detects server restarts per Section 8.5
|
||||
// - RFC-compliant retry timing: 3s initial, exponential backoff to 1024s max (Section 8.1.1)
|
||||
//
|
||||
// # Not Implemented
|
||||
//
|
||||
// - PEER opcode: For outbound peer connections (not needed for inbound NAT traversal)
|
||||
// - THIRD_PARTY option: For managing mappings on behalf of other devices
|
||||
// - PREFER_FAILURE option: Requires exact external port or fail (IPv4 NAT only, not needed for IPv6 pinholing)
|
||||
// - FILTER option: To restrict remote peer addresses
|
||||
//
|
||||
// These optional features are omitted because the primary use case is simple
|
||||
// port forwarding for WireGuard, which only requires MAP with default behavior.
|
||||
package pcp
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"net/netip"
|
||||
)
|
||||
|
||||
const (
|
||||
// Version is the PCP protocol version (RFC 6887).
|
||||
Version = 2
|
||||
|
||||
// Port is the standard PCP server port.
|
||||
Port = 5351
|
||||
|
||||
// DefaultLifetime is the default requested mapping lifetime in seconds.
|
||||
DefaultLifetime = 7200 // 2 hours
|
||||
|
||||
// Header sizes
|
||||
headerSize = 24
|
||||
mapPayloadSize = 36
|
||||
mapRequestSize = headerSize + mapPayloadSize // 60 bytes
|
||||
)
|
||||
|
||||
// Opcodes
|
||||
const (
|
||||
OpAnnounce = 0
|
||||
OpMap = 1
|
||||
OpPeer = 2
|
||||
OpReply = 0x80 // OR'd with opcode in responses
|
||||
)
|
||||
|
||||
// Protocol numbers for MAP requests
|
||||
const (
|
||||
ProtoUDP = 17
|
||||
ProtoTCP = 6
|
||||
)
|
||||
|
||||
// Result codes (RFC 6887 Section 7.4)
|
||||
const (
|
||||
ResultSuccess = 0
|
||||
ResultUnsuppVersion = 1
|
||||
ResultNotAuthorized = 2
|
||||
ResultMalformedRequest = 3
|
||||
ResultUnsuppOpcode = 4
|
||||
ResultUnsuppOption = 5
|
||||
ResultMalformedOption = 6
|
||||
ResultNetworkFailure = 7
|
||||
ResultNoResources = 8
|
||||
ResultUnsuppProtocol = 9
|
||||
ResultUserExQuota = 10
|
||||
ResultCannotProvideExt = 11
|
||||
ResultAddressMismatch = 12
|
||||
ResultExcessiveRemotePeers = 13
|
||||
)
|
||||
|
||||
// ResultCodeString returns a human-readable string for a result code.
|
||||
func ResultCodeString(code uint8) string {
|
||||
switch code {
|
||||
case ResultSuccess:
|
||||
return "SUCCESS"
|
||||
case ResultUnsuppVersion:
|
||||
return "UNSUPP_VERSION"
|
||||
case ResultNotAuthorized:
|
||||
return "NOT_AUTHORIZED"
|
||||
case ResultMalformedRequest:
|
||||
return "MALFORMED_REQUEST"
|
||||
case ResultUnsuppOpcode:
|
||||
return "UNSUPP_OPCODE"
|
||||
case ResultUnsuppOption:
|
||||
return "UNSUPP_OPTION"
|
||||
case ResultMalformedOption:
|
||||
return "MALFORMED_OPTION"
|
||||
case ResultNetworkFailure:
|
||||
return "NETWORK_FAILURE"
|
||||
case ResultNoResources:
|
||||
return "NO_RESOURCES"
|
||||
case ResultUnsuppProtocol:
|
||||
return "UNSUPP_PROTOCOL"
|
||||
case ResultUserExQuota:
|
||||
return "USER_EX_QUOTA"
|
||||
case ResultCannotProvideExt:
|
||||
return "CANNOT_PROVIDE_EXTERNAL"
|
||||
case ResultAddressMismatch:
|
||||
return "ADDRESS_MISMATCH"
|
||||
case ResultExcessiveRemotePeers:
|
||||
return "EXCESSIVE_REMOTE_PEERS"
|
||||
default:
|
||||
return fmt.Sprintf("UNKNOWN(%d)", code)
|
||||
}
|
||||
}
|
||||
|
||||
// Response represents a parsed PCP response header.
|
||||
type Response struct {
|
||||
Version uint8
|
||||
Opcode uint8
|
||||
ResultCode uint8
|
||||
Lifetime uint32
|
||||
Epoch uint32
|
||||
}
|
||||
|
||||
// MapResponse contains the full response to a MAP request.
|
||||
type MapResponse struct {
|
||||
Response
|
||||
Nonce [12]byte
|
||||
Protocol uint8
|
||||
InternalPort uint16
|
||||
ExternalPort uint16
|
||||
ExternalIP netip.Addr
|
||||
}
|
||||
|
||||
// addrTo16 converts an address to its 16-byte IPv4-mapped IPv6 representation.
|
||||
func addrTo16(addr netip.Addr) [16]byte {
|
||||
if addr.Is4() {
|
||||
return netip.AddrFrom4(addr.As4()).As16()
|
||||
}
|
||||
return addr.As16()
|
||||
}
|
||||
|
||||
// addrFrom16 extracts an address from a 16-byte representation, unmapping IPv4.
|
||||
func addrFrom16(b [16]byte) netip.Addr {
|
||||
return netip.AddrFrom16(b).Unmap()
|
||||
}
|
||||
|
||||
// buildAnnounceRequest creates a PCP ANNOUNCE request packet.
|
||||
func buildAnnounceRequest(clientIP netip.Addr) []byte {
|
||||
req := make([]byte, headerSize)
|
||||
req[0] = Version
|
||||
req[1] = OpAnnounce
|
||||
mapped := addrTo16(clientIP)
|
||||
copy(req[8:24], mapped[:])
|
||||
return req
|
||||
}
|
||||
|
||||
// buildMapRequest creates a PCP MAP request packet.
|
||||
func buildMapRequest(clientIP netip.Addr, nonce [12]byte, protocol uint8, internalPort, suggestedExtPort uint16, suggestedExtIP netip.Addr, lifetime uint32) []byte {
|
||||
req := make([]byte, mapRequestSize)
|
||||
|
||||
// Header
|
||||
req[0] = Version
|
||||
req[1] = OpMap
|
||||
binary.BigEndian.PutUint32(req[4:8], lifetime)
|
||||
mapped := addrTo16(clientIP)
|
||||
copy(req[8:24], mapped[:])
|
||||
|
||||
// MAP payload
|
||||
copy(req[24:36], nonce[:])
|
||||
req[36] = protocol
|
||||
binary.BigEndian.PutUint16(req[40:42], internalPort)
|
||||
binary.BigEndian.PutUint16(req[42:44], suggestedExtPort)
|
||||
if suggestedExtIP.IsValid() {
|
||||
extMapped := addrTo16(suggestedExtIP)
|
||||
copy(req[44:60], extMapped[:])
|
||||
}
|
||||
|
||||
return req
|
||||
}
|
||||
|
||||
// parseResponse parses the common PCP response header.
|
||||
func parseResponse(data []byte) (*Response, error) {
|
||||
if len(data) < headerSize {
|
||||
return nil, fmt.Errorf("response too short: %d bytes", len(data))
|
||||
}
|
||||
|
||||
resp := &Response{
|
||||
Version: data[0],
|
||||
Opcode: data[1],
|
||||
ResultCode: data[3], // Byte 2 is reserved, byte 3 is result code (RFC 6887 §7.2)
|
||||
Lifetime: binary.BigEndian.Uint32(data[4:8]),
|
||||
Epoch: binary.BigEndian.Uint32(data[8:12]),
|
||||
}
|
||||
|
||||
if resp.Version != Version {
|
||||
return nil, fmt.Errorf("unsupported PCP version: %d", resp.Version)
|
||||
}
|
||||
|
||||
if resp.Opcode&OpReply == 0 {
|
||||
return nil, fmt.Errorf("response missing reply bit: opcode=0x%02x", resp.Opcode)
|
||||
}
|
||||
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// parseMapResponse parses a complete MAP response.
|
||||
func parseMapResponse(data []byte) (*MapResponse, error) {
|
||||
if len(data) < mapRequestSize {
|
||||
return nil, fmt.Errorf("MAP response too short: %d bytes", len(data))
|
||||
}
|
||||
|
||||
resp, err := parseResponse(data)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse header: %w", err)
|
||||
}
|
||||
|
||||
mapResp := &MapResponse{
|
||||
Response: *resp,
|
||||
Protocol: data[36],
|
||||
InternalPort: binary.BigEndian.Uint16(data[40:42]),
|
||||
ExternalPort: binary.BigEndian.Uint16(data[42:44]),
|
||||
ExternalIP: addrFrom16([16]byte(data[44:60])),
|
||||
}
|
||||
copy(mapResp.Nonce[:], data[24:36])
|
||||
|
||||
return mapResp, nil
|
||||
}
|
||||
@@ -1,116 +0,0 @@
|
||||
//go:build !js
|
||||
|
||||
package portforward
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/netbirdio/go-nat"
|
||||
log "github.com/sirupsen/logrus"
|
||||
"github.com/sirupsen/logrus/hooks/test"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// mockPinholeNAT is a gateway that also reports an IPv6 pinhole outcome, the
|
||||
// shape a dual-stack gateway has.
|
||||
type mockPinholeNAT struct {
|
||||
*mockNAT
|
||||
pinholeErr error
|
||||
}
|
||||
|
||||
func (m *mockPinholeNAT) IPv6PinholeError() error {
|
||||
return m.pinholeErr
|
||||
}
|
||||
|
||||
func TestSetupLogsPinholeOutcome(t *testing.T) {
|
||||
pinholeErr := errors.New("pcp ipv6: NOT_AUTHORIZED")
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
pinholeErr error
|
||||
mappingErr error
|
||||
wantLevel log.Level
|
||||
wantText string
|
||||
}{
|
||||
{
|
||||
name: "an open pinhole is reported",
|
||||
wantLevel: log.InfoLevel,
|
||||
wantText: "IPv6 pinhole open",
|
||||
},
|
||||
{
|
||||
name: "a failed pinhole is reported without failing the mapping",
|
||||
// The IPv4 mapping is what the caller asked for, so the pinhole
|
||||
// failure surfaces only in the log.
|
||||
pinholeErr: pinholeErr,
|
||||
wantLevel: log.WarnLevel,
|
||||
wantText: pinholeErr.Error(),
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
gateway := &mockPinholeNAT{mockNAT: newMockNAT(), pinholeErr: tt.pinholeErr}
|
||||
hook := stubGatewayDiscovery(t, gateway)
|
||||
|
||||
m := NewManager()
|
||||
m.wgPort = 51820
|
||||
|
||||
_, mapping, err := m.setup(context.Background())
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, mapping)
|
||||
|
||||
entry := findEntry(hook, tt.wantText)
|
||||
require.NotNil(t, entry, "no log entry mentioning %q", tt.wantText)
|
||||
assert.Equal(t, tt.wantLevel, entry.Level)
|
||||
})
|
||||
}
|
||||
|
||||
t.Run("a failed mapping reports no pinhole outcome", func(t *testing.T) {
|
||||
// Nothing opened the pinhole, so whatever it currently reports says
|
||||
// nothing about this attempt.
|
||||
gateway := &mockPinholeNAT{mockNAT: newMockNAT()}
|
||||
gateway.addMappingErr = errors.New("gateway refused")
|
||||
hook := stubGatewayDiscovery(t, gateway)
|
||||
|
||||
m := NewManager()
|
||||
m.wgPort = 51820
|
||||
|
||||
_, _, err := m.setup(context.Background())
|
||||
|
||||
require.Error(t, err)
|
||||
assert.Nil(t, findEntry(hook, "IPv6 pinhole"))
|
||||
})
|
||||
}
|
||||
|
||||
// stubGatewayDiscovery makes discovery return gateway and captures log output.
|
||||
func stubGatewayDiscovery(t *testing.T, gateway nat.NAT) *test.Hook {
|
||||
t.Helper()
|
||||
|
||||
orig := discoverGateway
|
||||
discoverGateway = func(context.Context) (nat.NAT, error) { return gateway, nil }
|
||||
t.Cleanup(func() { discoverGateway = orig })
|
||||
|
||||
hook := test.NewGlobal()
|
||||
origLevel := log.GetLevel()
|
||||
log.SetLevel(log.DebugLevel)
|
||||
t.Cleanup(func() {
|
||||
hook.Reset()
|
||||
log.SetLevel(origLevel)
|
||||
})
|
||||
|
||||
return hook
|
||||
}
|
||||
|
||||
func findEntry(hook *test.Hook, substr string) *log.Entry {
|
||||
for _, entry := range hook.AllEntries() {
|
||||
if strings.Contains(entry.Message, substr) {
|
||||
return entry
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -4,94 +4,27 @@ package portforward
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/netbirdio/go-nat"
|
||||
"github.com/netbirdio/go-nat/pcp"
|
||||
"github.com/libp2p/go-nat"
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/internal/portforward/pcp"
|
||||
)
|
||||
|
||||
// discoverGateway is the function used for NAT gateway discovery.
|
||||
// It can be replaced in tests to avoid real network operations.
|
||||
// Tries PCP first, then falls back to NAT-PMP/UPnP.
|
||||
var discoverGateway = defaultDiscoverGateway
|
||||
|
||||
// pinholeDiscoveryTimeout is the slice of the discovery budget held back for
|
||||
// the IPv6 pinhole probe.
|
||||
//
|
||||
// Sizing it is coarser than it looks: PCP retransmits on a 3s socket timeout
|
||||
// and a 3s first backoff, so a second attempt needs about 9s. Anything from
|
||||
// roughly 1s to 8s therefore buys exactly one attempt, and this only sets how
|
||||
// long that attempt waits. A PCP server sits on the local link and answers in
|
||||
// milliseconds, so 3s is margin rather than need, and the rest is left to
|
||||
// gateway discovery, whose multicast SSDP search alone takes 5s. A probe lost
|
||||
// to a dropped packet is retried by the next discovery round.
|
||||
//
|
||||
// It is a variable so tests can shorten it.
|
||||
var pinholeDiscoveryTimeout = 3 * time.Second
|
||||
|
||||
// Discovery entry points, as variables so tests can drive the fallback without
|
||||
// touching the network.
|
||||
var (
|
||||
discoverNATGateway = nat.DiscoverGateway
|
||||
|
||||
discoverPCPPinhole = func(ctx context.Context) (nat.NAT, error) {
|
||||
pinhole, err := pcp.DiscoverPCP(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return pinhole, nil
|
||||
}
|
||||
)
|
||||
|
||||
// defaultDiscoverGateway finds a gateway that can make the WireGuard port
|
||||
// reachable. DiscoverGateway prefers PCP for IPv4, races UPnP and NAT-PMP
|
||||
// behind it, and attaches an IPv6 pinhole independently of which IPv4 protocol
|
||||
// wins.
|
||||
//
|
||||
// It reports no gateway on a network offering only IPv6, having no IPv4 mapping
|
||||
// to attach a pinhole to. Such a network still needs one: there is no
|
||||
// translation to traverse, but the router drops inbound IPv6 until something
|
||||
// opens it. Fall back to PCP alone, which yields a gateway holding just the
|
||||
// pinhole.
|
||||
func defaultDiscoverGateway(ctx context.Context) (nat.NAT, error) {
|
||||
gatewayCtx, cancel := reserveForPinhole(ctx)
|
||||
defer cancel()
|
||||
|
||||
gateway, err := discoverNATGateway(gatewayCtx)
|
||||
pcpGateway, err := pcp.DiscoverPCP(ctx)
|
||||
if err == nil {
|
||||
return gateway, nil
|
||||
}
|
||||
if !errors.Is(err, nat.ErrNoNATFound) {
|
||||
return nil, err
|
||||
return pcpGateway, nil
|
||||
}
|
||||
log.Debugf("PCP discovery failed: %v, trying NAT-PMP/UPnP", err)
|
||||
|
||||
pinhole, pinholeErr := discoverPCPPinhole(ctx)
|
||||
if pinholeErr != nil {
|
||||
log.Debugf("no IPv6 pinhole after %v: %v", err, pinholeErr)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
log.Infof("no IPv4 gateway, continuing with an IPv6 pinhole only")
|
||||
return pinhole, nil
|
||||
}
|
||||
|
||||
// reserveForPinhole shortens ctx so that a pinhole probe still has time to run
|
||||
// afterwards. Finding nothing takes gateway discovery everything it is given,
|
||||
// so on the unshortened context the probe would start already expired. A budget
|
||||
// too small to divide is left to gateway discovery, which is the likelier win.
|
||||
func reserveForPinhole(ctx context.Context) (context.Context, context.CancelFunc) {
|
||||
deadline, ok := ctx.Deadline()
|
||||
if !ok {
|
||||
return context.WithCancel(ctx)
|
||||
}
|
||||
|
||||
remaining := time.Until(deadline)
|
||||
if remaining <= pinholeDiscoveryTimeout {
|
||||
return context.WithCancel(ctx)
|
||||
}
|
||||
return context.WithTimeout(ctx, remaining-pinholeDiscoveryTimeout)
|
||||
return nat.DiscoverGateway(ctx)
|
||||
}
|
||||
|
||||
// State is persisted only for crash recovery cleanup
|
||||
|
||||
@@ -1,140 +0,0 @@
|
||||
//go:build !js
|
||||
|
||||
package portforward
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/netbirdio/go-nat"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// stubDiscovery replaces both discovery entry points for the duration of a
|
||||
// test. gatewayDelay simulates gateway discovery spending everything it is
|
||||
// given before reporting that it found nothing.
|
||||
func stubDiscovery(t *testing.T, gateway nat.NAT, gatewayErr error, gatewayDelay time.Duration, pinhole nat.NAT, pinholeErr error) {
|
||||
t.Helper()
|
||||
|
||||
origGateway, origPinhole := discoverNATGateway, discoverPCPPinhole
|
||||
discoverNATGateway = func(ctx context.Context) (nat.NAT, error) {
|
||||
if gatewayDelay > 0 {
|
||||
select {
|
||||
case <-time.After(gatewayDelay):
|
||||
case <-ctx.Done():
|
||||
}
|
||||
}
|
||||
return gateway, gatewayErr
|
||||
}
|
||||
discoverPCPPinhole = func(ctx context.Context) (nat.NAT, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return pinhole, pinholeErr
|
||||
}
|
||||
|
||||
t.Cleanup(func() { discoverNATGateway, discoverPCPPinhole = origGateway, origPinhole })
|
||||
}
|
||||
|
||||
func TestDefaultDiscoverGateway(t *testing.T) {
|
||||
ipv4Gateway := &mockNAT{natType: "PCP+PCPv6"}
|
||||
ipv6Pinhole := &mockNAT{natType: "PCP"}
|
||||
otherErr := errors.New("routing table unavailable")
|
||||
|
||||
t.Run("an IPv4 gateway is used as is", func(t *testing.T) {
|
||||
stubDiscovery(t, ipv4Gateway, nil, 0, ipv6Pinhole, nil)
|
||||
|
||||
got, err := defaultDiscoverGateway(context.Background())
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Same(t, ipv4Gateway, got)
|
||||
})
|
||||
|
||||
t.Run("no IPv4 gateway still opens an IPv6 pinhole", func(t *testing.T) {
|
||||
stubDiscovery(t, nil, nat.ErrNoNATFound, 0, ipv6Pinhole, nil)
|
||||
|
||||
got, err := defaultDiscoverGateway(context.Background())
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Same(t, ipv6Pinhole, got)
|
||||
})
|
||||
|
||||
t.Run("no gateway and no pinhole reports the original failure", func(t *testing.T) {
|
||||
stubDiscovery(t, nil, nat.ErrNoNATFound, 0, nil, errors.New("no IPv6 route"))
|
||||
|
||||
got, err := defaultDiscoverGateway(context.Background())
|
||||
|
||||
assert.Nil(t, got)
|
||||
assert.ErrorIs(t, err, nat.ErrNoNATFound, "the pinhole failure must not mask why no gateway was found")
|
||||
})
|
||||
|
||||
t.Run("a failure other than no-gateway is reported as is", func(t *testing.T) {
|
||||
stubDiscovery(t, nil, otherErr, 0, ipv6Pinhole, nil)
|
||||
|
||||
got, err := defaultDiscoverGateway(context.Background())
|
||||
|
||||
assert.Nil(t, got)
|
||||
assert.ErrorIs(t, err, otherErr)
|
||||
})
|
||||
|
||||
t.Run("the pinhole survives gateway discovery using its whole budget", func(t *testing.T) {
|
||||
// On one shared context the probe would start already expired, which is
|
||||
// how this failed against a real gateway.
|
||||
reserve := 50 * time.Millisecond
|
||||
origReserve := pinholeDiscoveryTimeout
|
||||
pinholeDiscoveryTimeout = reserve
|
||||
t.Cleanup(func() { pinholeDiscoveryTimeout = origReserve })
|
||||
|
||||
budget := 4 * reserve
|
||||
ctx, cancel := context.WithTimeout(context.Background(), budget)
|
||||
defer cancel()
|
||||
|
||||
stubDiscovery(t, nil, nat.ErrNoNATFound, budget, ipv6Pinhole, nil)
|
||||
|
||||
got, err := defaultDiscoverGateway(ctx)
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Same(t, ipv6Pinhole, got)
|
||||
})
|
||||
}
|
||||
|
||||
func TestReserveForPinhole(t *testing.T) {
|
||||
origReserve := pinholeDiscoveryTimeout
|
||||
pinholeDiscoveryTimeout = time.Second
|
||||
t.Cleanup(func() { pinholeDiscoveryTimeout = origReserve })
|
||||
|
||||
t.Run("a budget is divided", func(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
gatewayCtx, cancelGateway := reserveForPinhole(ctx)
|
||||
defer cancelGateway()
|
||||
|
||||
deadline, ok := gatewayCtx.Deadline()
|
||||
require.True(t, ok)
|
||||
assert.InDelta(t, 9*time.Second, time.Until(deadline), float64(500*time.Millisecond))
|
||||
})
|
||||
|
||||
t.Run("a budget too small to divide is left whole", func(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
gatewayCtx, cancelGateway := reserveForPinhole(ctx)
|
||||
defer cancelGateway()
|
||||
|
||||
deadline, ok := gatewayCtx.Deadline()
|
||||
require.True(t, ok)
|
||||
assert.InDelta(t, 500*time.Millisecond, time.Until(deadline), float64(100*time.Millisecond))
|
||||
})
|
||||
|
||||
t.Run("no deadline stays unbounded", func(t *testing.T) {
|
||||
gatewayCtx, cancelGateway := reserveForPinhole(context.Background())
|
||||
defer cancelGateway()
|
||||
|
||||
_, ok := gatewayCtx.Deadline()
|
||||
assert.False(t, ok)
|
||||
})
|
||||
}
|
||||
@@ -1,130 +0,0 @@
|
||||
package profilemanager
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
|
||||
"github.com/netbirdio/netbird/util"
|
||||
)
|
||||
|
||||
const prefsFileSuffix = ".prefs.json"
|
||||
|
||||
var prefsMu sync.Mutex
|
||||
|
||||
// Prefs is a namespaced per-profile preference store backed by a single JSON
|
||||
// file next to the profile config; it is deleted together with the profile.
|
||||
type Prefs struct {
|
||||
path string
|
||||
}
|
||||
|
||||
// ProfilePrefs returns the preference store of the profile identified by id.
|
||||
func (s *ServiceManager) ProfilePrefs(id ID, username string) (*Prefs, error) {
|
||||
if !IsValidProfileFilenameStem(id) {
|
||||
return nil, fmt.Errorf("invalid profile ID: %q", id)
|
||||
}
|
||||
if id == defaultProfileName {
|
||||
return &Prefs{path: filepath.Join(filepath.Dir(DefaultConfigPath), id.String()+prefsFileSuffix)}, nil
|
||||
}
|
||||
configDir, err := s.getConfigDir(username)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get config directory for user %s: %w", username, err)
|
||||
}
|
||||
return &Prefs{path: filepath.Join(configDir, id.String()+prefsFileSuffix)}, nil
|
||||
}
|
||||
|
||||
// Get unmarshals the namespace section into v and reports whether it exists.
|
||||
func (p *Prefs) Get(namespace string, v any) (bool, error) {
|
||||
if namespace == "" {
|
||||
return false, fmt.Errorf("empty prefs namespace")
|
||||
}
|
||||
|
||||
prefsMu.Lock()
|
||||
defer prefsMu.Unlock()
|
||||
|
||||
sections, err := readPrefsFile(p.path)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
raw, ok := sections[namespace]
|
||||
if !ok {
|
||||
return false, nil
|
||||
}
|
||||
if err := json.Unmarshal(raw, v); err != nil {
|
||||
return false, fmt.Errorf("decode prefs namespace %q: %w", namespace, err)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Put stores v as the namespace section, replacing any previous value.
|
||||
func (p *Prefs) Put(namespace string, v any) error {
|
||||
if namespace == "" {
|
||||
return fmt.Errorf("empty prefs namespace")
|
||||
}
|
||||
raw, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
return fmt.Errorf("encode prefs namespace %q: %w", namespace, err)
|
||||
}
|
||||
|
||||
prefsMu.Lock()
|
||||
defer prefsMu.Unlock()
|
||||
|
||||
sections, err := readPrefsFile(p.path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
sections[namespace] = raw
|
||||
return writePrefsFile(p.path, sections)
|
||||
}
|
||||
|
||||
// Remove deletes the namespace section; a missing one is not an error.
|
||||
func (p *Prefs) Remove(namespace string) error {
|
||||
if namespace == "" {
|
||||
return fmt.Errorf("empty prefs namespace")
|
||||
}
|
||||
|
||||
prefsMu.Lock()
|
||||
defer prefsMu.Unlock()
|
||||
|
||||
sections, err := readPrefsFile(p.path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, ok := sections[namespace]; !ok {
|
||||
return nil
|
||||
}
|
||||
delete(sections, namespace)
|
||||
return writePrefsFile(p.path, sections)
|
||||
}
|
||||
|
||||
func removePrefsFile(path string) error {
|
||||
prefsMu.Lock()
|
||||
defer prefsMu.Unlock()
|
||||
return os.Remove(path)
|
||||
}
|
||||
|
||||
func readPrefsFile(path string) (map[string]json.RawMessage, error) {
|
||||
data, err := os.ReadFile(path)
|
||||
if os.IsNotExist(err) {
|
||||
return map[string]json.RawMessage{}, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read prefs: %w", err)
|
||||
}
|
||||
|
||||
sections := map[string]json.RawMessage{}
|
||||
if err := json.Unmarshal(data, §ions); err != nil {
|
||||
return nil, fmt.Errorf("decode prefs: %w", err)
|
||||
}
|
||||
return sections, nil
|
||||
}
|
||||
|
||||
func writePrefsFile(path string, sections map[string]json.RawMessage) error {
|
||||
if err := util.WriteJsonWithRestrictedPermission(context.Background(), path, sections); err != nil {
|
||||
return fmt.Errorf("write prefs: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -1,138 +0,0 @@
|
||||
package profilemanager
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
type testPrefsSection struct {
|
||||
Mode uint8 `json:"mode"`
|
||||
Dest string `json:"dest"`
|
||||
}
|
||||
|
||||
func TestProfilePrefs_RoundTrip(t *testing.T) {
|
||||
withTestSM(t, func(sm *ServiceManager, username string) {
|
||||
created, err := sm.AddProfile("work", username)
|
||||
require.NoError(t, err)
|
||||
|
||||
prefs, err := sm.ProfilePrefs(created.ID, username)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NoError(t, prefs.Put("filedrop", testPrefsSection{Mode: 2, Dest: "/tmp/x"}))
|
||||
require.NoError(t, prefs.Put("other", map[string]int{"n": 1}))
|
||||
|
||||
var got testPrefsSection
|
||||
found, err := prefs.Get("filedrop", &got)
|
||||
require.NoError(t, err)
|
||||
assert.True(t, found)
|
||||
assert.Equal(t, testPrefsSection{Mode: 2, Dest: "/tmp/x"}, got)
|
||||
|
||||
var other map[string]int
|
||||
found, err = prefs.Get("other", &other)
|
||||
require.NoError(t, err)
|
||||
assert.True(t, found)
|
||||
assert.Equal(t, map[string]int{"n": 1}, other)
|
||||
})
|
||||
}
|
||||
|
||||
func TestProfilePrefs_GetMissingNamespace(t *testing.T) {
|
||||
withTestSM(t, func(sm *ServiceManager, username string) {
|
||||
created, err := sm.AddProfile("work", username)
|
||||
require.NoError(t, err)
|
||||
|
||||
prefs, err := sm.ProfilePrefs(created.ID, username)
|
||||
require.NoError(t, err)
|
||||
|
||||
var got testPrefsSection
|
||||
found, err := prefs.Get("filedrop", &got)
|
||||
require.NoError(t, err)
|
||||
assert.False(t, found)
|
||||
})
|
||||
}
|
||||
|
||||
func TestProfilePrefs_RemoveNamespace(t *testing.T) {
|
||||
withTestSM(t, func(sm *ServiceManager, username string) {
|
||||
created, err := sm.AddProfile("work", username)
|
||||
require.NoError(t, err)
|
||||
|
||||
prefs, err := sm.ProfilePrefs(created.ID, username)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NoError(t, prefs.Put("filedrop", testPrefsSection{Mode: 1}))
|
||||
require.NoError(t, prefs.Put("other", map[string]int{"n": 1}))
|
||||
require.NoError(t, prefs.Remove("filedrop"))
|
||||
require.NoError(t, prefs.Remove("missing"))
|
||||
|
||||
var got testPrefsSection
|
||||
found, err := prefs.Get("filedrop", &got)
|
||||
require.NoError(t, err)
|
||||
assert.False(t, found)
|
||||
|
||||
var other map[string]int
|
||||
found, err = prefs.Get("other", &other)
|
||||
require.NoError(t, err)
|
||||
assert.True(t, found)
|
||||
assert.Equal(t, map[string]int{"n": 1}, other)
|
||||
})
|
||||
}
|
||||
|
||||
func TestProfilePrefs_RejectsInvalidID(t *testing.T) {
|
||||
withTestSM(t, func(sm *ServiceManager, username string) {
|
||||
_, err := sm.ProfilePrefs("../escape", username)
|
||||
assert.Error(t, err)
|
||||
})
|
||||
}
|
||||
|
||||
func TestProfilePrefs_RejectsEmptyNamespace(t *testing.T) {
|
||||
withTestSM(t, func(sm *ServiceManager, username string) {
|
||||
created, err := sm.AddProfile("work", username)
|
||||
require.NoError(t, err)
|
||||
|
||||
prefs, err := sm.ProfilePrefs(created.ID, username)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = prefs.Get("", &testPrefsSection{})
|
||||
assert.Error(t, err)
|
||||
assert.Error(t, prefs.Put("", testPrefsSection{}))
|
||||
assert.Error(t, prefs.Remove(""))
|
||||
})
|
||||
}
|
||||
|
||||
func TestProfilePrefs_DefaultProfile(t *testing.T) {
|
||||
withTestSM(t, func(sm *ServiceManager, username string) {
|
||||
prefs, err := sm.ProfilePrefs(defaultProfileName, username)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NoError(t, prefs.Put("filedrop", testPrefsSection{Mode: 1}))
|
||||
|
||||
expected := filepath.Join(filepath.Dir(DefaultConfigPath), "default"+prefsFileSuffix)
|
||||
_, err = os.Stat(expected)
|
||||
require.NoError(t, err)
|
||||
})
|
||||
}
|
||||
|
||||
func TestRemoveProfile_DeletesPrefsFile(t *testing.T) {
|
||||
withTestSM(t, func(sm *ServiceManager, username string) {
|
||||
created, err := sm.AddProfile("work", username)
|
||||
require.NoError(t, err)
|
||||
|
||||
prefs, err := sm.ProfilePrefs(created.ID, username)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, prefs.Put("filedrop", testPrefsSection{Mode: 2}))
|
||||
|
||||
configDir, err := sm.getConfigDir(username)
|
||||
require.NoError(t, err)
|
||||
prefsPath := filepath.Join(configDir, created.ID.String()+prefsFileSuffix)
|
||||
_, err = os.Stat(prefsPath)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NoError(t, sm.RemoveProfile(created.ID, username))
|
||||
_, err = os.Stat(prefsPath)
|
||||
assert.True(t, errors.Is(err, os.ErrNotExist), "prefs file should be removed")
|
||||
})
|
||||
}
|
||||
@@ -420,11 +420,6 @@ func (s *ServiceManager) RemoveProfile(id ID, username string) error {
|
||||
log.Warnf("failed to remove profile state file %s: %v", stateFile, err)
|
||||
}
|
||||
|
||||
prefsFile := filepath.Join(filepath.Dir(target.Path), id.String()+prefsFileSuffix)
|
||||
if err := removePrefsFile(prefsFile); err != nil && !os.IsNotExist(err) {
|
||||
log.Warnf("failed to remove profile prefs file %s: %v", prefsFile, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -18,8 +18,8 @@ type Service struct {
|
||||
}
|
||||
|
||||
func New() (*Service, error) {
|
||||
d, err := NewDetector() //nolint:staticcheck
|
||||
if err != nil { //nolint:staticcheck // always errors on platforms without a sleep detector
|
||||
d, err := NewDetector()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
|
||||
@@ -37,32 +37,23 @@
|
||||
// Updater Process (Setup):
|
||||
//
|
||||
// 1. Receives parameters from service via command-line arguments
|
||||
// 2. Terminates the UI so the installer does not have to replace a locked image
|
||||
// file, which would otherwise leave the install needing a reboot
|
||||
// 3. Runs installer with appropriate silent/quiet flags:
|
||||
// 2. Runs installer with appropriate silent/quiet flags:
|
||||
// - Windows EXE: installer.exe /S
|
||||
// - Windows MSI: msiexec.exe /i installer.msi /qn /norestart REBOOT=ReallySuppress /l*v msi.log
|
||||
// - Windows MSI: msiexec.exe /i installer.msi /quiet /qn /l*v msi.log
|
||||
// - macOS PKG: installer -pkg installer.pkg -target /
|
||||
// - macOS Homebrew: brew upgrade netbirdio/tap/netbird
|
||||
// 4. Installer terminates the daemon
|
||||
// 5. Installer replaces binaries with new version
|
||||
// 6. Updater waits for installer to complete. On Windows, MSI exit codes 3010
|
||||
// (ERROR_SUCCESS_REBOOT_REQUIRED) and 1641 (ERROR_SUCCESS_REBOOT_INITIATED)
|
||||
// are a pending-reboot outcome, not a failure: the install succeeded, but
|
||||
// some files are only replaced on the next restart (the reboot itself is
|
||||
// suppressed via /norestart and REBOOT=ReallySuppress), and the flow
|
||||
// continues as on success
|
||||
// 7. Updater restarts daemon:
|
||||
// 3. Installer terminates daemon and UI processes
|
||||
// 4. Installer replaces binaries with new version
|
||||
// 5. Updater waits for installer to complete
|
||||
// 6. Updater restarts daemon:
|
||||
// - Windows: netbird.exe service start
|
||||
// - macOS/Linux: netbird service start
|
||||
// 8. Updater restarts UI:
|
||||
// - Windows: Launches netbird-ui.exe using CreateProcessAsUser in every
|
||||
// session it was terminated in, falling back to the active console session
|
||||
// 7. Updater restarts UI:
|
||||
// - Windows: Launches netbird-ui.exe as active console user using CreateProcessAsUser
|
||||
// - macOS: Uses launchctl asuser to launch NetBird.app for console user
|
||||
// - Linux: Not implemented (UI typically auto-starts)
|
||||
// 9. Updater writes result.json with success/error status (a pending reboot is
|
||||
// recorded as success)
|
||||
// 10. Updater process exits
|
||||
// 8. Updater writes result.json with success/error status
|
||||
// 9. Updater process exits
|
||||
//
|
||||
// # Result Communication
|
||||
//
|
||||
|
||||
@@ -42,9 +42,6 @@ func NewWithDir(tempDir string) *Installer {
|
||||
// This will run by the original service process
|
||||
func (u *Installer) RunInstallation(ctx context.Context, targetVersion string) (err error) {
|
||||
resultHandler := NewResultHandler(u.tempDir)
|
||||
if err := resultHandler.ClearStaleResult(); err != nil {
|
||||
log.Warnf("clear stale installer result: %v", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
if err != nil {
|
||||
|
||||
@@ -2,7 +2,6 @@ package installer
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
@@ -23,12 +22,6 @@ const (
|
||||
|
||||
msiLogFile = "msi.log"
|
||||
|
||||
// ERROR_SUCCESS_REBOOT_REQUIRED and ERROR_SUCCESS_REBOOT_INITIATED
|
||||
msiRebootRequired = 3010
|
||||
msiRebootInitiated = 1641
|
||||
|
||||
processExitWait = 10 * time.Second
|
||||
|
||||
msiDownloadURL = "https://github.com/netbirdio/netbird/releases/download/v%version/netbird_installer_%version_windows_%arch.msi"
|
||||
exeDownloadURL = "https://github.com/netbirdio/netbird/releases/download/v%version/netbird_installer_%version_windows_%arch.exe"
|
||||
)
|
||||
@@ -45,8 +38,6 @@ var (
|
||||
func (u *Installer) Setup(ctx context.Context, dryRun bool, installerFile string, daemonFolder string) (resultErr error) {
|
||||
resultHandler := NewResultHandler(u.tempDir)
|
||||
|
||||
var uiSessions []uint32
|
||||
|
||||
// Always ensure daemon and UI are restarted after setup
|
||||
defer func() {
|
||||
log.Infof("starting daemon back")
|
||||
@@ -55,7 +46,7 @@ func (u *Installer) Setup(ctx context.Context, dryRun bool, installerFile string
|
||||
}
|
||||
|
||||
log.Infof("starting UI back")
|
||||
if err := u.startUI(daemonFolder, uiSessions); err != nil {
|
||||
if err := u.startUIAsUser(daemonFolder); err != nil {
|
||||
log.Errorf("failed to start UI: %v", err)
|
||||
}
|
||||
|
||||
@@ -84,14 +75,6 @@ func (u *Installer) Setup(ctx context.Context, dryRun bool, installerFile string
|
||||
return
|
||||
}
|
||||
|
||||
// The UI holds an open handle on its own image. Left running, Restart Manager
|
||||
// cannot shut it down (msiexec runs as LocalSystem here, the UI as the
|
||||
// interactive user), so the MSI falls back to replacing the file on reboot and
|
||||
// marks the install as restart-required. The deferred close-application action
|
||||
// in the package runs too late to prevent that, it happens after
|
||||
// InstallValidate has already registered the file as in use.
|
||||
uiSessions = killUI()
|
||||
|
||||
var cmd *exec.Cmd
|
||||
switch installerType {
|
||||
case TypeExe:
|
||||
@@ -101,9 +84,7 @@ func (u *Installer) Setup(ctx context.Context, dryRun bool, installerFile string
|
||||
installerDir := filepath.Dir(installerFile)
|
||||
logPath := filepath.Join(installerDir, msiLogFile)
|
||||
log.Infof("run msi installer: %s", installerFile)
|
||||
// REBOOT=ReallySuppress: a silent install has no way to ask, so without it
|
||||
// msiexec reboots the machine on its own if it decides one is needed.
|
||||
cmd = exec.CommandContext(ctx, "msiexec.exe", "/i", filepath.Base(installerFile), "/qn", "/norestart", "REBOOT=ReallySuppress", "/l*v", logPath)
|
||||
cmd = exec.CommandContext(ctx, "msiexec.exe", "/i", filepath.Base(installerFile), "/quiet", "/qn", "/l*v", logPath)
|
||||
}
|
||||
|
||||
cmd.Dir = filepath.Dir(installerFile)
|
||||
@@ -114,13 +95,9 @@ func (u *Installer) Setup(ctx context.Context, dryRun bool, installerFile string
|
||||
}
|
||||
|
||||
log.Infof("installer started with PID %d", cmd.Process.Pid)
|
||||
if err := cmd.Wait(); err != nil {
|
||||
if !isRebootPending(err) {
|
||||
resultErr = err
|
||||
log.Errorf("installer process finished with error: %v", err)
|
||||
return
|
||||
}
|
||||
log.Warnf("installer completed but reported a pending reboot, some files will be replaced on the next restart")
|
||||
if resultErr = cmd.Wait(); resultErr != nil {
|
||||
log.Errorf("installer process finished with error: %v", resultErr)
|
||||
return
|
||||
}
|
||||
|
||||
return nil
|
||||
@@ -140,142 +117,16 @@ func (u *Installer) startDaemon(daemonFolder string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (u *Installer) startUI(daemonFolder string, sessionIDs []uint32) error {
|
||||
func (u *Installer) startUIAsUser(daemonFolder string) error {
|
||||
uiPath := filepath.Join(daemonFolder, uiName)
|
||||
log.Infof("starting netbird-ui: %s", uiPath)
|
||||
|
||||
if len(sessionIDs) == 0 {
|
||||
sessionID := windows.WTSGetActiveConsoleSessionId()
|
||||
if sessionID == 0xFFFFFFFF {
|
||||
return fmt.Errorf("no active user session found")
|
||||
}
|
||||
sessionIDs = []uint32{sessionID}
|
||||
// Get the active console session ID
|
||||
sessionID := windows.WTSGetActiveConsoleSessionId()
|
||||
if sessionID == 0xFFFFFFFF {
|
||||
return fmt.Errorf("no active user session found")
|
||||
}
|
||||
|
||||
var errs []error
|
||||
for _, sessionID := range sessionIDs {
|
||||
if err := startUIInSession(uiPath, sessionID); err != nil {
|
||||
errs = append(errs, fmt.Errorf("session %d: %w", sessionID, err))
|
||||
continue
|
||||
}
|
||||
log.Infof("netbird-ui started successfully in session %d", sessionID)
|
||||
}
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
// isRebootPending reports whether the installer exit code means it succeeded but
|
||||
// left work for the next restart. The reboot itself is suppressed, so this is not
|
||||
// a failure.
|
||||
func isRebootPending(err error) bool {
|
||||
var exitErr *exec.ExitError
|
||||
if !errors.As(err, &exitErr) {
|
||||
return false
|
||||
}
|
||||
|
||||
switch exitErr.ExitCode() {
|
||||
case msiRebootRequired, msiRebootInitiated:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// killUI terminates any running netbird-ui process and returns the IDs of the
|
||||
// interactive sessions the terminated processes belonged to. Setup starts the
|
||||
// UI again in those sessions once the installer is done.
|
||||
func killUI() []uint32 {
|
||||
pids, err := processIDsByName(uiName)
|
||||
if err != nil {
|
||||
log.Warnf("failed to look up %s processes: %v", uiName, err)
|
||||
return nil
|
||||
}
|
||||
|
||||
sessions := make(map[uint32]struct{})
|
||||
for _, pid := range pids {
|
||||
var sessionID uint32
|
||||
if err := windows.ProcessIdToSessionId(pid, &sessionID); err != nil {
|
||||
log.Warnf("failed to look up session of %s (PID %d): %v", uiName, pid, err)
|
||||
}
|
||||
|
||||
if err := terminateProcess(pid); err != nil {
|
||||
log.Warnf("failed to terminate %s (PID %d): %v", uiName, pid, err)
|
||||
continue
|
||||
}
|
||||
log.Infof("terminated %s (PID %d) in session %d", uiName, pid, sessionID)
|
||||
|
||||
if sessionID != 0 {
|
||||
sessions[sessionID] = struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
sessionIDs := make([]uint32, 0, len(sessions))
|
||||
for sessionID := range sessions {
|
||||
sessionIDs = append(sessionIDs, sessionID)
|
||||
}
|
||||
return sessionIDs
|
||||
}
|
||||
|
||||
func processIDsByName(name string) ([]uint32, error) {
|
||||
snapshot, err := windows.CreateToolhelp32Snapshot(windows.TH32CS_SNAPPROCESS, 0)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create process snapshot: %w", err)
|
||||
}
|
||||
defer func() {
|
||||
if err := windows.CloseHandle(snapshot); err != nil {
|
||||
log.Warnf("failed to close process snapshot: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
var entry windows.ProcessEntry32
|
||||
entry.Size = uint32(unsafe.Sizeof(entry))
|
||||
|
||||
var pids []uint32
|
||||
for err = windows.Process32First(snapshot, &entry); err == nil; err = windows.Process32Next(snapshot, &entry) {
|
||||
if strings.EqualFold(windows.UTF16ToString(entry.ExeFile[:]), name) {
|
||||
pids = append(pids, entry.ProcessID)
|
||||
}
|
||||
}
|
||||
if !errors.Is(err, windows.ERROR_NO_MORE_FILES) {
|
||||
return nil, fmt.Errorf("enumerate processes: %w", err)
|
||||
}
|
||||
|
||||
return pids, nil
|
||||
}
|
||||
|
||||
func terminateProcess(pid uint32) error {
|
||||
handle, err := windows.OpenProcess(windows.PROCESS_TERMINATE|windows.SYNCHRONIZE, false, pid)
|
||||
if err != nil {
|
||||
// The process may have exited between enumeration and now.
|
||||
if errors.Is(err, windows.ERROR_INVALID_PARAMETER) {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("open process: %w", err)
|
||||
}
|
||||
defer func() {
|
||||
if err := windows.CloseHandle(handle); err != nil {
|
||||
log.Warnf("failed to close process handle: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
if err := windows.TerminateProcess(handle, 0); err != nil {
|
||||
return fmt.Errorf("terminate process: %w", err)
|
||||
}
|
||||
|
||||
// Wait for the handle to signal so the image file is released before the
|
||||
// installer tries to overwrite it. A timeout is reported through the returned
|
||||
// event, not through err, which stays nil unless the wait itself failed.
|
||||
event, err := windows.WaitForSingleObject(handle, uint32(processExitWait.Milliseconds()))
|
||||
if err != nil {
|
||||
return fmt.Errorf("wait for process exit: %w", err)
|
||||
}
|
||||
if event != windows.WAIT_OBJECT_0 {
|
||||
return fmt.Errorf("wait for process exit: unexpected wait result %#x", event)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func startUIInSession(uiPath string, sessionID uint32) error {
|
||||
// Get the user token for that session
|
||||
var userToken windows.Token
|
||||
err := windows.WTSQueryUserToken(sessionID, &userToken)
|
||||
@@ -307,16 +158,6 @@ func startUIInSession(uiPath string, sessionID uint32) error {
|
||||
}
|
||||
}()
|
||||
|
||||
var env *uint16
|
||||
if err := windows.CreateEnvironmentBlock(&env, primaryToken, false); err != nil {
|
||||
return fmt.Errorf("create environment block: %w", err)
|
||||
}
|
||||
defer func() {
|
||||
if err := windows.DestroyEnvironmentBlock(env); err != nil {
|
||||
log.Warnf("failed to destroy environment block: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
// Prepare startup info
|
||||
var si windows.StartupInfo
|
||||
si.Cb = uint32(unsafe.Sizeof(si))
|
||||
@@ -339,7 +180,7 @@ func startUIInSession(uiPath string, sessionID uint32) error {
|
||||
nil,
|
||||
false,
|
||||
creationFlags,
|
||||
env,
|
||||
nil,
|
||||
nil,
|
||||
&si,
|
||||
&pi,
|
||||
@@ -356,6 +197,7 @@ func startUIInSession(uiPath string, sessionID uint32) error {
|
||||
log.Warnf("failed to close thread handle: %v", err)
|
||||
}
|
||||
|
||||
log.Infof("netbird-ui started successfully in session %d", sessionID)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -1,108 +0,0 @@
|
||||
package installer
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"os/exec"
|
||||
"slices"
|
||||
"strconv"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// exitErrorWithCode returns a real *exec.ExitError carrying the given exit code.
|
||||
func exitErrorWithCode(t *testing.T, code int) error {
|
||||
t.Helper()
|
||||
|
||||
err := exec.Command("cmd.exe", "/c", "exit "+strconv.Itoa(code)).Run()
|
||||
if err == nil {
|
||||
t.Fatalf("expected a non-zero exit for code %d", code)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func TestIsRebootPending(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
code int
|
||||
want bool
|
||||
}{
|
||||
{name: "reboot required", code: msiRebootRequired, want: true},
|
||||
{name: "reboot initiated", code: msiRebootInitiated, want: true},
|
||||
{name: "generic failure", code: 1603, want: false},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := isRebootPending(exitErrorWithCode(t, tt.code)); got != tt.want {
|
||||
t.Errorf("isRebootPending(exit %d) = %v, want %v", tt.code, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestProcessIDsByNameAndTerminate spawns a long-running system process, finds it
|
||||
// by name and terminates it, covering the path the updater uses to release the UI
|
||||
// image file before the installer replaces it.
|
||||
func TestProcessIDsByNameAndTerminate(t *testing.T) {
|
||||
cmd := exec.Command("ping.exe", "-n", "60", "127.0.0.1")
|
||||
if err := cmd.Start(); err != nil {
|
||||
t.Fatalf("start ping: %v", err)
|
||||
}
|
||||
|
||||
pid := uint32(cmd.Process.Pid)
|
||||
killed := false
|
||||
t.Cleanup(func() {
|
||||
if !killed {
|
||||
_ = cmd.Process.Kill()
|
||||
}
|
||||
_ = cmd.Wait()
|
||||
})
|
||||
|
||||
// Name matching must be case-insensitive: the snapshot reports PING.EXE.
|
||||
pids, err := processIDsByName("ping.exe")
|
||||
if err != nil {
|
||||
t.Fatalf("processIDsByName: %v", err)
|
||||
}
|
||||
|
||||
if !slices.Contains(pids, pid) {
|
||||
t.Fatalf("PID %d not among the ping.exe processes found: %v", pid, pids)
|
||||
}
|
||||
|
||||
if err := terminateProcess(pid); err != nil {
|
||||
t.Fatalf("terminateProcess: %v", err)
|
||||
}
|
||||
killed = true
|
||||
|
||||
// terminateProcess only returns once the handle has signalled, so the process
|
||||
// is already gone and Wait must not block. It exits with the code passed to
|
||||
// TerminateProcess, which is 0, so Wait reports no error.
|
||||
if err := cmd.Wait(); err != nil {
|
||||
t.Fatalf("wait for terminated ping: %v", err)
|
||||
}
|
||||
if !cmd.ProcessState.Exited() {
|
||||
t.Error("process did not exit after terminateProcess")
|
||||
}
|
||||
|
||||
remaining, err := processIDsByName("ping.exe")
|
||||
if err != nil {
|
||||
t.Fatalf("processIDsByName after terminate: %v", err)
|
||||
}
|
||||
if slices.Contains(remaining, pid) {
|
||||
t.Errorf("PID %d still listed after terminateProcess", pid)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProcessIDsByNameNoMatch(t *testing.T) {
|
||||
pids, err := processIDsByName("netbird-nonexistent-process.exe")
|
||||
if err != nil {
|
||||
t.Fatalf("processIDsByName: %v", err)
|
||||
}
|
||||
if len(pids) != 0 {
|
||||
t.Errorf("expected no matches, got %v", pids)
|
||||
}
|
||||
}
|
||||
|
||||
func TestIsRebootPendingNonExitError(t *testing.T) {
|
||||
if isRebootPending(errors.New("start installer: file not found")) {
|
||||
t.Error("a non-exit error must not be treated as a pending reboot")
|
||||
}
|
||||
}
|
||||
@@ -54,12 +54,6 @@ func (rh *ResultHandler) GetErrorResultReason() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
// ClearStaleResult removes a result file left over from a previous installation
|
||||
// attempt so result watchers cannot read an outdated outcome for the current attempt.
|
||||
func (rh *ResultHandler) ClearStaleResult() error {
|
||||
return rh.cleanup()
|
||||
}
|
||||
|
||||
func (rh *ResultHandler) WriteSuccess() error {
|
||||
result := Result{
|
||||
Success: true,
|
||||
|
||||
@@ -435,7 +435,7 @@ func (m *Manager) install(ctx context.Context, pendingVersion *v.Version) error
|
||||
}
|
||||
|
||||
inst := installer.New()
|
||||
if err := inst.RunInstallation(ctx, pendingVersion.String()); err != nil { //nolint:staticcheck // always errors on platforms without an installer
|
||||
if err := inst.RunInstallation(ctx, pendingVersion.String()); err != nil {
|
||||
log.Errorf("error triggering update: %v", err)
|
||||
m.statusRecorder.PublishEvent(
|
||||
cProto.SystemEvent_ERROR,
|
||||
|
||||
@@ -22,8 +22,6 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/listener"
|
||||
"github.com/netbirdio/netbird/client/internal/peer"
|
||||
"github.com/netbirdio/netbird/client/internal/profilemanager"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
"github.com/netbirdio/netbird/client/system"
|
||||
"github.com/netbirdio/netbird/formatter"
|
||||
"github.com/netbirdio/netbird/route"
|
||||
@@ -38,6 +36,11 @@ const (
|
||||
AnonymizeLevelStrict = nbAnonymize.LevelStrictString
|
||||
)
|
||||
|
||||
// ConnectionListener export internal Listener for mobile
|
||||
type ConnectionListener interface {
|
||||
peer.Listener
|
||||
}
|
||||
|
||||
// RouteListener export internal RouteListener for mobile
|
||||
type NetworkChangeListener interface {
|
||||
listener.NetworkChangeListener
|
||||
@@ -84,12 +87,6 @@ type Client struct {
|
||||
onHostDnsFn func([]string)
|
||||
dnsManager dns.IosDnsManager
|
||||
loginComplete bool
|
||||
// netState outlives engine restarts: it mirrors the OS connectivity, not
|
||||
// the engine lifecycle. Run injects it into each new ConnectClient, which
|
||||
// distributes it to every reconnection loop.
|
||||
netState *netstate.State
|
||||
// sweeper also outlives engine restarts; NotifyNetworkChange sweeps it.
|
||||
sweeper *netsweep.Sweeper
|
||||
// preloadedConfig holds config loaded from JSON (used on tvOS where file writes are blocked)
|
||||
preloadedConfig *profilemanager.Config
|
||||
|
||||
@@ -112,8 +109,6 @@ func NewClient(cfgFile, stateFile, cacheDir, logFilePath, deviceName string, osV
|
||||
ctxCancelLock: &sync.Mutex{},
|
||||
networkChangeListener: networkChangeListener,
|
||||
dnsManager: dnsManager,
|
||||
netState: netstate.New(),
|
||||
sweeper: netsweep.New(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -189,8 +184,7 @@ func (c *Client) Run(fd int32, interfaceName string, envList *EnvList) error {
|
||||
c.onHostDnsFn = func([]string) {}
|
||||
cfg.WgIface = interfaceName
|
||||
|
||||
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder,
|
||||
internal.WithNetworkState(c.netState), internal.WithSweeper(c.sweeper))
|
||||
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder)
|
||||
c.setState(cfg, connectClient)
|
||||
// Persist the latest sync response so DebugBundle can include the network
|
||||
// map. On iOS this is backed by disk to keep it out of the constrained
|
||||
@@ -199,25 +193,6 @@ func (c *Client) Run(fd int32, interfaceName string, envList *EnvList) error {
|
||||
return connectClient.RunOniOS(fd, c.networkChangeListener, c.dnsManager, c.stateFile, c.cacheDir, c.logFilePath)
|
||||
}
|
||||
|
||||
// SetNetworkAvailable feeds OS-reported network availability into the client
|
||||
// (e.g. from NWPathMonitor). While unavailable, the internal reconnect loops
|
||||
// suspend their attempts and the connection listener reports NoNetwork
|
||||
// instead of Connecting; when availability returns, the loops resume
|
||||
// immediately with a fresh backoff.
|
||||
func (c *Client) SetNetworkAvailable(available bool) {
|
||||
c.netState.Set(available)
|
||||
c.recorder.SetNetworkAvailable(available)
|
||||
}
|
||||
|
||||
// NotifyNetworkChange marks the management, signal and relay connections
|
||||
// stale after the OS switched networks and schedules a sweep that cuts
|
||||
// whatever has not redialed on the new network by then. The engine and the
|
||||
// TUN device stay untouched.
|
||||
func (c *Client) NotifyNetworkChange() {
|
||||
c.sweeper.MarkNetworkChange()
|
||||
log.Infof("network change: connections marked stale")
|
||||
}
|
||||
|
||||
// Stop the internal client and free the resources
|
||||
func (c *Client) Stop() {
|
||||
c.ctxCancelLock.Lock()
|
||||
@@ -356,11 +331,7 @@ func (c *Client) GetStatusDetails() *StatusDetails {
|
||||
|
||||
// SetConnectionListener set the network connection listener
|
||||
func (c *Client) SetConnectionListener(listener ConnectionListener) {
|
||||
if listener == nil {
|
||||
c.recorder.RemoveConnectionListener()
|
||||
return
|
||||
}
|
||||
c.recorder.SetConnectionListener(connectionListenerAdapter{listener})
|
||||
c.recorder.SetConnectionListener(listener)
|
||||
}
|
||||
|
||||
// RemoveConnectionListener remove connection listener
|
||||
|
||||
@@ -1,43 +0,0 @@
|
||||
//go:build ios
|
||||
|
||||
package NetBirdSDK
|
||||
|
||||
import (
|
||||
"github.com/netbirdio/netbird/client/internal/peer"
|
||||
)
|
||||
|
||||
// Client state values, re-exported as basic constants so gomobile emits them
|
||||
// into the generated bindings. They mirror peer.ClientState*: append-only,
|
||||
// never reorder.
|
||||
const (
|
||||
ClientStateDisconnected = int(peer.ClientStateDisconnected)
|
||||
ClientStateConnected = int(peer.ClientStateConnected)
|
||||
ClientStateConnecting = int(peer.ClientStateConnecting)
|
||||
ClientStateDisconnecting = int(peer.ClientStateDisconnecting)
|
||||
ClientStateNoNetwork = int(peer.ClientStateNoNetwork)
|
||||
)
|
||||
|
||||
// ConnectionListener export internal Listener for mobile.
|
||||
//
|
||||
// It intentionally lacks OnStateChanged for now: adding a method to a gomobile
|
||||
// interface breaks every Swift implementation, so the iOS app keeps building
|
||||
// against the legacy per-state callbacks. A follow-up will extend it together
|
||||
// with the app.
|
||||
type ConnectionListener interface {
|
||||
OnConnected()
|
||||
OnDisconnected()
|
||||
OnConnecting()
|
||||
OnDisconnecting()
|
||||
OnAddressChanged(string, string)
|
||||
OnPeersListChanged(int)
|
||||
}
|
||||
|
||||
// connectionListenerAdapter adapts the gomobile-facing ConnectionListener to
|
||||
// peer.Listener.
|
||||
type connectionListenerAdapter struct {
|
||||
ConnectionListener
|
||||
}
|
||||
|
||||
// OnStateChanged is dropped on iOS until the app adopts the state callback;
|
||||
// the legacy per-state callbacks continue to fire.
|
||||
func (a connectionListenerAdapter) OnStateChanged(peer.ClientState) {}
|
||||
@@ -323,7 +323,7 @@ func (a *Auth) login(urlOpener URLOpener, forceDeviceAuth bool, deviceName strin
|
||||
const authInfoRequestTimeout = 30 * time.Second
|
||||
|
||||
func (a *Auth) foregroundGetTokenInfo(authClient *auth.Auth, urlOpener URLOpener, forceDeviceAuth bool) (*auth.TokenInfo, error) {
|
||||
oAuthFlow, err := authClient.GetOAuthFlow(a.ctx, forceDeviceAuth, "")
|
||||
oAuthFlow, err := authClient.GetOAuthFlow(a.ctx, forceDeviceAuth)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get OAuth flow: %v", err)
|
||||
}
|
||||
|
||||
@@ -1,110 +0,0 @@
|
||||
// Package netstate tracks OS-reported network availability for the client.
|
||||
//
|
||||
// A State instance is owned by the platform integration (e.g. the Android or
|
||||
// iOS bindings, fed from ConnectivityManager callbacks or NWPathMonitor) and
|
||||
// is injected into the connection retry loops (management, signal, relay,
|
||||
// peer guards and the top-level connect loop), which consult it to avoid
|
||||
// burning CPU and battery on reconnect attempts while the device has no
|
||||
// network at all (e.g. airplane mode), and to reset their backoff as soon as
|
||||
// the network returns.
|
||||
//
|
||||
// Consumers hold a *State that may be nil — every non-mobile platform leaves
|
||||
// it unset. The read methods are safe on a nil receiver: they report online
|
||||
// and never block, so consumers behave as if this package did not exist.
|
||||
package netstate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// State holds the OS-reported network availability. The zero value is not
|
||||
// usable; create instances with New.
|
||||
type State struct {
|
||||
mu sync.Mutex
|
||||
online bool
|
||||
changed chan struct{}
|
||||
}
|
||||
|
||||
// New creates a State that starts online. Platforms without network tracking
|
||||
// pass a nil *State instead: the read methods treat nil as always online and
|
||||
// never block, so consumers need no nil guards.
|
||||
func New() *State {
|
||||
return &State{
|
||||
online: true,
|
||||
changed: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// Set records whether the OS reports any usable network. Transitions wake up
|
||||
// all Wait callers immediately. Unlike the read methods, Set is not nil-safe:
|
||||
// it is only for the platform owner that created the State with New.
|
||||
func (s *State) Set(online bool) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.online == online {
|
||||
return
|
||||
}
|
||||
s.online = online
|
||||
close(s.changed)
|
||||
s.changed = make(chan struct{})
|
||||
log.Infof("OS network availability changed: online=%t", online)
|
||||
}
|
||||
|
||||
// IsOnline reports whether the OS reports at least one usable network. On a
|
||||
// nil receiver — no State injected — it reports online.
|
||||
func (s *State) IsOnline() bool {
|
||||
if s == nil {
|
||||
return true
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.online
|
||||
}
|
||||
|
||||
// Changed returns a channel closed on the next availability transition, for
|
||||
// callers that already own a select loop and cannot block in Wait. Re-read it
|
||||
// after every fire: each transition installs a fresh channel. On a nil
|
||||
// receiver — no State injected — it returns nil, which blocks forever in a
|
||||
// select, so the caller simply never observes a transition.
|
||||
func (s *State) Changed() <-chan struct{} {
|
||||
if s == nil {
|
||||
return nil
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.changed
|
||||
}
|
||||
|
||||
// Wait blocks while the network is offline. It reports whether it had to
|
||||
// wait, so callers can reset their backoff after an outage. It returns early
|
||||
// with the context error when ctx is done. On a nil receiver — no State
|
||||
// injected — it returns immediately.
|
||||
func (s *State) Wait(ctx context.Context) (bool, error) {
|
||||
if s == nil {
|
||||
return false, nil
|
||||
}
|
||||
waited := false
|
||||
for {
|
||||
s.mu.Lock()
|
||||
if s.online {
|
||||
s.mu.Unlock()
|
||||
return waited, nil
|
||||
}
|
||||
ch := s.changed
|
||||
s.mu.Unlock()
|
||||
|
||||
if !waited {
|
||||
waited = true
|
||||
log.Debugf("network is offline, pausing connection attempts")
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return waited, ctx.Err()
|
||||
case <-ch:
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,170 +0,0 @@
|
||||
package netstate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestNewStateIsOnline(t *testing.T) {
|
||||
assert.True(t, New().IsOnline(), "a fresh State should start online")
|
||||
}
|
||||
|
||||
func TestSetTogglesOnlineState(t *testing.T) {
|
||||
s := New()
|
||||
|
||||
s.Set(false)
|
||||
assert.False(t, s.IsOnline(), "state should be offline after Set(false)")
|
||||
|
||||
s.Set(true)
|
||||
assert.True(t, s.IsOnline(), "state should be online after Set(true)")
|
||||
}
|
||||
|
||||
func TestWaitReturnsImmediatelyWhenOnline(t *testing.T) {
|
||||
s := New()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
|
||||
waited, err := s.Wait(ctx)
|
||||
require.NoError(t, err)
|
||||
assert.False(t, waited, "Wait should not block when the network is online")
|
||||
}
|
||||
|
||||
func TestWaitBlocksUntilOnline(t *testing.T) {
|
||||
s := New()
|
||||
s.Set(false)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
result := make(chan bool, 1)
|
||||
go func() {
|
||||
waited, err := s.Wait(ctx)
|
||||
if err != nil {
|
||||
result <- false
|
||||
return
|
||||
}
|
||||
result <- waited
|
||||
}()
|
||||
|
||||
// Verify Wait is actually blocking while offline
|
||||
select {
|
||||
case <-result:
|
||||
t.Fatal("Wait should block while the network is offline")
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
}
|
||||
|
||||
s.Set(true)
|
||||
|
||||
select {
|
||||
case waited := <-result:
|
||||
assert.True(t, waited, "Wait should report that it had to wait for the network")
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("Wait should return promptly after the network becomes available")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWaitReturnsOnContextCancel(t *testing.T) {
|
||||
s := New()
|
||||
s.Set(false)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
result := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := s.Wait(ctx)
|
||||
result <- err
|
||||
}()
|
||||
|
||||
cancel()
|
||||
|
||||
select {
|
||||
case err := <-result:
|
||||
assert.ErrorIs(t, err, context.Canceled)
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("Wait should return promptly after context cancellation")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWaitWakesAllWaiters(t *testing.T) {
|
||||
s := New()
|
||||
s.Set(false)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
const waiters = 10
|
||||
var wg sync.WaitGroup
|
||||
results := make(chan bool, waiters)
|
||||
for i := 0; i < waiters; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
waited, err := s.Wait(ctx)
|
||||
if err != nil {
|
||||
results <- false
|
||||
return
|
||||
}
|
||||
results <- waited
|
||||
}()
|
||||
}
|
||||
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
s.Set(true)
|
||||
wg.Wait()
|
||||
|
||||
close(results)
|
||||
count := 0
|
||||
for waited := range results {
|
||||
assert.True(t, waited, "every waiter should report that it waited")
|
||||
count++
|
||||
}
|
||||
assert.Equal(t, waiters, count, "all waiters should have returned")
|
||||
}
|
||||
|
||||
func TestNilStateReadsAreNoops(t *testing.T) {
|
||||
var s *State
|
||||
|
||||
assert.True(t, s.IsOnline(), "nil State should report online")
|
||||
|
||||
waited, err := s.Wait(context.Background())
|
||||
require.NoError(t, err)
|
||||
assert.False(t, waited, "nil State's Wait should not block")
|
||||
}
|
||||
|
||||
func TestConcurrentSetAndWait(t *testing.T) {
|
||||
s := New()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 4; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for j := 0; j < 100; j++ {
|
||||
s.Set(j%2 == 0)
|
||||
s.IsOnline()
|
||||
}
|
||||
}()
|
||||
}
|
||||
for i := 0; i < 4; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for j := 0; j < 100; j++ {
|
||||
if _, err := s.Wait(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
}
|
||||
@@ -1,267 +0,0 @@
|
||||
// Package netsweep cuts network-bound activity when the OS switches networks:
|
||||
// a sweep closes the registered connections and aborts the in-flight dials, so
|
||||
// their owners redial immediately instead of waiting for the old sockets to
|
||||
// time out.
|
||||
//
|
||||
// A nil *Sweeper disables everything: all methods are nil-safe no-ops.
|
||||
package netsweep
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
// DefaultSweepDelay absorbs network flapping while the OS settles on a
|
||||
// default network before the stale registrations are cut.
|
||||
const DefaultSweepDelay = 500 * time.Millisecond
|
||||
|
||||
const recentMarkWindow = 3 * time.Second
|
||||
|
||||
// Config customizes a Sweeper. The zero value applies the defaults.
|
||||
type Config struct {
|
||||
// SweepDelay overrides DefaultSweepDelay when positive.
|
||||
SweepDelay time.Duration
|
||||
}
|
||||
|
||||
// ErrSwept reports that a dial finished after a network change swept its
|
||||
// registration. The connection is already closed; the caller must treat it
|
||||
// as a failed dial and redial on the new network.
|
||||
var ErrSwept = errors.New("netsweep: connection swept by network change")
|
||||
|
||||
// sweepID identifies one registration in a sweeper. Connections and dials
|
||||
// draw from the same counter, so an id is unique across both registries.
|
||||
type sweepID uint64
|
||||
|
||||
type connEntry struct {
|
||||
conn net.Conn
|
||||
gen uint64
|
||||
}
|
||||
|
||||
// Dial tracks one dial from start to connection registration. It hands the
|
||||
// dialed connection to the sweeper atomically, so a sweep can never fall
|
||||
// between the dial finishing and the connection being registered.
|
||||
type Dial struct {
|
||||
sweeper *Sweeper
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
id sweepID
|
||||
done bool // set by a sweep, WrapConn or Release; guarded by sweeper.mu
|
||||
gen uint64
|
||||
}
|
||||
|
||||
// Ctx returns the dial's context. A sweep cancels it, so a dial started on the
|
||||
// old network aborts instead of waiting out its handshake timeout.
|
||||
func (d *Dial) Ctx() context.Context {
|
||||
return d.ctx
|
||||
}
|
||||
|
||||
// Release ends the dial's registration and cancels its context. It is
|
||||
// idempotent and safe after WrapConn, so callers can defer it.
|
||||
func (d *Dial) Release() {
|
||||
s := d.sweeper
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
d.done = true
|
||||
delete(s.dials, d.id)
|
||||
s.mu.Unlock()
|
||||
|
||||
d.cancel()
|
||||
}
|
||||
|
||||
// sweptConn deregisters itself from the sweeper when closed.
|
||||
type sweptConn struct {
|
||||
net.Conn
|
||||
sweeper *Sweeper
|
||||
id sweepID
|
||||
}
|
||||
|
||||
func (c *sweptConn) Close() error {
|
||||
c.sweeper.deregister(c.id)
|
||||
return c.Conn.Close()
|
||||
}
|
||||
|
||||
// Sweeper registers live connections and in-flight dials so the
|
||||
// network-change sweep can cut everything registered before the change.
|
||||
type Sweeper struct {
|
||||
mu sync.Mutex
|
||||
conns map[sweepID]connEntry
|
||||
dials map[sweepID]*Dial
|
||||
nextID sweepID
|
||||
gen uint64
|
||||
timer *time.Timer
|
||||
sweepDelay time.Duration
|
||||
lastMark time.Time
|
||||
}
|
||||
|
||||
// New creates an empty sweeper with the default configuration.
|
||||
func New() *Sweeper {
|
||||
return NewWithConfig(Config{})
|
||||
}
|
||||
|
||||
// NewWithConfig creates an empty sweeper customized by cfg.
|
||||
func NewWithConfig(cfg Config) *Sweeper {
|
||||
delay := cfg.SweepDelay
|
||||
if delay <= 0 {
|
||||
delay = DefaultSweepDelay
|
||||
}
|
||||
return &Sweeper{
|
||||
conns: make(map[sweepID]connEntry),
|
||||
dials: make(map[sweepID]*Dial),
|
||||
sweepDelay: delay,
|
||||
}
|
||||
}
|
||||
|
||||
// StartDial registers an in-flight dial. Dial with Ctx, hand the result to
|
||||
// WrapConn, and Release the dial when the attempt is over, typically deferred.
|
||||
func (s *Sweeper) StartDial(ctx context.Context) *Dial {
|
||||
if s == nil {
|
||||
return &Dial{ctx: ctx}
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
d := &Dial{sweeper: s, ctx: ctx, cancel: cancel}
|
||||
|
||||
s.mu.Lock()
|
||||
d.id = s.nextID
|
||||
s.nextID++
|
||||
d.gen = s.gen
|
||||
s.dials[d.id] = d
|
||||
s.mu.Unlock()
|
||||
|
||||
return d
|
||||
}
|
||||
|
||||
// WrapConn hands conn over to the sweeper. If a sweep ran since StartDial,
|
||||
// the connection belongs to the old network: it is closed and ErrSwept is
|
||||
// returned. Otherwise conn is registered against the next sweep and returned
|
||||
// wrapped, deregistering itself on Close. Call it once, before Release.
|
||||
func (d *Dial) WrapConn(conn net.Conn) (net.Conn, error) {
|
||||
s := d.sweeper
|
||||
if s == nil {
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
if d.done {
|
||||
s.mu.Unlock()
|
||||
if err := conn.Close(); err != nil {
|
||||
log.Debugf("swept dial close error: %v", err)
|
||||
}
|
||||
return nil, ErrSwept
|
||||
}
|
||||
d.done = true
|
||||
delete(s.dials, d.id)
|
||||
id := s.nextID
|
||||
s.nextID++
|
||||
// The conn inherits the dial's generation: the socket was bound to the
|
||||
// network that was default when the dial started, not when it finished.
|
||||
s.conns[id] = connEntry{conn: conn, gen: d.gen}
|
||||
s.mu.Unlock()
|
||||
|
||||
return &sweptConn{Conn: conn, sweeper: s, id: id}, nil
|
||||
}
|
||||
|
||||
// MarkNetworkChange records that the OS switched networks: everything
|
||||
// registered so far becomes stale, and a sweep is (re)scheduled after the
|
||||
// configured delay to cut whatever is still stale by then. Owners that
|
||||
// redialed in the meantime hold fresh-generation registrations and survive,
|
||||
// so no cancellation is needed around the sweep.
|
||||
func (s *Sweeper) MarkNetworkChange() {
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
s.gen++
|
||||
cutoff := s.gen
|
||||
s.lastMark = time.Now()
|
||||
if s.timer != nil {
|
||||
s.timer.Stop()
|
||||
}
|
||||
s.timer = time.AfterFunc(s.sweepDelay, func() {
|
||||
n := s.sweep(cutoff)
|
||||
log.Infof("network change sweep: closed %d stale connections", n)
|
||||
})
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// QuickRetryBackoff wraps bo so that after each Reset the first retry comes
|
||||
// quickly when the disconnect followed a recent network change and the
|
||||
// network is online. Any other failure keeps bo's spread, so the clients of
|
||||
// a restarted server still scatter their reconnects. A nil sweeper returns
|
||||
// bo unchanged.
|
||||
func (s *Sweeper) QuickRetryBackoff(ctx context.Context, bo backoff.BackOff, netState *netstate.State) backoff.BackOff {
|
||||
if s == nil {
|
||||
return bo
|
||||
}
|
||||
return backoff.WithContext(newQuickRetryBackoff(bo, s, netState), ctx)
|
||||
}
|
||||
|
||||
func (s *Sweeper) markedRecently() bool {
|
||||
if s == nil {
|
||||
return false
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return !s.lastMark.IsZero() && time.Since(s.lastMark) < recentMarkWindow
|
||||
}
|
||||
|
||||
// sweep closes the registered connections and aborts the in-flight dials
|
||||
// older than cutoff, and returns how many connections it closed. A dial
|
||||
// whose connection was not yet handed to WrapConn is marked, so the late
|
||||
// WrapConn closes it instead of registering it.
|
||||
func (s *Sweeper) sweep(cutoff uint64) int {
|
||||
if s == nil {
|
||||
return 0
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
var conns []net.Conn
|
||||
for id, e := range s.conns {
|
||||
if e.gen < cutoff {
|
||||
delete(s.conns, id)
|
||||
conns = append(conns, e.conn)
|
||||
}
|
||||
}
|
||||
var dials []*Dial
|
||||
for id, d := range s.dials {
|
||||
if d.gen < cutoff {
|
||||
d.done = true
|
||||
delete(s.dials, id)
|
||||
dials = append(dials, d)
|
||||
}
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
if len(dials) > 0 {
|
||||
log.Debugf("aborting %d in-flight dials", len(dials))
|
||||
for _, d := range dials {
|
||||
d.cancel()
|
||||
}
|
||||
}
|
||||
|
||||
for _, conn := range conns {
|
||||
log.Debugf("sweeping connection %s -> %s", conn.LocalAddr(), conn.RemoteAddr())
|
||||
if err := conn.Close(); err != nil {
|
||||
log.Debugf("swept connection close error: %v", err)
|
||||
}
|
||||
}
|
||||
return len(conns)
|
||||
}
|
||||
|
||||
func (s *Sweeper) deregister(id sweepID) {
|
||||
s.mu.Lock()
|
||||
delete(s.conns, id)
|
||||
s.mu.Unlock()
|
||||
}
|
||||
@@ -1,241 +0,0 @@
|
||||
package netsweep
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math"
|
||||
"net"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestSweepClosesRegisteredConns(t *testing.T) {
|
||||
sweeper := New()
|
||||
|
||||
c1 := wrap(t, sweeper, connPair(t))
|
||||
c2 := wrap(t, sweeper, connPair(t))
|
||||
|
||||
assert.Equal(t, 2, sweeper.sweepAll(), "both live connections should be closed")
|
||||
|
||||
// The wrappers must report closed now.
|
||||
buf := make([]byte, 1)
|
||||
_, err := c1.Read(buf)
|
||||
assert.Error(t, err, "first connection should be unusable after the sweep")
|
||||
_, err = c2.Read(buf)
|
||||
assert.Error(t, err, "second connection should be unusable after the sweep")
|
||||
|
||||
assert.Equal(t, 0, sweeper.sweepAll(), "second sweep should find nothing")
|
||||
}
|
||||
|
||||
func TestCloseDeregisters(t *testing.T) {
|
||||
sweeper := New()
|
||||
|
||||
conn := wrap(t, sweeper, connPair(t))
|
||||
require.NoError(t, conn.Close())
|
||||
|
||||
assert.Equal(t, 0, sweeper.sweepAll(), "closed connection must leave the registry")
|
||||
}
|
||||
|
||||
func TestCloseIsIdempotent(t *testing.T) {
|
||||
sweeper := New()
|
||||
|
||||
conn := wrap(t, sweeper, connPair(t))
|
||||
require.NoError(t, conn.Close())
|
||||
assert.Error(t, conn.Close(), "double close surfaces the underlying error but must not panic")
|
||||
}
|
||||
|
||||
func TestSweepOnlyAffectsOlderConns(t *testing.T) {
|
||||
sweeper := New()
|
||||
|
||||
_ = wrap(t, sweeper, connPair(t))
|
||||
assert.Equal(t, 1, sweeper.sweepAll())
|
||||
|
||||
// A connection dialed after the sweep must survive until the next one.
|
||||
_ = wrap(t, sweeper, connPair(t))
|
||||
assert.Equal(t, 1, sweeper.sweepAll(), "post-sweep connection belongs to the next sweep")
|
||||
}
|
||||
|
||||
func TestSweepAbortsInFlightDials(t *testing.T) {
|
||||
sweeper := New()
|
||||
|
||||
dial := sweeper.StartDial(context.Background())
|
||||
defer dial.Release()
|
||||
|
||||
sweeper.sweepAll()
|
||||
|
||||
assert.ErrorIs(t, dial.Ctx().Err(), context.Canceled, "sweep must cancel the in-flight dial context")
|
||||
}
|
||||
|
||||
func TestReleasedDialIsNotAborted(t *testing.T) {
|
||||
sweeper := New()
|
||||
|
||||
// Simulate a dial that finished before the sweep.
|
||||
released := sweeper.StartDial(context.Background())
|
||||
released.Release()
|
||||
|
||||
// A dial still in flight during the sweep.
|
||||
pending := sweeper.StartDial(context.Background())
|
||||
defer pending.Release()
|
||||
|
||||
sweeper.sweepAll()
|
||||
assert.ErrorIs(t, pending.Ctx().Err(), context.Canceled, "pending dial must be aborted")
|
||||
}
|
||||
|
||||
func TestSweepBetweenDialAndHandoffClosesConn(t *testing.T) {
|
||||
sweeper := New()
|
||||
|
||||
dial := sweeper.StartDial(context.Background())
|
||||
defer dial.Release()
|
||||
|
||||
// The dial succeeds on the old network, then the sweep lands before the
|
||||
// connection is handed over.
|
||||
conn := connPair(t)
|
||||
assert.Equal(t, 0, sweeper.sweepAll(), "the connection is not registered yet")
|
||||
|
||||
wrapped, err := dial.WrapConn(conn)
|
||||
require.ErrorIs(t, err, ErrSwept)
|
||||
require.Nil(t, wrapped)
|
||||
|
||||
buf := make([]byte, 1)
|
||||
_, err = conn.Read(buf)
|
||||
assert.Error(t, err, "the old-network connection must be closed, not leaked")
|
||||
|
||||
assert.Equal(t, 0, sweeper.sweepAll(), "nothing may leak into the next sweep")
|
||||
}
|
||||
|
||||
func TestMarkNetworkChangeSparesFreshConns(t *testing.T) {
|
||||
sweeper := NewWithConfig(Config{SweepDelay: 10 * time.Millisecond})
|
||||
|
||||
stale := wrap(t, sweeper, connPair(t))
|
||||
sweeper.MarkNetworkChange()
|
||||
_ = wrap(t, sweeper, connPair(t))
|
||||
|
||||
_ = stale.SetReadDeadline(time.Now().Add(time.Second))
|
||||
buf := make([]byte, 1)
|
||||
_, err := stale.Read(buf)
|
||||
require.ErrorIs(t, err, net.ErrClosed, "stale connection must be closed by the delayed sweep")
|
||||
|
||||
assert.Equal(t, 1, sweeper.sweepAll(), "the fresh connection must survive the stale sweep")
|
||||
}
|
||||
|
||||
func TestMarkNetworkChangeAbortsStaleDials(t *testing.T) {
|
||||
sweeper := NewWithConfig(Config{SweepDelay: 10 * time.Millisecond})
|
||||
|
||||
stale := sweeper.StartDial(context.Background())
|
||||
defer stale.Release()
|
||||
sweeper.MarkNetworkChange()
|
||||
fresh := sweeper.StartDial(context.Background())
|
||||
defer fresh.Release()
|
||||
|
||||
assert.Eventually(t, func() bool {
|
||||
return stale.Ctx().Err() != nil
|
||||
}, time.Second, 5*time.Millisecond, "stale dial must be aborted by the delayed sweep")
|
||||
assert.NoError(t, fresh.Ctx().Err(), "post-mark dial must not be aborted")
|
||||
}
|
||||
|
||||
func TestConnInheritsDialGeneration(t *testing.T) {
|
||||
sweeper := NewWithConfig(Config{SweepDelay: 20 * time.Millisecond})
|
||||
|
||||
// The dial starts before the network change but completes after it: the
|
||||
// socket is bound to the old network, so the sweep must still cut it.
|
||||
dial := sweeper.StartDial(context.Background())
|
||||
defer dial.Release()
|
||||
sweeper.MarkNetworkChange()
|
||||
|
||||
wrapped, err := dial.WrapConn(connPair(t))
|
||||
require.NoError(t, err)
|
||||
|
||||
_ = wrapped.SetReadDeadline(time.Now().Add(time.Second))
|
||||
buf := make([]byte, 1)
|
||||
_, err = wrapped.Read(buf)
|
||||
require.ErrorIs(t, err, net.ErrClosed, "old-generation connection must be swept")
|
||||
}
|
||||
|
||||
func TestRepeatedMarksCoalesce(t *testing.T) {
|
||||
sweeper := NewWithConfig(Config{SweepDelay: 20 * time.Millisecond})
|
||||
|
||||
first := wrap(t, sweeper, connPair(t))
|
||||
sweeper.MarkNetworkChange()
|
||||
second := wrap(t, sweeper, connPair(t))
|
||||
sweeper.MarkNetworkChange()
|
||||
_ = wrap(t, sweeper, connPair(t))
|
||||
|
||||
buf := make([]byte, 1)
|
||||
for _, conn := range []net.Conn{first, second} {
|
||||
_ = conn.SetReadDeadline(time.Now().Add(time.Second))
|
||||
_, err := conn.Read(buf)
|
||||
require.ErrorIs(t, err, net.ErrClosed, "every pre-mark connection must be swept by the rescheduled sweep")
|
||||
}
|
||||
assert.Equal(t, 1, sweeper.sweepAll(), "only the newest-generation connection may remain")
|
||||
}
|
||||
|
||||
func TestNilSweeperIsNoop(t *testing.T) {
|
||||
var sweeper *Sweeper
|
||||
|
||||
conn := connPair(t)
|
||||
dial := sweeper.StartDial(context.Background())
|
||||
defer dial.Release()
|
||||
|
||||
wrapped, err := dial.WrapConn(conn)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, conn, wrapped, "nil sweeper must return the conn unchanged")
|
||||
assert.NoError(t, dial.Ctx().Err(), "nil sweeper must not cancel the dial context")
|
||||
assert.Equal(t, 0, sweeper.sweepAll(), "nil sweeper closes nothing")
|
||||
}
|
||||
|
||||
// wrap registers conn with the sweeper through a completed dial.
|
||||
func wrap(t *testing.T, sweeper *Sweeper, conn net.Conn) net.Conn {
|
||||
t.Helper()
|
||||
|
||||
dial := sweeper.StartDial(context.Background())
|
||||
defer dial.Release()
|
||||
|
||||
wrapped, err := dial.WrapConn(conn)
|
||||
require.NoError(t, err)
|
||||
return wrapped
|
||||
}
|
||||
|
||||
// connPair dials a loopback TCP connection and keeps the accepted peer open
|
||||
// until the test ends: a peer that closed early would make the connection
|
||||
// unreadable on its own, so a read error after the sweep would prove nothing.
|
||||
func connPair(t *testing.T) net.Conn {
|
||||
t.Helper()
|
||||
|
||||
l, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() {
|
||||
if err := l.Close(); err != nil {
|
||||
t.Logf("listener close error: %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
accepted := make(chan net.Conn, 1)
|
||||
go func() {
|
||||
conn, err := l.Accept()
|
||||
if err != nil {
|
||||
close(accepted)
|
||||
return
|
||||
}
|
||||
accepted <- conn
|
||||
}()
|
||||
|
||||
conn, err := net.Dial("tcp", l.Addr().String())
|
||||
require.NoError(t, err)
|
||||
|
||||
peer, ok := <-accepted
|
||||
require.True(t, ok, "listener must accept the dialed connection")
|
||||
t.Cleanup(func() {
|
||||
if err := peer.Close(); err != nil {
|
||||
t.Logf("peer close error: %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
return conn
|
||||
}
|
||||
|
||||
// sweepAll cuts every registration regardless of generation.
|
||||
func (s *Sweeper) sweepAll() int {
|
||||
return s.sweep(math.MaxUint64)
|
||||
}
|
||||
@@ -1,39 +0,0 @@
|
||||
package netsweep
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
const quickRetryDelay = 200 * time.Millisecond
|
||||
|
||||
type quickRetryBackoff struct {
|
||||
backoff.BackOff
|
||||
sweeper *Sweeper
|
||||
netState *netstate.State
|
||||
used bool
|
||||
}
|
||||
|
||||
func newQuickRetryBackoff(bo backoff.BackOff, sweeper *Sweeper, netState *netstate.State) *quickRetryBackoff {
|
||||
return &quickRetryBackoff{
|
||||
BackOff: bo,
|
||||
sweeper: sweeper,
|
||||
netState: netState,
|
||||
}
|
||||
}
|
||||
|
||||
func (b *quickRetryBackoff) NextBackOff() time.Duration {
|
||||
if !b.used && b.sweeper.markedRecently() && b.netState.IsOnline() {
|
||||
b.used = true
|
||||
return quickRetryDelay
|
||||
}
|
||||
return b.BackOff.NextBackOff()
|
||||
}
|
||||
|
||||
func (b *quickRetryBackoff) Reset() {
|
||||
b.used = false
|
||||
b.BackOff.Reset()
|
||||
}
|
||||
@@ -1,58 +0,0 @@
|
||||
package netsweep
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestQuickRetryAfterRecentMark(t *testing.T) {
|
||||
sweeper := New()
|
||||
sweeper.MarkNetworkChange()
|
||||
|
||||
slow := backoff.NewConstantBackOff(5 * time.Second)
|
||||
bo := sweeper.QuickRetryBackoff(context.Background(), slow, nil)
|
||||
|
||||
assert.Equal(t, quickRetryDelay, bo.NextBackOff(), "first retry after a mark must be quick")
|
||||
assert.Equal(t, 5*time.Second, bo.NextBackOff(), "second retry must fall back to the wrapped backoff")
|
||||
|
||||
bo.Reset()
|
||||
assert.Equal(t, quickRetryDelay, bo.NextBackOff(), "reset must re-arm the quick retry")
|
||||
}
|
||||
|
||||
func TestQuickRetryWithoutMarkKeepsSpread(t *testing.T) {
|
||||
sweeper := New()
|
||||
|
||||
slow := backoff.NewConstantBackOff(5 * time.Second)
|
||||
bo := sweeper.QuickRetryBackoff(context.Background(), slow, nil)
|
||||
|
||||
assert.Equal(t, 5*time.Second, bo.NextBackOff(), "without a mark the wrapped backoff decides")
|
||||
|
||||
sweeper.mu.Lock()
|
||||
sweeper.lastMark = time.Now().Add(-recentMarkWindow)
|
||||
sweeper.mu.Unlock()
|
||||
assert.Equal(t, 5*time.Second, bo.NextBackOff(), "a stale mark must not trigger the quick retry")
|
||||
}
|
||||
|
||||
func TestQuickRetryNilSweeperPassthrough(t *testing.T) {
|
||||
var sweeper *Sweeper
|
||||
|
||||
slow := backoff.NewConstantBackOff(5 * time.Second)
|
||||
bo := sweeper.QuickRetryBackoff(context.Background(), slow, nil)
|
||||
|
||||
assert.Equal(t, backoff.BackOff(slow), bo, "nil sweeper must return the backoff unchanged")
|
||||
}
|
||||
|
||||
func TestQuickRetryHonorsContext(t *testing.T) {
|
||||
sweeper := New()
|
||||
sweeper.MarkNetworkChange()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
bo := sweeper.QuickRetryBackoff(ctx, backoff.NewConstantBackOff(time.Millisecond), nil)
|
||||
|
||||
assert.Equal(t, backoff.Stop, bo.NextBackOff(), "cancelled context must stop the retry loop")
|
||||
}
|
||||
@@ -3,7 +3,6 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path"
|
||||
@@ -70,7 +69,7 @@ func setStdHandle(f *os.File) error {
|
||||
handle := f.Fd()
|
||||
r0, _, e1 := setStdHandleFn.Call(stdErrorHandle, handle)
|
||||
if r0 == 0 {
|
||||
if !errors.Is(e1, syscall.Errno(0)) {
|
||||
if e1 != nil {
|
||||
return e1
|
||||
}
|
||||
return syscall.EINVAL
|
||||
|
||||
@@ -10,7 +10,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go.uber.org/mock/gomock"
|
||||
"github.com/golang/mock/gomock"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.opentelemetry.io/otel"
|
||||
|
||||
|
||||
@@ -313,23 +313,21 @@ func Dial(ctx context.Context, addr, user string, opts DialOptions) (*Client, er
|
||||
|
||||
// dialSSH establishes an SSH connection without JWT authentication
|
||||
func dialSSH(ctx context.Context, network, addr string, config *ssh.ClientConfig) (*Client, error) {
|
||||
if config.Timeout > 0 {
|
||||
var cancel context.CancelFunc
|
||||
ctx, cancel = context.WithTimeout(ctx, config.Timeout)
|
||||
defer cancel()
|
||||
}
|
||||
|
||||
dialer := &net.Dialer{}
|
||||
conn, err := dialer.DialContext(ctx, network, addr)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("dial %s: %w", addr, err)
|
||||
}
|
||||
|
||||
client, err := nbssh.Handshake(ctx, conn, addr, config)
|
||||
clientConn, chans, reqs, err := ssh.NewClientConn(conn, addr, config)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
if closeErr := conn.Close(); closeErr != nil {
|
||||
log.Debugf("connection close after handshake failure: %v", closeErr)
|
||||
}
|
||||
return nil, fmt.Errorf("ssh handshake: %w", err)
|
||||
}
|
||||
|
||||
client := ssh.NewClient(clientConn, chans, reqs)
|
||||
return &Client{
|
||||
client: client,
|
||||
}, nil
|
||||
|
||||
@@ -12,8 +12,6 @@ import (
|
||||
log "github.com/sirupsen/logrus"
|
||||
"golang.org/x/crypto/ssh"
|
||||
"golang.org/x/term"
|
||||
|
||||
nbssh "github.com/netbirdio/netbird/client/ssh"
|
||||
)
|
||||
|
||||
func (c *Client) setupTerminalMode(ctx context.Context, session *ssh.Session) error {
|
||||
@@ -84,7 +82,37 @@ func (c *Client) setupTerminal(session *ssh.Session, fd int) error {
|
||||
return fmt.Errorf("get terminal size: %w", err)
|
||||
}
|
||||
|
||||
modes := nbssh.DefaultTerminalModes
|
||||
modes := ssh.TerminalModes{
|
||||
ssh.ECHO: 1,
|
||||
ssh.TTY_OP_ISPEED: 14400,
|
||||
ssh.TTY_OP_OSPEED: 14400,
|
||||
// Ctrl+C
|
||||
ssh.VINTR: 3,
|
||||
// Ctrl+\
|
||||
ssh.VQUIT: 28,
|
||||
// Backspace
|
||||
ssh.VERASE: 127,
|
||||
// Ctrl+U
|
||||
ssh.VKILL: 21,
|
||||
// Ctrl+D
|
||||
ssh.VEOF: 4,
|
||||
ssh.VEOL: 0,
|
||||
ssh.VEOL2: 0,
|
||||
// Ctrl+Q
|
||||
ssh.VSTART: 17,
|
||||
// Ctrl+S
|
||||
ssh.VSTOP: 19,
|
||||
// Ctrl+Z
|
||||
ssh.VSUSP: 26,
|
||||
// Ctrl+O
|
||||
ssh.VDISCARD: 15,
|
||||
// Ctrl+R
|
||||
ssh.VREPRINT: 18,
|
||||
// Ctrl+W
|
||||
ssh.VWERASE: 23,
|
||||
// Ctrl+V
|
||||
ssh.VLNEXT: 22,
|
||||
}
|
||||
|
||||
terminal := os.Getenv("TERM")
|
||||
if terminal == "" {
|
||||
|
||||
@@ -10,8 +10,6 @@ import (
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
"golang.org/x/crypto/ssh"
|
||||
|
||||
nbssh "github.com/netbirdio/netbird/client/ssh"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -82,14 +80,28 @@ func (c *Client) setupTerminalMode(_ context.Context, session *ssh.Session) erro
|
||||
w, h := c.getWindowsConsoleSize()
|
||||
|
||||
modes := ssh.TerminalModes{
|
||||
ssh.ICRNL: 1,
|
||||
ssh.OPOST: 1,
|
||||
ssh.ONLCR: 1,
|
||||
ssh.ISIG: 1,
|
||||
ssh.ICANON: 1,
|
||||
}
|
||||
for mode, value := range nbssh.DefaultTerminalModes {
|
||||
modes[mode] = value
|
||||
ssh.ECHO: 1,
|
||||
ssh.TTY_OP_ISPEED: 14400,
|
||||
ssh.TTY_OP_OSPEED: 14400,
|
||||
ssh.ICRNL: 1,
|
||||
ssh.OPOST: 1,
|
||||
ssh.ONLCR: 1,
|
||||
ssh.ISIG: 1,
|
||||
ssh.ICANON: 1,
|
||||
ssh.VINTR: 3, // Ctrl+C
|
||||
ssh.VQUIT: 28, // Ctrl+\
|
||||
ssh.VERASE: 127, // Backspace
|
||||
ssh.VKILL: 21, // Ctrl+U
|
||||
ssh.VEOF: 4, // Ctrl+D
|
||||
ssh.VEOL: 0,
|
||||
ssh.VEOL2: 0,
|
||||
ssh.VSTART: 17, // Ctrl+Q
|
||||
ssh.VSTOP: 19, // Ctrl+S
|
||||
ssh.VSUSP: 26, // Ctrl+Z
|
||||
ssh.VDISCARD: 15, // Ctrl+O
|
||||
ssh.VWERASE: 23, // Ctrl+W
|
||||
ssh.VLNEXT: 22, // Ctrl+V
|
||||
ssh.VREPRINT: 18, // Ctrl+R
|
||||
}
|
||||
|
||||
if err := session.RequestPty("xterm-256color", h, w, modes); err != nil {
|
||||
|
||||
@@ -35,19 +35,6 @@ type HostKeyVerifier interface {
|
||||
VerifySSHHostKey(peerAddress string, key []byte) error
|
||||
}
|
||||
|
||||
// PeerKeyLookup returns the stored SSH host key for a peer address.
|
||||
type PeerKeyLookup func(peerAddress string) ([]byte, bool)
|
||||
|
||||
// VerifySSHHostKey implements HostKeyVerifier by looking up the stored key
|
||||
// and comparing it against the presented key.
|
||||
func (l PeerKeyLookup) VerifySSHHostKey(peerAddress string, presentedKey []byte) error {
|
||||
storedKey, found := l(peerAddress)
|
||||
if !found {
|
||||
return ErrPeerNotFound
|
||||
}
|
||||
return VerifyHostKey(storedKey, presentedKey, peerAddress)
|
||||
}
|
||||
|
||||
// DaemonHostKeyVerifier implements HostKeyVerifier using the NetBird daemon
|
||||
type DaemonHostKeyVerifier struct {
|
||||
client proto.DaemonServiceClient
|
||||
|
||||
@@ -1,45 +0,0 @@
|
||||
package ssh
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
"golang.org/x/crypto/ssh"
|
||||
)
|
||||
|
||||
// Handshake runs the SSH client handshake on an already dialed conn and
|
||||
// returns the resulting client. Dialing bounds only the TCP establishment;
|
||||
// without a deadline on the socket a peer that accepts and then goes silent
|
||||
// blocks the handshake forever, so the context deadline is applied to conn
|
||||
// for the duration of the handshake. conn is closed on any error.
|
||||
func Handshake(ctx context.Context, conn net.Conn, addr string, config *ssh.ClientConfig) (*ssh.Client, error) {
|
||||
if deadline, ok := ctx.Deadline(); ok {
|
||||
if err := conn.SetDeadline(deadline); err != nil {
|
||||
closeHandshake(conn, "conn after deadline error")
|
||||
return nil, fmt.Errorf("set handshake deadline: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
sshConn, chans, reqs, err := ssh.NewClientConn(conn, addr, config)
|
||||
if err != nil {
|
||||
closeHandshake(conn, "conn after handshake error")
|
||||
return nil, fmt.Errorf("ssh handshake: %w", err)
|
||||
}
|
||||
|
||||
if err := conn.SetDeadline(time.Time{}); err != nil {
|
||||
closeHandshake(sshConn, "ssh conn after deadline clear error")
|
||||
return nil, fmt.Errorf("clear handshake deadline: %w", err)
|
||||
}
|
||||
|
||||
return ssh.NewClient(sshConn, chans, reqs), nil
|
||||
}
|
||||
|
||||
func closeHandshake(c io.Closer, label string) {
|
||||
if err := c.Close(); err != nil {
|
||||
log.Debugf("ssh: close %s: %v", label, err)
|
||||
}
|
||||
}
|
||||
@@ -610,10 +610,13 @@ func (p *SSHProxy) dialBackend(ctx context.Context, addr, user, jwtToken string)
|
||||
return nil, fmt.Errorf("connect to server: %w", err)
|
||||
}
|
||||
|
||||
handshakeCtx, cancel := context.WithTimeout(ctx, sshHandshakeTimeout)
|
||||
defer cancel()
|
||||
clientConn, chans, reqs, err := cryptossh.NewClientConn(conn, addr, config)
|
||||
if err != nil {
|
||||
_ = conn.Close()
|
||||
return nil, fmt.Errorf("SSH handshake: %w", err)
|
||||
}
|
||||
|
||||
return nbssh.Handshake(handshakeCtx, conn, addr, config)
|
||||
return cryptossh.NewClient(clientConn, chans, reqs), nil
|
||||
}
|
||||
|
||||
func (p *SSHProxy) verifyHostKey(hostname string, remote net.Addr, key cryptossh.PublicKey) error {
|
||||
|
||||
@@ -75,8 +75,8 @@ func (s *Server) createCommand(logger *log.Entry, privilegeResult PrivilegeCheck
|
||||
}
|
||||
|
||||
// Try su first for system integration (PAM/audit) when privileged
|
||||
cmd, err := s.createSuCommand(logger, session, localUser, hasPty) //nolint:staticcheck
|
||||
if err != nil || privilegeResult.UsedFallback { //nolint:staticcheck // always errors on platforms without su
|
||||
cmd, err := s.createSuCommand(logger, session, localUser, hasPty)
|
||||
if err != nil || privilegeResult.UsedFallback {
|
||||
logger.Debugf("su command failed, falling back to executor: %v", err)
|
||||
cmd, cleanup, err := s.createExecutorCommand(logger, session, localUser, hasPty)
|
||||
if err != nil {
|
||||
|
||||
@@ -1,84 +0,0 @@
|
||||
package ssh
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
"golang.org/x/crypto/ssh"
|
||||
)
|
||||
|
||||
// DefaultTerminalModes are the PTY modes used by the interactive terminal clients.
|
||||
var DefaultTerminalModes = ssh.TerminalModes{
|
||||
ssh.ECHO: 1,
|
||||
ssh.TTY_OP_ISPEED: 14400,
|
||||
ssh.TTY_OP_OSPEED: 14400,
|
||||
ssh.VINTR: 3, // Ctrl+C
|
||||
ssh.VQUIT: 28, // Ctrl+\
|
||||
ssh.VERASE: 127, // Backspace
|
||||
ssh.VKILL: 21, // Ctrl+U
|
||||
ssh.VEOF: 4, // Ctrl+D
|
||||
ssh.VEOL: 0,
|
||||
ssh.VEOL2: 0,
|
||||
ssh.VSTART: 17, // Ctrl+Q
|
||||
ssh.VSTOP: 19, // Ctrl+S
|
||||
ssh.VSUSP: 26, // Ctrl+Z
|
||||
ssh.VDISCARD: 15, // Ctrl+O
|
||||
ssh.VREPRINT: 18, // Ctrl+R
|
||||
ssh.VWERASE: 23, // Ctrl+W
|
||||
ssh.VLNEXT: 22, // Ctrl+V
|
||||
}
|
||||
|
||||
// PTYSession is an interactive shell session with a PTY and its I/O pipes.
|
||||
type PTYSession struct {
|
||||
Session *ssh.Session
|
||||
Stdin io.WriteCloser
|
||||
Stdout io.Reader
|
||||
Stderr io.Reader
|
||||
}
|
||||
|
||||
// StartPTYSession opens a session on the client, requests an xterm-256color PTY
|
||||
// with the default terminal modes, wires up the I/O pipes and starts a shell.
|
||||
// The session is closed on any error.
|
||||
func StartPTYSession(client *ssh.Client, cols, rows int) (*PTYSession, error) {
|
||||
session, err := client.NewSession()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("new session: %w", err)
|
||||
}
|
||||
|
||||
pty, err := setupPTYSession(session, cols, rows)
|
||||
if err != nil {
|
||||
if closeErr := session.Close(); closeErr != nil {
|
||||
log.Debugf("ssh: session close after setup error: %v", closeErr)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return pty, nil
|
||||
}
|
||||
|
||||
// setupPTYSession requests the PTY, opens the pipes and starts the shell on an
|
||||
// already created session.
|
||||
func setupPTYSession(session *ssh.Session, cols, rows int) (*PTYSession, error) {
|
||||
if err := session.RequestPty("xterm-256color", rows, cols, DefaultTerminalModes); err != nil {
|
||||
return nil, fmt.Errorf("request pty: %w", err)
|
||||
}
|
||||
|
||||
stdin, err := session.StdinPipe()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("stdin pipe: %w", err)
|
||||
}
|
||||
stdout, err := session.StdoutPipe()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("stdout pipe: %w", err)
|
||||
}
|
||||
stderr, err := session.StderrPipe()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("stderr pipe: %w", err)
|
||||
}
|
||||
|
||||
if err := session.Shell(); err != nil {
|
||||
return nil, fmt.Errorf("start shell: %w", err)
|
||||
}
|
||||
|
||||
return &PTYSession{Session: session, Stdin: stdin, Stdout: stdout, Stderr: stderr}, nil
|
||||
}
|
||||
@@ -30,11 +30,6 @@ func GetInfo(ctx context.Context) *Info {
|
||||
kernelVersion = osInfo[2]
|
||||
}
|
||||
|
||||
addrs, err := networkAddresses()
|
||||
if err != nil {
|
||||
log.Warnf("discover network addresses: %s", err)
|
||||
}
|
||||
|
||||
gio := &Info{
|
||||
GoOS: runtime.GOOS,
|
||||
Kernel: kernel,
|
||||
@@ -46,7 +41,6 @@ func GetInfo(ctx context.Context) *Info {
|
||||
NetbirdVersion: version.NetbirdVersion(),
|
||||
UIVersion: extractUIVersion(ctx),
|
||||
KernelVersion: kernelVersion,
|
||||
NetworkAddresses: addrs,
|
||||
SystemSerialNumber: serial(),
|
||||
SystemProductName: productModel(),
|
||||
SystemManufacturer: productManufacturer(),
|
||||
|
||||
@@ -15,7 +15,7 @@ func UpdateStaticInfoAsync() {
|
||||
}
|
||||
|
||||
// GetInfo retrieves system information for WASM environment
|
||||
func GetInfo(ctx context.Context) *Info {
|
||||
func GetInfo(_ context.Context) *Info {
|
||||
info := &Info{
|
||||
GoOS: runtime.GOOS,
|
||||
Kernel: runtime.GOARCH,
|
||||
@@ -30,13 +30,6 @@ func GetInfo(ctx context.Context) *Info {
|
||||
collectBrowserInfo(info)
|
||||
collectLocationInfo(info)
|
||||
collectSystemInfo(info)
|
||||
|
||||
// A caller-provided device name wins, as on the other platforms. A peer
|
||||
// registered over an API keeps reporting the name it was registered with,
|
||||
// so its meta does not change on the first sync.
|
||||
if name := extractDeviceName(ctx, info.Hostname); name != "" {
|
||||
info.Hostname = name
|
||||
}
|
||||
return info
|
||||
}
|
||||
|
||||
|
||||
@@ -1,27 +0,0 @@
|
||||
//go:build js
|
||||
|
||||
package system
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestGetInfoHonorsDeviceName covers a caller-provided device name reaching the
|
||||
// reported hostname, so a peer registered over an API keeps reporting the name
|
||||
// it was registered with instead of renaming itself on its first sync.
|
||||
func TestGetInfoHonorsDeviceName(t *testing.T) {
|
||||
ctx := context.WithValue(context.Background(), DeviceNameCtxKey, "session-name")
|
||||
if got := GetInfo(ctx).Hostname; got != "session-name" {
|
||||
t.Errorf("hostname should carry the caller's device name, got %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestGetInfoWithoutDeviceNameKeepsFallback covers the embed layer's habit of
|
||||
// always setting the context value: an empty name must not blank the hostname.
|
||||
func TestGetInfoWithoutDeviceNameKeepsFallback(t *testing.T) {
|
||||
ctx := context.WithValue(context.Background(), DeviceNameCtxKey, "")
|
||||
if got := GetInfo(ctx).Hostname; got == "" {
|
||||
t.Error("an empty device name must not blank the hostname")
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
//go:build !ios && !android
|
||||
//go:build !ios
|
||||
|
||||
package system
|
||||
|
||||
|
||||
@@ -1,89 +0,0 @@
|
||||
package system
|
||||
|
||||
import (
|
||||
"net/netip"
|
||||
"strings"
|
||||
)
|
||||
|
||||
var iFaceDiscover IFaceDiscover
|
||||
|
||||
type IFaceDiscover interface {
|
||||
IFaces() (string, error)
|
||||
}
|
||||
|
||||
// SetIFaceDiscover configures the Android interface discovery provider.
|
||||
func SetIFaceDiscover(discover IFaceDiscover) {
|
||||
iFaceDiscover = discover
|
||||
}
|
||||
|
||||
func networkAddresses() ([]NetworkAddress, error) {
|
||||
if iFaceDiscover == nil {
|
||||
return nil, nil
|
||||
}
|
||||
ifaces, err := iFaceDiscover.IFaces()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var netAddresses []NetworkAddress
|
||||
for _, line := range strings.Split(ifaces, "\n") {
|
||||
addresses, ok := interfaceAddresses(line)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
for _, address := range addresses {
|
||||
netAddr, ok := toNetworkAddress(address)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if isDuplicated(netAddresses, netAddr) {
|
||||
continue
|
||||
}
|
||||
netAddresses = append(netAddresses, netAddr)
|
||||
}
|
||||
}
|
||||
return netAddresses, nil
|
||||
}
|
||||
|
||||
func interfaceAddresses(line string) ([]string, bool) {
|
||||
parts := strings.Split(line, "|")
|
||||
if len(parts) != 2 {
|
||||
return nil, false
|
||||
}
|
||||
flags := strings.Fields(parts[0])
|
||||
if len(flags) != 8 {
|
||||
return nil, false
|
||||
}
|
||||
up, loopback := flags[3], flags[5]
|
||||
if up != "true" || loopback == "true" {
|
||||
return nil, false
|
||||
}
|
||||
return strings.Fields(parts[1]), true
|
||||
}
|
||||
|
||||
func toNetworkAddress(address string) (NetworkAddress, bool) {
|
||||
prefix, err := netip.ParsePrefix(address)
|
||||
if err != nil {
|
||||
return NetworkAddress{}, false
|
||||
}
|
||||
if prefix.Addr().Is4In6() {
|
||||
if prefix.Bits() < 96 {
|
||||
return NetworkAddress{}, false
|
||||
}
|
||||
prefix = netip.PrefixFrom(prefix.Addr().Unmap(), prefix.Bits()-96)
|
||||
}
|
||||
ip := prefix.Addr()
|
||||
if ip.IsLoopback() || ip.IsLinkLocalUnicast() || ip.IsMulticast() {
|
||||
return NetworkAddress{}, false
|
||||
}
|
||||
return NetworkAddress{NetIP: prefix}, true
|
||||
}
|
||||
|
||||
func isDuplicated(addresses []NetworkAddress, addr NetworkAddress) bool {
|
||||
for _, duplicated := range addresses {
|
||||
if duplicated.NetIP == addr.NetIP {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
//go:build !ios && !android
|
||||
//go:build !ios
|
||||
|
||||
package system
|
||||
|
||||
|
||||
@@ -1,5 +1,3 @@
|
||||
//go:build windows || (linux && !android) || (darwin && !ios) || freebsd
|
||||
|
||||
package system
|
||||
|
||||
import (
|
||||
|
||||
18
client/ui/frontend/src/components/ReadySignal.tsx
Normal file
18
client/ui/frontend/src/components/ReadySignal.tsx
Normal file
@@ -0,0 +1,18 @@
|
||||
import { useEffect, useRef } from "react";
|
||||
import { Events } from "@wailsio/runtime";
|
||||
import { useStatus } from "@/contexts/StatusContext.tsx";
|
||||
|
||||
const EVENT_WINDOW_PAINTED = "netbird:window-painted";
|
||||
|
||||
export const ReadySignal = () => {
|
||||
const { isReady } = useStatus();
|
||||
const sent = useRef(false);
|
||||
|
||||
useEffect(() => {
|
||||
if (!isReady || sent.current) return;
|
||||
sent.current = true;
|
||||
void Events.Emit(EVENT_WINDOW_PAINTED);
|
||||
}, [isReady]);
|
||||
|
||||
return null;
|
||||
};
|
||||
Some files were not shown because too many files have changed in this diff Show More
Reference in New Issue
Block a user