mirror of
https://github.com/netbirdio/netbird.git
synced 2026-07-25 01:41:29 +02:00
Compare commits
11 Commits
add-atomic
...
revert/com
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7ed3737cda | ||
|
|
b65ec8b68a | ||
|
|
e13bcdbd44 | ||
|
|
46568f7af8 | ||
|
|
3358138ccc | ||
|
|
6d15d0729a | ||
|
|
0936918d24 | ||
|
|
d4a4418969 | ||
|
|
31ed241a1a | ||
|
|
178e6a8530 | ||
|
|
96963b6751 |
3
.github/workflows/agent-network-e2e.yml
vendored
3
.github/workflows/agent-network-e2e.yml
vendored
@@ -51,6 +51,9 @@ jobs:
|
||||
# token (and URL, for gateways) is unset, so partial coverage is fine.
|
||||
OPENAI_TOKEN: ${{ secrets.E2E_OPENAI_TOKEN }}
|
||||
ANTHROPIC_TOKEN: ${{ secrets.E2E_ANTHROPIC_TOKEN }}
|
||||
# Moonshot AI platform key (platform.kimi.ai); drives both Kimi wire
|
||||
# shapes (OpenAI /v1 and Anthropic /anthropic) through kimi_api.
|
||||
KIMI_TOKEN: ${{ secrets.E2E_KIMI_TOKEN }}
|
||||
VERCEL_URL: ${{ secrets.E2E_VERCEL_URL }}
|
||||
VERCEL_TOKEN: ${{ secrets.E2E_VERCEL_TOKEN }}
|
||||
OPENROUTER_URL: ${{ secrets.E2E_OPENROUTER_URL }}
|
||||
|
||||
@@ -569,17 +569,17 @@ func Test_ConnectPeers(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// todo: investigate why in some tests execution we need 30s
|
||||
// The peers use userspace WireGuard (stdnet transport). A tight busy-loop
|
||||
// here starves the wireguard-go goroutines that process the handshake, so
|
||||
// poll on a ticker instead and yield the CPU between checks. WireGuard also
|
||||
// only retries a lost handshake initiation every REKEY_TIMEOUT (5s), which
|
||||
// is why the overall wait can occasionally stretch to tens of seconds.
|
||||
timeout := 30 * time.Second
|
||||
timeoutChannel := time.After(timeout)
|
||||
ticker := time.NewTicker(500 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-timeoutChannel:
|
||||
t.Fatalf("waiting for peer handshake timeout after %s", timeout.String())
|
||||
default:
|
||||
}
|
||||
|
||||
peer, gpErr := getPeer(peer1ifaceName, peer2Key.PublicKey().String())
|
||||
if gpErr != nil {
|
||||
t.Fatal(gpErr)
|
||||
@@ -588,6 +588,12 @@ func Test_ConnectPeers(t *testing.T) {
|
||||
t.Log("peers successfully handshake")
|
||||
break
|
||||
}
|
||||
|
||||
select {
|
||||
case <-timeoutChannel:
|
||||
t.Fatalf("waiting for peer handshake timeout after %s", timeout.String())
|
||||
case <-ticker.C:
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -49,11 +49,21 @@ type ConnMgr struct {
|
||||
// engine.syncMsgMux; all other reads stay under engine.syncMsgMux only.
|
||||
lazyConnMgrMu sync.RWMutex
|
||||
|
||||
// reconcileRoutedIPs re-applies a peer's routed allowed IPs after its lazy wake endpoint is
|
||||
// (re)armed (Mode A at arm time). Injected by the engine; nil disables the reconcile.
|
||||
reconcileRoutedIPs func(peerKey string) error
|
||||
|
||||
wg sync.WaitGroup
|
||||
lazyCtx context.Context
|
||||
lazyCtxCancel context.CancelFunc
|
||||
}
|
||||
|
||||
// SetRoutedIPsReconciler injects the callback used to re-apply a peer's routed allowed IPs when
|
||||
// its lazy wake endpoint is (re)armed. Must be called before the lazy manager starts.
|
||||
func (e *ConnMgr) SetRoutedIPsReconciler(fn func(peerKey string) error) {
|
||||
e.reconcileRoutedIPs = fn
|
||||
}
|
||||
|
||||
func NewConnMgr(engineConfig *EngineConfig, statusRecorder *peer.Status, peerStore *peerstore.Store, iface lazyconn.WGIface) *ConnMgr {
|
||||
e := &ConnMgr{
|
||||
peerStore: peerStore,
|
||||
@@ -291,6 +301,7 @@ func (e *ConnMgr) Close() {
|
||||
func (e *ConnMgr) initLazyManager(engineCtx context.Context) {
|
||||
cfg := manager.Config{
|
||||
InactivityThreshold: inactivityThresholdEnv(),
|
||||
ReconcileAllowedIPs: e.reconcileRoutedIPs,
|
||||
}
|
||||
|
||||
e.lazyConnMgrMu.Lock()
|
||||
|
||||
Binary file not shown.
Binary file not shown.
@@ -52,11 +52,14 @@ int xdp_dns_fwd(struct iphdr *ip, struct udphdr *udp) {
|
||||
|
||||
if (udp->dest == GENERAL_DNS_PORT && ip->daddr == dns_ip) {
|
||||
udp->dest = dns_port;
|
||||
// Clear the now-stale checksum; zero means "not computed" for IPv4.
|
||||
udp->check = 0;
|
||||
return XDP_PASS;
|
||||
}
|
||||
|
||||
if (udp->source == dns_port && ip->saddr == dns_ip) {
|
||||
udp->source = GENERAL_DNS_PORT;
|
||||
udp->check = 0;
|
||||
return XDP_PASS;
|
||||
}
|
||||
|
||||
|
||||
@@ -50,5 +50,11 @@ int xdp_wg_proxy(struct iphdr *ip, struct udphdr *udp) {
|
||||
__be16 new_dst_port = htons(proxy_port);
|
||||
udp->dest = new_dst_port;
|
||||
udp->source = new_src_port;
|
||||
|
||||
// The ports are covered by the UDP checksum. This is an IPv4 loopback hop
|
||||
// and the payload is already integrity-protected, so clear the checksum (a
|
||||
// zero UDP checksum means "not computed" for IPv4) rather than leave a
|
||||
// stale value the kernel would drop as UDP_CSUM.
|
||||
udp->check = 0;
|
||||
return XDP_PASS;
|
||||
}
|
||||
|
||||
@@ -663,6 +663,12 @@ func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL)
|
||||
iceCfg := e.createICEConfig()
|
||||
|
||||
e.connMgr = NewConnMgr(e.config, e.statusRecorder, e.peerStore, wgIface)
|
||||
e.connMgr.SetRoutedIPsReconciler(func(peerKey string) error {
|
||||
if e.routeManager == nil {
|
||||
return nil
|
||||
}
|
||||
return e.routeManager.ReconcilePeerAllowedIPs(peerKey)
|
||||
})
|
||||
e.connMgr.Start(e.ctx)
|
||||
|
||||
// Wire DNS-time lazy-connection warm-up now that the connection manager
|
||||
|
||||
@@ -29,6 +29,11 @@ type managedPeer struct {
|
||||
|
||||
type Config struct {
|
||||
InactivityThreshold *time.Duration
|
||||
// ReconcileAllowedIPs re-applies a peer's routed allowed IPs after its wake endpoint is
|
||||
// armed. The activity listener creates the wake peer with the overlay /32 only; without the
|
||||
// routed prefixes WireGuard would not steer subnet-bound traffic to the wake endpoint, so an
|
||||
// idle routing peer could never be woken by that traffic. Optional; nil disables the reconcile.
|
||||
ReconcileAllowedIPs func(peerKey string) error
|
||||
}
|
||||
|
||||
// Manager manages lazy connections
|
||||
@@ -56,6 +61,9 @@ type Manager struct {
|
||||
peerToHAGroups map[string][]route.HAUniqueID // peer ID -> HA groups they belong to
|
||||
haGroupToPeers map[route.HAUniqueID][]string // HA group -> peer IDs in the group
|
||||
routesMu sync.RWMutex
|
||||
|
||||
// reconcileAllowedIPs re-applies a peer's routed allowed IPs after its wake endpoint is armed.
|
||||
reconcileAllowedIPs func(peerKey string) error
|
||||
}
|
||||
|
||||
// NewManager creates a new lazy connection manager
|
||||
@@ -73,6 +81,7 @@ func NewManager(config Config, engineCtx context.Context, peerStore *peerstore.S
|
||||
activityManager: activity.NewManager(wgIface),
|
||||
peerToHAGroups: make(map[string][]route.HAUniqueID),
|
||||
haGroupToPeers: make(map[route.HAUniqueID][]string),
|
||||
reconcileAllowedIPs: config.ReconcileAllowedIPs,
|
||||
}
|
||||
|
||||
if wgIface.IsUserspaceBind() {
|
||||
@@ -201,7 +210,7 @@ func (m *Manager) AddPeer(peerCfg lazyconn.PeerConfig) (bool, error) {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
if err := m.activityManager.MonitorPeerActivity(peerCfg); err != nil {
|
||||
if err := m.armActivityListener(peerCfg); err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
@@ -288,7 +297,7 @@ func (m *Manager) DeactivatePeer(peerID peerid.ConnID) {
|
||||
|
||||
m.inactivityManager.RemovePeer(mp.peerCfg.PublicKey)
|
||||
|
||||
if err := m.activityManager.MonitorPeerActivity(*mp.peerCfg); err != nil {
|
||||
if err := m.armActivityListener(*mp.peerCfg); err != nil {
|
||||
mp.peerCfg.Log.Errorf("failed to create activity monitor: %v", err)
|
||||
return
|
||||
}
|
||||
@@ -465,6 +474,31 @@ func (m *Manager) close() {
|
||||
}
|
||||
|
||||
// shouldDeferIdleForHA checks if peer should stay connected due to HA group requirements
|
||||
// armRoutedAllowedIPs re-applies the peer's routed allowed IPs onto its freshly armed wake
|
||||
// endpoint. The activity listener creates the wake peer with the overlay /32 only, so without
|
||||
// this the routed prefixes would be missing and traffic to a routed subnet could not wake the
|
||||
// idle routing peer. It is a no-op when no reconciler is configured.
|
||||
// armActivityListener (re)arms the peer's wake endpoint via the activity manager and then
|
||||
// re-applies its routed allowed IPs, so traffic to a routed subnet can wake an idle routing
|
||||
// peer. The routed prefixes must be re-applied after the wake endpoint exists because the
|
||||
// listener creates it with the overlay /32 only.
|
||||
func (m *Manager) armActivityListener(peerCfg lazyconn.PeerConfig) error {
|
||||
if err := m.activityManager.MonitorPeerActivity(peerCfg); err != nil {
|
||||
return err
|
||||
}
|
||||
m.armRoutedAllowedIPs(&peerCfg)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) armRoutedAllowedIPs(peerCfg *lazyconn.PeerConfig) {
|
||||
if m.reconcileAllowedIPs == nil {
|
||||
return
|
||||
}
|
||||
if err := m.reconcileAllowedIPs(peerCfg.PublicKey); err != nil {
|
||||
peerCfg.Log.Errorf("failed to reconcile routed allowed IPs on wake endpoint: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) shouldDeferIdleForHA(inactivePeers map[string]struct{}, peerID string) bool {
|
||||
m.routesMu.RLock()
|
||||
defer m.routesMu.RUnlock()
|
||||
@@ -577,7 +611,7 @@ func (m *Manager) onPeerInactivityTimedOut(peerIDs map[string]struct{}) {
|
||||
|
||||
mp.peerCfg.Log.Infof("start activity monitor")
|
||||
|
||||
if err := m.activityManager.MonitorPeerActivity(*mp.peerCfg); err != nil {
|
||||
if err := m.armActivityListener(*mp.peerCfg); err != nil {
|
||||
mp.peerCfg.Log.Errorf("failed to create activity monitor: %v", err)
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -61,6 +61,7 @@ type Manager interface {
|
||||
InitialRouteRange() []string
|
||||
SetFirewall(firewall.Manager) error
|
||||
SetDNSForwarderPort(port uint16)
|
||||
ReconcilePeerAllowedIPs(peerKey string) error
|
||||
Stop(stateManager *statemanager.Manager)
|
||||
}
|
||||
|
||||
@@ -232,6 +233,30 @@ func (m *DefaultManager) setupRefCounters(useNoop bool) {
|
||||
)
|
||||
}
|
||||
|
||||
// ReconcilePeerAllowedIPs re-applies every routed allowed IP currently tracked for the peer
|
||||
// onto the WireGuard device. The allowed-IP refcounter only calls its AddFunc (which pushes to
|
||||
// the device) on a prefix's 0->1 transition, so a peer whose device entry was rebuilt without a
|
||||
// matching refcounter change — e.g. a lazy connection cycling through idle->wake, which recreates
|
||||
// the WireGuard peer with the overlay /32 only — ends up missing routed prefixes the refcounter
|
||||
// still considers installed, and nothing retries. Calling this when the peer's WireGuard entry is
|
||||
// (re)created restores convergence. It is add-only and idempotent: AddAllowedIP is update-only, so
|
||||
// prefixes are re-added to an existing peer and an absent peer is left untouched.
|
||||
func (m *DefaultManager) ReconcilePeerAllowedIPs(peerKey string) error {
|
||||
if m.allowedIPsRefCounter == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
return m.allowedIPsRefCounter.ReapplyMatching(
|
||||
func(out string) bool { return out == peerKey },
|
||||
func(prefix netip.Prefix) error {
|
||||
if err := m.wgInterface.AddAllowedIP(peerKey, prefix); err != nil {
|
||||
return fmt.Errorf("add allowed IP %s for peer %s: %w", prefix, peerKey, err)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
// Init sets up the routing
|
||||
func (m *DefaultManager) Init() error {
|
||||
m.routeSelector = m.initSelector()
|
||||
|
||||
@@ -112,6 +112,11 @@ func (m *MockManager) SetFirewall(firewall.Manager) error {
|
||||
func (m *MockManager) SetDNSForwarderPort(port uint16) {
|
||||
}
|
||||
|
||||
// ReconcilePeerAllowedIPs mock implementation of ReconcilePeerAllowedIPs from Manager interface
|
||||
func (m *MockManager) ReconcilePeerAllowedIPs(peerKey string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop mock implementation of Stop from Manager interface
|
||||
func (m *MockManager) Stop(stateManager *statemanager.Manager) {
|
||||
if m.StopFunc != nil {
|
||||
|
||||
90
client/internal/routemanager/reconcile_test.go
Normal file
90
client/internal/routemanager/reconcile_test.go
Normal file
@@ -0,0 +1,90 @@
|
||||
//go:build !windows
|
||||
|
||||
package routemanager
|
||||
|
||||
import (
|
||||
"net"
|
||||
"net/netip"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"golang.zx2c4.com/wireguard/tun/netstack"
|
||||
|
||||
"github.com/netbirdio/netbird/client/iface/device"
|
||||
"github.com/netbirdio/netbird/client/iface/wgaddr"
|
||||
"github.com/netbirdio/netbird/client/internal/routemanager/refcounter"
|
||||
)
|
||||
|
||||
// reconcileWGMock is a minimal iface.WGIface that only records AddAllowedIP calls; every other
|
||||
// method is an inert stub because ReconcilePeerAllowedIPs exercises none of them.
|
||||
type reconcileWGMock struct {
|
||||
mu sync.Mutex
|
||||
adds map[string][]netip.Prefix
|
||||
}
|
||||
|
||||
func (m *reconcileWGMock) AddAllowedIP(peerKey string, allowedIP netip.Prefix) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if m.adds == nil {
|
||||
m.adds = map[string][]netip.Prefix{}
|
||||
}
|
||||
m.adds[peerKey] = append(m.adds[peerKey], allowedIP)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *reconcileWGMock) added(peerKey string) []netip.Prefix {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.adds[peerKey]
|
||||
}
|
||||
|
||||
func (m *reconcileWGMock) RemoveAllowedIP(string, netip.Prefix) error { return nil }
|
||||
func (m *reconcileWGMock) Name() string { return "utun-test" }
|
||||
func (m *reconcileWGMock) Address() wgaddr.Address { return wgaddr.Address{} }
|
||||
func (m *reconcileWGMock) ToInterface() *net.Interface { return nil }
|
||||
func (m *reconcileWGMock) IsUserspaceBind() bool { return false }
|
||||
func (m *reconcileWGMock) GetFilter() device.PacketFilter { return nil }
|
||||
func (m *reconcileWGMock) GetDevice() *device.FilteredDevice { return nil }
|
||||
func (m *reconcileWGMock) GetNet() *netstack.Net { return nil }
|
||||
|
||||
// TestReconcilePeerAllowedIPs verifies the declarative reconcile re-applies every routed prefix
|
||||
// tracked for the peer (self-heal, independent of refcount level) and stays scoped to that peer.
|
||||
func TestReconcilePeerAllowedIPs(t *testing.T) {
|
||||
wg := &reconcileWGMock{}
|
||||
m := &DefaultManager{wgInterface: wg}
|
||||
m.allowedIPsRefCounter = refcounter.New[netip.Prefix, string, string](
|
||||
func(_ netip.Prefix, peerKey string) (string, error) { return peerKey, nil },
|
||||
func(netip.Prefix, string) error { return nil },
|
||||
)
|
||||
|
||||
peerA1 := netip.MustParsePrefix("10.0.0.0/24")
|
||||
peerA2 := netip.MustParsePrefix("10.1.0.0/24")
|
||||
peerB1 := netip.MustParsePrefix("10.2.0.0/24")
|
||||
|
||||
for prefix, peer := range map[netip.Prefix]string{peerA1: "peerA", peerA2: "peerA", peerB1: "peerB"} {
|
||||
_, err := m.allowedIPsRefCounter.Increment(prefix, peer)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
// Extra reference: reconcile must still re-apply the prefix even though its refcount never
|
||||
// hit 0 again (the exact case the plain incremental path skips).
|
||||
_, err := m.allowedIPsRefCounter.Increment(peerA1, "peerA")
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NoError(t, m.ReconcilePeerAllowedIPs("peerA"))
|
||||
|
||||
assert.ElementsMatch(t, []netip.Prefix{peerA1, peerA2}, wg.added("peerA"),
|
||||
"reconcile must re-apply all routed prefixes of the peer")
|
||||
assert.Empty(t, wg.added("peerB"), "reconcile must not touch another peer's prefixes")
|
||||
}
|
||||
|
||||
// TestReconcilePeerAllowedIPsNoCounter verifies reconcile is a safe no-op before the refcounter is
|
||||
// set up.
|
||||
func TestReconcilePeerAllowedIPsNoCounter(t *testing.T) {
|
||||
wg := &reconcileWGMock{}
|
||||
m := &DefaultManager{wgInterface: wg}
|
||||
|
||||
require.NoError(t, m.ReconcilePeerAllowedIPs("peerA"))
|
||||
assert.Empty(t, wg.added("peerA"))
|
||||
}
|
||||
@@ -94,6 +94,26 @@ func (rm *Counter[Key, I, O]) Get(key Key) (Ref[O], bool) {
|
||||
return ref, ok
|
||||
}
|
||||
|
||||
// ReapplyMatching calls apply for every key whose stored Out satisfies pred, holding the
|
||||
// counter lock for the whole pass. Running apply under the lock keeps it atomic with respect
|
||||
// to Increment/Decrement: a prefix dropped to zero is removed from the map (and had its
|
||||
// RemoveFunc called) before this pass observes it, so a stale key can never be re-applied.
|
||||
// pred and apply are invoked under the lock, so they must not call back into the counter.
|
||||
func (rm *Counter[Key, I, O]) ReapplyMatching(pred func(out O) bool, apply func(key Key) error) error {
|
||||
rm.mu.Lock()
|
||||
defer rm.mu.Unlock()
|
||||
|
||||
var merr *multierror.Error
|
||||
for key, ref := range rm.refCountMap {
|
||||
if pred(ref.Out) {
|
||||
if err := apply(key); err != nil {
|
||||
merr = multierror.Append(merr, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nberrors.FormatErrorOrNil(merr)
|
||||
}
|
||||
|
||||
// Increment increments the reference count for the given key.
|
||||
// If this is the first reference to the key, the AddFunc is called.
|
||||
func (rm *Counter[Key, I, O]) Increment(key Key, in I) (Ref[O], error) {
|
||||
|
||||
47
client/internal/routemanager/refcounter/refcounter_test.go
Normal file
47
client/internal/routemanager/refcounter/refcounter_test.go
Normal file
@@ -0,0 +1,47 @@
|
||||
package refcounter
|
||||
|
||||
import (
|
||||
"net/netip"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestReapplyMatching verifies ReapplyMatching invokes apply for exactly the keys whose stored
|
||||
// Out satisfies the predicate (no duplicates for multiply-referenced keys) — the primitive
|
||||
// ReconcilePeerAllowedIPs relies on to re-apply a single peer's routed prefixes.
|
||||
func TestReapplyMatching(t *testing.T) {
|
||||
rc := New[netip.Prefix, string, string](
|
||||
func(_ netip.Prefix, peerKey string) (string, error) { return peerKey, nil },
|
||||
func(netip.Prefix, string) error { return nil },
|
||||
)
|
||||
|
||||
peerA1 := netip.MustParsePrefix("10.0.0.0/24")
|
||||
peerA2 := netip.MustParsePrefix("10.1.0.0/24")
|
||||
peerB1 := netip.MustParsePrefix("10.2.0.0/24")
|
||||
|
||||
for prefix, peer := range map[netip.Prefix]string{peerA1: "peerA", peerA2: "peerA", peerB1: "peerB"} {
|
||||
_, err := rc.Increment(prefix, peer)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
// a second reference must not make the key applied twice
|
||||
_, err := rc.Increment(peerA1, "peerA")
|
||||
require.NoError(t, err)
|
||||
|
||||
var applied []netip.Prefix
|
||||
err = rc.ReapplyMatching(
|
||||
func(out string) bool { return out == "peerA" },
|
||||
func(key netip.Prefix) error { applied = append(applied, key); return nil },
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []netip.Prefix{peerA1, peerA2}, applied)
|
||||
|
||||
var none []netip.Prefix
|
||||
err = rc.ReapplyMatching(
|
||||
func(out string) bool { return out == "missing" },
|
||||
func(key netip.Prefix) error { none = append(none, key); return nil },
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, none)
|
||||
}
|
||||
@@ -20,14 +20,15 @@ import (
|
||||
// covers whatever credentials are present (source ~/.llm-keys locally / set the
|
||||
// Actions secrets in CI).
|
||||
type providerCase struct {
|
||||
name string
|
||||
catalogID string
|
||||
upstream string
|
||||
apiKey string
|
||||
model string // body model (chat/messages) or path model@version (vertex)
|
||||
kind string // harness.WireChat, harness.WireMessages, or harness.WireVertex
|
||||
project string // vertex only: GCP project for the rawPredict path
|
||||
region string // vertex only: GCP region for the rawPredict path
|
||||
name string
|
||||
catalogID string
|
||||
upstream string
|
||||
apiKey string
|
||||
model string // body model (chat/messages) or path model@version (vertex)
|
||||
kind string // harness.WireChat, harness.WireMessages, or harness.WireVertex
|
||||
project string // vertex only: GCP project for the rawPredict path
|
||||
region string // vertex only: GCP region for the rawPredict path
|
||||
pathPrefix string // base-URL path prefix the agent carries (e.g. "/anthropic" for Kimi)
|
||||
}
|
||||
|
||||
// availableProviders builds the matrix from the provider env vars that are set.
|
||||
@@ -39,6 +40,24 @@ func availableProviders() []providerCase {
|
||||
if k := os.Getenv("ANTHROPIC_TOKEN"); k != "" {
|
||||
ps = append(ps, providerCase{name: "anthropic", catalogID: "anthropic_api", upstream: "https://api.anthropic.com", apiKey: k, model: "claude-haiku-4-5", kind: harness.WireMessages})
|
||||
}
|
||||
if k := os.Getenv("KIMI_TOKEN"); k != "" {
|
||||
// Kimi (Moonshot AI) serves two body shapes from the same key: OpenAI
|
||||
// Chat Completions on the bare host (/v1/...) and the Anthropic
|
||||
// Messages API under the /anthropic path prefix (the endpoint
|
||||
// Moonshot's Claude Code guide uses). The provider keeps the bare
|
||||
// default upstream and the AGENT carries the /anthropic prefix in
|
||||
// its base URL — exactly the documented Claude Code / Kimi CLI
|
||||
// setup (ANTHROPIC_BASE_URL=https://<endpoint>/anthropic) — so one
|
||||
// provider serves both shapes and the prefix rides through to
|
||||
// Moonshot. Run the Anthropic shape, the flagship Claude Code path;
|
||||
// the OpenAI wire shape is covered live by the other chat-shaped
|
||||
// matrix providers, and Kimi-over-chat passed with kimi-k3 before
|
||||
// the single-model constraint surfaced (run #73 on the kimi feature
|
||||
// branch). The platform serves this account exactly ONE model —
|
||||
// kimi-k3 (kimi-k2-thinking and even kimi-latest return
|
||||
// resource_not_found_error on both surfaces).
|
||||
ps = append(ps, providerCase{name: "kimi", catalogID: "kimi_api", upstream: "https://api.moonshot.ai", apiKey: k, model: "kimi-k3", kind: harness.WireMessages, pathPrefix: "/anthropic"})
|
||||
}
|
||||
if k, u := os.Getenv("VERCEL_TOKEN"), os.Getenv("VERCEL_URL"); k != "" && u != "" {
|
||||
ps = append(ps, providerCase{name: "vercel", catalogID: "vercel_ai_gateway", upstream: u, apiKey: k, model: "openai/gpt-4o-mini", kind: harness.WireChat})
|
||||
}
|
||||
@@ -84,12 +103,18 @@ func availableProviders() []providerCase {
|
||||
}
|
||||
}
|
||||
|
||||
// Bedrock: path-routed, bearer auth. Model is a cross-region inference
|
||||
// profile id (distinct string from the first-party Anthropic case).
|
||||
// Bedrock: path-routed, bearer auth. Model is the FULL cross-region
|
||||
// inference-profile id exactly as AWS issues it — region-family prefix
|
||||
// plus the date/version suffix. A bare or wrong-region id makes Bedrock
|
||||
// reject the request with "The provided model identifier is invalid"
|
||||
// before any inference runs. The proxy normalizes this id to the catalog
|
||||
// key (anthropic.claude-haiku-4-5) for routing/pricing/allowlists.
|
||||
// Defaults pair eu-central-1 with the eu.* profile; AWS_REGION overrides
|
||||
// the region and the prefix follows its family.
|
||||
if k := os.Getenv("AWS_BEARER_TOKEN_BEDROCK"); k != "" {
|
||||
region := os.Getenv("AWS_REGION")
|
||||
if region == "" {
|
||||
region = "us-east-1"
|
||||
region = "eu-central-1"
|
||||
}
|
||||
// A valid Bedrock inference-profile id (region prefix + date + version),
|
||||
// overridable per account. `global.` profiles can be invoked from any
|
||||
@@ -216,8 +241,7 @@ func TestProvidersMatrix(t *testing.T) {
|
||||
t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
|
||||
|
||||
require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
|
||||
// Resolve first: the DNS lookup triggers the lazy-connection warm-up, waking
|
||||
// the proxy peer so WaitProxyPeer then observes it connected.
|
||||
// Probe first: the GET resolves the endpoint (DNS error fails) and its first packet wakes the lazy proxy peer, so WaitProxyPeer sees it connected; any HTTP status counts.
|
||||
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
|
||||
require.NoError(t, err, "resolve agent-network endpoint to proxy IP")
|
||||
if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
|
||||
@@ -247,7 +271,7 @@ func TestProvidersMatrix(t *testing.T) {
|
||||
case harness.WireBedrock:
|
||||
c, b, cerr = cl.Bedrock(ctx, settings.Endpoint, proxyIP, pc.model, "Reply with exactly: pong", sessionID)
|
||||
default:
|
||||
c, b, cerr = cl.Chat(ctx, settings.Endpoint, proxyIP, pc.kind, pc.model, "Reply with exactly: pong", sessionID)
|
||||
c, b, cerr = cl.ChatPrefixed(ctx, settings.Endpoint, proxyIP, pc.pathPrefix, pc.kind, pc.model, "Reply with exactly: pong", sessionID)
|
||||
}
|
||||
if cerr == nil {
|
||||
code, body = c, b
|
||||
|
||||
@@ -52,7 +52,9 @@ func catalogModel(pc providerCase) string {
|
||||
func disallowedModel(pc providerCase) string {
|
||||
switch pc.kind {
|
||||
case harness.WireBedrock:
|
||||
return "us.anthropic.claude-opus-4-8"
|
||||
// Same profile prefix as the allowed model so only the model name
|
||||
// differs; the guardrail must deny it before it reaches AWS.
|
||||
return strings.SplitN(pc.model, ".", 2)[0] + ".anthropic.claude-opus-4-8"
|
||||
case harness.WireVertex:
|
||||
return "claude-opus-4-8@20250101"
|
||||
default:
|
||||
@@ -72,7 +74,7 @@ func sendModel(ctx context.Context, t *testing.T, cl *harness.Client, endpoint,
|
||||
case harness.WireVertex:
|
||||
code, _, err = cl.Vertex(ctx, endpoint, proxyIP, pc.project, pc.region, model, "Reply with exactly: pong", "")
|
||||
default:
|
||||
code, _, err = cl.Chat(ctx, endpoint, proxyIP, pc.kind, model, "Reply with exactly: pong", "")
|
||||
code, _, err = cl.ChatPrefixed(ctx, endpoint, proxyIP, pc.pathPrefix, pc.kind, model, "Reply with exactly: pong", "")
|
||||
}
|
||||
require.NoError(t, err, "request must reach the proxy for %s", pc.name)
|
||||
return code
|
||||
@@ -164,8 +166,7 @@ func TestModelAllowlistEnforced(t *testing.T) {
|
||||
t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
|
||||
|
||||
require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
|
||||
// Resolve first: the DNS lookup triggers the lazy-connection warm-up, waking
|
||||
// the proxy peer so WaitProxyPeer then observes it connected.
|
||||
// Probe first: the GET resolves the endpoint (DNS error fails) and its first packet wakes the lazy proxy peer, so WaitProxyPeer sees it connected; any HTTP status counts.
|
||||
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
|
||||
require.NoError(t, err, "resolve agent-network endpoint to proxy IP")
|
||||
if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
|
||||
|
||||
@@ -104,8 +104,7 @@ func TestProviderSkipTLSVerification(t *testing.T) {
|
||||
t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
|
||||
|
||||
require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
|
||||
// Resolve first: the DNS lookup triggers the lazy-connection warm-up, waking
|
||||
// the proxy peer so WaitProxyPeer then observes it connected.
|
||||
// Probe first: the GET resolves the endpoint (DNS error fails) and its first packet wakes the lazy proxy peer, so WaitProxyPeer sees it connected; any HTTP status counts.
|
||||
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
|
||||
require.NoError(t, err, "resolve endpoint to proxy IP")
|
||||
if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
|
||||
|
||||
@@ -106,8 +106,7 @@ func TestVLLMProvider(t *testing.T) {
|
||||
t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
|
||||
|
||||
require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
|
||||
// Resolve first: the DNS lookup triggers the lazy-connection warm-up, waking
|
||||
// the proxy peer so WaitProxyPeer then observes it connected.
|
||||
// Probe first: the GET resolves the endpoint (DNS error fails) and its first packet wakes the lazy proxy peer, so WaitProxyPeer sees it connected; any HTTP status counts.
|
||||
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
|
||||
require.NoError(t, err, "resolve endpoint to proxy IP")
|
||||
if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
|
||||
|
||||
@@ -4,6 +4,7 @@ package harness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os/exec"
|
||||
@@ -167,22 +168,54 @@ func (cl *Client) pollStatus(ctx context.Context, timeout time.Duration, want st
|
||||
return fmt.Errorf("timed out waiting for %q; last status:\n%s", want, last)
|
||||
}
|
||||
|
||||
// ResolveProxyIP resolves the agent-network endpoint to the proxy peer's
|
||||
// NetBird IP from inside the client (via magic DNS).
|
||||
const (
|
||||
// curlExitCouldNotResolve is curl's exit code for a DNS resolution failure, distinct from connection-level failures.
|
||||
curlExitCouldNotResolve = 6
|
||||
// dnsProbeRetryWindow bounds DNS-failure retries: the synthesized zone lands a beat after management connects, so early NXDOMAIN is propagation; a zone still absent after this window is a real failure.
|
||||
dnsProbeRetryWindow = 30 * time.Second
|
||||
dnsProbeRetryInterval = 2 * time.Second
|
||||
)
|
||||
|
||||
// ResolveProxyIP GETs https://<endpoint>/ from the client's netns: any HTTP status proves DNS + tunnel and wakes the lazy proxy peer; only DNS failures retry, within dnsProbeRetryWindow. Returns the connected IP for --resolve pinning.
|
||||
func (cl *Client) ResolveProxyIP(ctx context.Context, endpoint string) (string, error) {
|
||||
code, reader, err := cl.container.Exec(ctx, []string{"getent", "hosts", endpoint}, tcexec.Multiplexed())
|
||||
if err != nil {
|
||||
return "", err
|
||||
args := []string{
|
||||
"run", "--rm",
|
||||
"--network", "container:" + cl.container.GetContainerID(),
|
||||
curlImage,
|
||||
"-ksS", "-o", "/dev/null",
|
||||
"--connect-timeout", "30", "--max-time", "60",
|
||||
"-w", "%{remote_ip}",
|
||||
"https://" + endpoint + "/",
|
||||
}
|
||||
out, _ := io.ReadAll(reader)
|
||||
if code != 0 {
|
||||
return "", fmt.Errorf("getent hosts %s exited %d", endpoint, code)
|
||||
deadline := time.Now().Add(dnsProbeRetryWindow)
|
||||
for {
|
||||
cmd := exec.CommandContext(ctx, "docker", args...)
|
||||
var stdout, stderr strings.Builder
|
||||
cmd.Stdout = &stdout
|
||||
cmd.Stderr = &stderr
|
||||
err := cmd.Run()
|
||||
if err == nil {
|
||||
ip := strings.TrimSpace(stdout.String())
|
||||
if ip == "" {
|
||||
return "", fmt.Errorf("got an HTTP response from %s but no remote IP", endpoint)
|
||||
}
|
||||
return ip, nil
|
||||
}
|
||||
|
||||
var exitErr *exec.ExitError
|
||||
if !errors.As(err, &exitErr) || exitErr.ExitCode() != curlExitCouldNotResolve {
|
||||
return "", fmt.Errorf("no HTTP response from %s: %w (%s)", endpoint, err, strings.TrimSpace(stderr.String()))
|
||||
}
|
||||
dnsErr := fmt.Errorf("DNS resolution failed for %s: %s", endpoint, strings.TrimSpace(stderr.String()))
|
||||
if time.Until(deadline) < dnsProbeRetryInterval {
|
||||
return "", dnsErr
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return "", fmt.Errorf("%w (%w)", dnsErr, ctx.Err())
|
||||
case <-time.After(dnsProbeRetryInterval):
|
||||
}
|
||||
}
|
||||
fields := strings.Fields(string(out))
|
||||
if len(fields) == 0 {
|
||||
return "", fmt.Errorf("no address for %s", endpoint)
|
||||
}
|
||||
return fields[0], nil
|
||||
}
|
||||
|
||||
// Wire shapes for Chat.
|
||||
@@ -206,6 +239,17 @@ const (
|
||||
// the wire shape: WireChat (OpenAI) or WireMessages (Anthropic). A non-empty
|
||||
// sessionID is sent as the universal x-session-id header the proxy records.
|
||||
func (cl *Client) Chat(ctx context.Context, endpoint, proxyIP, kind, model, prompt, sessionID string) (int, string, error) {
|
||||
return cl.ChatPrefixed(ctx, endpoint, proxyIP, "", kind, model, prompt, sessionID)
|
||||
}
|
||||
|
||||
// ChatPrefixed is Chat with a base-URL path prefix prepended to the wire
|
||||
// path, mirroring agents whose base URL carries a shape-selecting prefix that
|
||||
// rides through to the upstream — e.g. Claude Code against a Kimi provider
|
||||
// sets ANTHROPIC_BASE_URL=https://<endpoint>/anthropic so the proxy forwards
|
||||
// /anthropic/v1/messages to Moonshot's Anthropic surface while the provider's
|
||||
// upstream URL stays the bare https://api.moonshot.ai. Empty prefix is plain
|
||||
// Chat.
|
||||
func (cl *Client) ChatPrefixed(ctx context.Context, endpoint, proxyIP, pathPrefix, kind, model, prompt, sessionID string) (int, string, error) {
|
||||
var path, body string
|
||||
var headers []string
|
||||
switch kind {
|
||||
@@ -217,7 +261,7 @@ func (cl *Client) Chat(ctx context.Context, endpoint, proxyIP, kind, model, prom
|
||||
path = "/v1/chat/completions"
|
||||
body = fmt.Sprintf(`{"model":%q,"messages":[{"role":"user","content":%q}]}`, model, prompt)
|
||||
}
|
||||
return cl.post(ctx, endpoint, proxyIP, path, body, withSessionID(headers, sessionID))
|
||||
return cl.post(ctx, endpoint, proxyIP, pathPrefix+path, body, withSessionID(headers, sessionID))
|
||||
}
|
||||
|
||||
// Vertex issues an Anthropic-on-Vertex rawPredict POST over the tunnel. Unlike
|
||||
|
||||
@@ -234,9 +234,6 @@ init_environment() {
|
||||
|
||||
NETBIRD_LICENSE_KEY=$(read_secret "Enter license key (input hidden)")
|
||||
|
||||
GHCR_USERNAME="netbirdExtAccess1"
|
||||
GHCR_TOKEN=$(read_secret "Enter GHCR token (input hidden)")
|
||||
|
||||
POSTGRES_USER="netbird"
|
||||
POSTGRES_DB="netbird"
|
||||
POSTGRES_PASSWORD=$(rand_secret)
|
||||
@@ -263,10 +260,6 @@ init_environment() {
|
||||
install -m 600 /dev/null config.yaml
|
||||
render_config_yaml >> config.yaml
|
||||
|
||||
echo "Logging in to ghcr.io ..."
|
||||
printf '%s' "$GHCR_TOKEN" | docker login ghcr.io -u "$GHCR_USERNAME" --password-stdin
|
||||
unset GHCR_TOKEN
|
||||
|
||||
echo ""
|
||||
echo "Pulling images ..."
|
||||
$DOCKER_COMPOSE_COMMAND pull
|
||||
|
||||
@@ -490,8 +490,6 @@ init_migration() {
|
||||
echo ""
|
||||
echo "Step 1: Image swap (community → Enterprise). License key required."
|
||||
NB_LICENSE_KEY=$(read_secret " License key")
|
||||
GHCR_USERNAME="netbirdExtAccess1"
|
||||
GHCR_TOKEN=$(read_secret " GHCR token (input hidden)")
|
||||
|
||||
# Step 2 — optional
|
||||
echo ""
|
||||
@@ -588,11 +586,6 @@ apply_changes() {
|
||||
fi
|
||||
} >> "$ENV_FILE"
|
||||
|
||||
echo ""
|
||||
echo "Logging in to ghcr.io ..."
|
||||
printf '%s' "$GHCR_TOKEN" | docker login ghcr.io -u "$GHCR_USERNAME" --password-stdin
|
||||
unset GHCR_TOKEN
|
||||
|
||||
echo ""
|
||||
echo "Pulling enterprise images ..."
|
||||
$DOCKER_COMPOSE_COMMAND pull
|
||||
|
||||
@@ -420,6 +420,47 @@ var providers = []Provider{
|
||||
{ID: "mistral-embed", Label: "Mistral Embed", InputPer1k: 0.0001, OutputPer1k: 0, ContextWindow: 8192},
|
||||
},
|
||||
},
|
||||
{
|
||||
ID: "kimi_api",
|
||||
Kind: KindProvider,
|
||||
Name: "Kimi (Moonshot AI) API",
|
||||
Description: "Kimi K3 / K2 models via the Moonshot AI platform",
|
||||
DefaultHost: "api.moonshot.ai",
|
||||
AuthHeaderName: "Authorization",
|
||||
AuthHeaderTemplate: "Bearer ${API_KEY}",
|
||||
DefaultContentType: "application/json",
|
||||
BrandColor: "#1A1A2E",
|
||||
// ParserID empty on purpose: Moonshot serves two body shapes on
|
||||
// the same host and key, and the proxy's URL sniffer dispatches
|
||||
// both (same pattern as Bifrost). /v1/chat/completions matches
|
||||
// OpenAIParser; the Anthropic-compatible endpoint the official
|
||||
// Claude Code guide uses (/anthropic/v1/messages) contains
|
||||
// "/v1/messages" and matches AnthropicParser. Pinning "openai"
|
||||
// here would misparse the Claude Code path — the primary way
|
||||
// teams consume Kimi for coding today. Both endpoints accept the
|
||||
// same Moonshot key via Authorization: Bearer (Claude Code's
|
||||
// ANTHROPIC_AUTH_TOKEN rides that header too).
|
||||
//
|
||||
// api.moonshot.ai is the international platform; mainland-China
|
||||
// accounts live on api.moonshot.cn with separate billing —
|
||||
// operators there override the host on the provider record. The
|
||||
// kimi.com subscription coding endpoint (api.kimi.com/coding,
|
||||
// model id "k3") is account-bound seat licensing rather than a
|
||||
// meterable platform key, so it's deliberately not the default.
|
||||
ParserID: "",
|
||||
// Pricing per Moonshot's platform rates at K3 launch (July 2026):
|
||||
// $3/$15 per MTok with $0.30 cached input, flat across the 1M-token
|
||||
// window. kimi-k3 is the ONLY model the platform serves newer
|
||||
// accounts — K2-era ids (kimi-k2-thinking) and even the kimi-latest
|
||||
// alias return resource_not_found_error, verified live 2026-07-21 —
|
||||
// so it's the only catalog entry. Grandfathered accounts with K2
|
||||
// access can still type those ids on the provider's model rows.
|
||||
// The consumer app's "K3 Swarm Max" mode is not an API SKU, so it
|
||||
// doesn't appear here.
|
||||
Models: []Model{
|
||||
{ID: "kimi-k3", Label: "Kimi K3", InputPer1k: 0.003, OutputPer1k: 0.015, ContextWindow: 1000000},
|
||||
},
|
||||
},
|
||||
{
|
||||
ID: "litellm_proxy",
|
||||
Kind: KindGateway,
|
||||
|
||||
@@ -283,8 +283,8 @@ func (a *Account) SynthesizePrivateServiceZones(peerID string) []nbdns.CustomZon
|
||||
// it, adding a single private service would black-hole every
|
||||
// other name under the zone apex.
|
||||
zone = &nbdns.CustomZone{
|
||||
Domain: dns.Fqdn(serviceDomainZone),
|
||||
Records: []nbdns.SimpleRecord{},
|
||||
Domain: dns.Fqdn(serviceDomainZone),
|
||||
Records: []nbdns.SimpleRecord{},
|
||||
NonAuthoritative: true,
|
||||
SearchDomainDisabled: true,
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ COPY encryption ./encryption
|
||||
COPY flow ./flow
|
||||
COPY formatter ./formatter
|
||||
COPY monotime ./monotime
|
||||
COPY management ./management
|
||||
COPY proxy ./proxy
|
||||
COPY route ./route
|
||||
COPY shared ./shared
|
||||
|
||||
@@ -161,6 +161,16 @@ openai:
|
||||
input_per_1k: 0.0001
|
||||
output_per_1k: 0
|
||||
|
||||
# Kimi / Moonshot AI (kimi_api) — OpenAI-compatible /v1 endpoint. Moonshot
|
||||
# reports cache hits OpenAI-style when present; cached input is 10% of
|
||||
# input ($0.30 vs $3.00 per MTok). kimi-k3 is the only model the platform
|
||||
# serves newer accounts (K2-era ids and kimi-latest 404), matching the
|
||||
# management catalog.
|
||||
kimi-k3:
|
||||
input_per_1k: 0.003
|
||||
output_per_1k: 0.015
|
||||
cached_input_per_1k: 0.0003
|
||||
|
||||
anthropic:
|
||||
# Claude 4.x family — cache reads ≈10% of input, cache writes ≈125% of input.
|
||||
# Pricing source: Anthropic's current published rates per million tokens,
|
||||
@@ -206,6 +216,20 @@ anthropic:
|
||||
cache_read_per_1k: 0.0001
|
||||
cache_creation_per_1k: 0.00125
|
||||
|
||||
# Kimi / Moonshot AI (kimi_api) via the Anthropic-compatible endpoint
|
||||
# (/anthropic/v1/messages — the official Claude Code setup). Same rates
|
||||
# as the OpenAI-shape entry above. "kimi-k3[1m]" is the model id some
|
||||
# Claude Code guides set for the 1M-context alias; priced identically so
|
||||
# cost metering doesn't silently skip those requests.
|
||||
kimi-k3:
|
||||
input_per_1k: 0.003
|
||||
output_per_1k: 0.015
|
||||
cache_read_per_1k: 0.0003
|
||||
"kimi-k3[1m]":
|
||||
input_per_1k: 0.003
|
||||
output_per_1k: 0.015
|
||||
cache_read_per_1k: 0.0003
|
||||
|
||||
bedrock:
|
||||
# AWS Bedrock model ids, normalised by the request parser (cross-region
|
||||
# inference-profile prefix + version/throughput suffix stripped), e.g.
|
||||
|
||||
@@ -3,8 +3,8 @@
|
||||
# FreeBSD Port Diff Generator for NetBird
|
||||
#
|
||||
# This script generates the diff file required for submitting a FreeBSD port update.
|
||||
# It works on macOS, Linux, and FreeBSD by fetching files from FreeBSD cgit and
|
||||
# computing checksums from the Go module proxy.
|
||||
# It works on macOS, Linux, and FreeBSD by fetching files from the FreeBSD ports
|
||||
# GitHub mirror and computing checksums from the Go module proxy.
|
||||
#
|
||||
# Usage: ./freebsd-port-diff.sh [new_version]
|
||||
# Example: ./freebsd-port-diff.sh 0.60.7
|
||||
@@ -14,7 +14,7 @@
|
||||
set -e
|
||||
|
||||
GITHUB_REPO="netbirdio/netbird"
|
||||
PORTS_CGIT_BASE="https://cgit.freebsd.org/ports/plain/security/netbird"
|
||||
PORTS_MIRROR_BASE="https://raw.githubusercontent.com/freebsd/freebsd-ports/main/security/netbird"
|
||||
GO_PROXY="https://proxy.golang.org/github.com/netbirdio/netbird/@v"
|
||||
OUTPUT_DIR="${OUTPUT_DIR:-.}"
|
||||
AWK_FIRST_FIELD='{print $1}'
|
||||
@@ -30,10 +30,17 @@ fetch_all_tags() {
|
||||
|
||||
fetch_current_ports_version() {
|
||||
echo "Fetching current version from FreeBSD ports..." >&2
|
||||
curl -sL "${PORTS_CGIT_BASE}/Makefile" 2>/dev/null | \
|
||||
local makefile version
|
||||
makefile=$(fetch_ports_file "Makefile") || return 1
|
||||
version=$(echo "$makefile" | \
|
||||
grep -E "^DISTVERSION=" | \
|
||||
sed 's/DISTVERSION=[[:space:]]*//' | \
|
||||
tr -d '\t '
|
||||
tr -d '\t ')
|
||||
if [[ -z "$version" ]]; then
|
||||
echo "Error: Could not extract DISTVERSION from ports Makefile" >&2
|
||||
return 1
|
||||
fi
|
||||
echo "$version"
|
||||
return 0
|
||||
}
|
||||
|
||||
@@ -45,7 +52,16 @@ fetch_latest_github_release() {
|
||||
|
||||
fetch_ports_file() {
|
||||
local filename="$1"
|
||||
curl -sL "${PORTS_CGIT_BASE}/${filename}" 2>/dev/null
|
||||
local content
|
||||
if ! content=$(curl -fsL --proto '=https' --proto-redir '=https' --retry 3 "${PORTS_MIRROR_BASE}/${filename}" 2>/dev/null); then
|
||||
echo "Error: Could not fetch ${filename} from ${PORTS_MIRROR_BASE}" >&2
|
||||
return 1
|
||||
fi
|
||||
if [[ "$content" == \<* ]]; then
|
||||
echo "Error: Received HTML instead of ${filename} from ${PORTS_MIRROR_BASE}" >&2
|
||||
return 1
|
||||
fi
|
||||
printf '%s' "$content"
|
||||
return 0
|
||||
}
|
||||
|
||||
|
||||
@@ -9,18 +9,22 @@
|
||||
# Example: ./freebsd-port-issue-body.sh 0.56.0 0.59.1
|
||||
#
|
||||
# If no versions are provided, the script will:
|
||||
# - Fetch OLD version from FreeBSD ports cgit (current version in ports tree)
|
||||
# - Fetch OLD version from the FreeBSD ports GitHub mirror (current version in ports tree)
|
||||
# - Fetch NEW version from latest NetBird GitHub release tag
|
||||
|
||||
set -e
|
||||
|
||||
GITHUB_REPO="netbirdio/netbird"
|
||||
PORTS_CGIT_URL="https://cgit.freebsd.org/ports/plain/security/netbird/Makefile"
|
||||
PORTS_MAKEFILE_URL="https://raw.githubusercontent.com/freebsd/freebsd-ports/main/security/netbird/Makefile"
|
||||
|
||||
fetch_current_ports_version() {
|
||||
echo "Fetching current version from FreeBSD ports..." >&2
|
||||
local makefile_content
|
||||
makefile_content=$(curl -sL "$PORTS_CGIT_URL" 2>/dev/null)
|
||||
makefile_content=$(curl -fsL --proto '=https' --proto-redir '=https' --retry 3 "$PORTS_MAKEFILE_URL" 2>/dev/null) || makefile_content=""
|
||||
if [[ "$makefile_content" == \<* ]]; then
|
||||
echo "Error: Received HTML instead of Makefile from ${PORTS_MAKEFILE_URL}" >&2
|
||||
return 1
|
||||
fi
|
||||
if [[ -z "$makefile_content" ]]; then
|
||||
echo "Error: Could not fetch Makefile from FreeBSD ports" >&2
|
||||
return 1
|
||||
|
||||
@@ -187,16 +187,16 @@ func (c *GrpcClient) ready() bool {
|
||||
// Sync wraps the real client's Sync endpoint call and takes care of retries and encryption/decryption of messages
|
||||
// Blocking request. The result will be sent via msgHandler callback function
|
||||
func (c *GrpcClient) Sync(ctx context.Context, sysInfo *system.Info, msgHandler func(msg *proto.SyncResponse) error) error {
|
||||
return c.withMgmtStream(ctx, func(ctx context.Context, serverPubKey wgtypes.Key) error {
|
||||
return c.handleSyncStream(ctx, serverPubKey, sysInfo, msgHandler)
|
||||
return c.withMgmtStream(ctx, func(ctx context.Context, serverPubKey wgtypes.Key, backOff backoff.BackOff) error {
|
||||
return c.handleSyncStream(ctx, serverPubKey, sysInfo, msgHandler, backOff)
|
||||
})
|
||||
}
|
||||
|
||||
// Job wraps the real client's Job endpoint call and takes care of retries and encryption/decryption of messages
|
||||
// Blocking request. The result will be sent via msgHandler callback function
|
||||
func (c *GrpcClient) Job(ctx context.Context, msgHandler func(msg *proto.JobRequest) *proto.JobResponse) error {
|
||||
return c.withMgmtStream(ctx, func(ctx context.Context, serverPubKey wgtypes.Key) error {
|
||||
return c.handleJobStream(ctx, serverPubKey, msgHandler)
|
||||
return c.withMgmtStream(ctx, func(ctx context.Context, serverPubKey wgtypes.Key, backOff backoff.BackOff) error {
|
||||
return c.handleJobStream(ctx, serverPubKey, msgHandler, backOff)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -204,7 +204,7 @@ func (c *GrpcClient) Job(ctx context.Context, msgHandler func(msg *proto.JobRequ
|
||||
// It takes care of retries, connection readiness, and fetching server public key.
|
||||
func (c *GrpcClient) withMgmtStream(
|
||||
ctx context.Context,
|
||||
handler func(ctx context.Context, serverPubKey wgtypes.Key) error,
|
||||
handler func(ctx context.Context, serverPubKey wgtypes.Key, backOff backoff.BackOff) error,
|
||||
) error {
|
||||
backOff := defaultBackoff(ctx)
|
||||
operation := func() error {
|
||||
@@ -224,7 +224,7 @@ func (c *GrpcClient) withMgmtStream(
|
||||
return err
|
||||
}
|
||||
|
||||
return handler(ctx, *serverPubKey)
|
||||
return handler(ctx, *serverPubKey, backOff)
|
||||
}
|
||||
|
||||
err := backoff.Retry(operation, backOff)
|
||||
@@ -239,6 +239,7 @@ func (c *GrpcClient) handleJobStream(
|
||||
ctx context.Context,
|
||||
serverPubKey wgtypes.Key,
|
||||
msgHandler func(msg *proto.JobRequest) *proto.JobResponse,
|
||||
backOff backoff.BackOff,
|
||||
) error {
|
||||
ctx, cancelStream := context.WithCancel(ctx)
|
||||
defer cancelStream()
|
||||
@@ -256,6 +257,19 @@ func (c *GrpcClient) handleJobStream(
|
||||
|
||||
log.Debug("job stream handshake sent successfully")
|
||||
|
||||
// The stream is up, so reset the backoff. This matters for two reasons,
|
||||
// both caused by the backoff lib not resetting its state on a successful
|
||||
// connection:
|
||||
// 1. Without a reset, after a connect followed by an error the next retry
|
||||
// starts from the accumulated (large) interval instead of retrying
|
||||
// promptly, delaying reconnection.
|
||||
// 2. Worse, once the accumulated elapsed time exceeds MaxElapsedTime, the
|
||||
// next stream error makes NextBackOff() return Stop, so the retry loop
|
||||
// exits immediately. That error is then mislabeled unrecoverable and
|
||||
// bubbles up to trigger a full engine restart / data-plane teardown
|
||||
// instead of a silent reconnection.
|
||||
backOff.Reset()
|
||||
|
||||
// Main loop: receive, process, respond
|
||||
for {
|
||||
jobReq, err := c.receiveJobRequest(ctx, stream, serverPubKey)
|
||||
@@ -371,7 +385,7 @@ func (c *GrpcClient) sendJobResponse(
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *GrpcClient) handleSyncStream(ctx context.Context, serverPubKey wgtypes.Key, sysInfo *system.Info, msgHandler func(msg *proto.SyncResponse) error) error {
|
||||
func (c *GrpcClient) handleSyncStream(ctx context.Context, serverPubKey wgtypes.Key, sysInfo *system.Info, msgHandler func(msg *proto.SyncResponse) error, backOff backoff.BackOff) error {
|
||||
ctx, cancelStream := context.WithCancel(ctx)
|
||||
defer cancelStream()
|
||||
|
||||
@@ -390,6 +404,19 @@ func (c *GrpcClient) handleSyncStream(ctx context.Context, serverPubKey wgtypes.
|
||||
c.notifyConnected()
|
||||
c.setSyncStreamConnected()
|
||||
|
||||
// The stream is up, so reset the backoff. This matters for two reasons,
|
||||
// both caused by the backoff lib not resetting its state on a successful
|
||||
// connection:
|
||||
// 1. Without a reset, after a connect followed by an error the next retry
|
||||
// starts from the accumulated (large) interval instead of retrying
|
||||
// promptly, delaying reconnection.
|
||||
// 2. Worse, once the accumulated elapsed time exceeds MaxElapsedTime, the
|
||||
// next stream error makes NextBackOff() return Stop, so the retry loop
|
||||
// exits immediately. That error is then mislabeled unrecoverable and
|
||||
// bubbles up to trigger a full engine restart / data-plane teardown
|
||||
// instead of a silent reconnection.
|
||||
backOff.Reset()
|
||||
|
||||
// blocking until error
|
||||
err = c.receiveUpdatesEvents(stream, serverPubKey, msgHandler)
|
||||
if err != nil {
|
||||
|
||||
@@ -21,9 +21,9 @@ import (
|
||||
"net/netip"
|
||||
|
||||
nbpeer "github.com/netbirdio/netbird/management/server/peer"
|
||||
"github.com/netbirdio/netbird/shared/management/types"
|
||||
nbroute "github.com/netbirdio/netbird/route"
|
||||
"github.com/netbirdio/netbird/shared/management/proto"
|
||||
"github.com/netbirdio/netbird/shared/management/types"
|
||||
"github.com/netbirdio/netbird/shared/netiputil"
|
||||
"github.com/netbirdio/netbird/shared/sshauth"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user