m/rdp-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix VPN fabric-only routing guard

Browse Source
This commit is contained in:
Mikhail
2026-05-14 23:26:19 +03:00
parent 8f69d53193
commit 26cb65e936
5 changed files with 260 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agents/rap-node-agent/cmd/rap-node-agent/main.go
+40 -31
View File
@@ -85,7 +85,15 @@ func main() {
}
defer stopMeshEndpoint()
supervisor := supervisor.StubSupervisor{Version: agent.Version}
supervisor := supervisor.StubSupervisor{
Version: agent.Version,
RemoteWorkspaceRealAdapter: supervisor.RemoteWorkspaceRealAdapterConfig{
EnabledRequested: cfg.RemoteWorkspaceRealAdapterEnabled,
Command: cfg.RemoteWorkspaceRealAdapterCommand,
ArgsJSON: cfg.RemoteWorkspaceRealAdapterArgsJSON,
WorkDir: cfg.RemoteWorkspaceRealAdapterWorkDir,
},
}
startedAt := time.Now().UTC()
ticker := time.NewTicker(cfg.HeartbeatInterval)
defer ticker.Stop()
@@ -805,6 +813,7 @@ func newVPNFabricIngress(meshState *syntheticMeshState, identity state.Identity,
if ingress == nil {
ingress = &vpnruntime.FabricClientPacketIngress{}
}
ingress.PreventLastRouteWithdrawal = true
ingress.UpdateRuntime(
meshState.ProductionForwardTransport,
meshState.VPNFabricInbox,
@@ -2484,8 +2493,9 @@ func fabricServiceChannelRuntimeReport(meshState *syntheticMeshState, identity s
"service_class": "vpn_packets",
"channel_class": mesh.ProductionChannelVPNPacket,
"route_manager": "primary_sticky_with_alternate_route_failover",
"backend_relay_fallback": true,
"backend_relay_fallback_position": "after_all_fabric_routes_fail",
"backend_relay_fallback": false,
"backend_relay_fallback_position": "disabled_farm_owned_dataplane",
"route_authority": "fabric_farm",
"application_protocol_agnostic": true,
"observed_at": observedAt.UTC().Format(time.RFC3339Nano),
}
@@ -4191,6 +4201,24 @@ func ensureVPNGatewayRuntime(ctx context.Context, api *client.Client, identity s
}
activeOwner := false
for _, assignment := range assignments {
if assignment.AssignmentReason == "eligible_candidate" && assignment.DesiredState == "enabled" {
lease, err := api.AcquireNodeVPNAssignmentLease(ctx, identity.ClusterID, identity.NodeID, assignment.VPNConnectionID, client.NodeVPNAssignmentLeaseAcquireRequest{
TTLSeconds: 300,
Metadata: map[string]any{
"reason": "node_agent_auto_acquire",
"node_id": identity.NodeID,
"agent": "rap-node-agent",
"acquired_at": time.Now().UTC().Format(time.RFC3339Nano),
},
})
if err != nil {
log.Printf("vpn assignment lease auto-acquire skipped: vpn_connection_id=%s error=%v", assignment.VPNConnectionID, err)
} else if lease != nil {
assignment.AssignmentReason = "active_owner"
assignment.ActiveLease = lease
log.Printf("vpn assignment lease auto-acquired: vpn_connection_id=%s lease_id=%s", assignment.VPNConnectionID, lease.LeaseID)
}
}
if assignment.AssignmentReason != "active_owner" {
continue
}
@@ -4220,6 +4248,11 @@ func ensureVPNGatewayRuntime(ctx context.Context, api *client.Client, identity s
} else if _, ok := gateway.Transport.(*vpnruntime.AdaptivePacketTransport); ok {
gateway.Stop()
gateway.Transport = nil
} else {
gateway.Stop()
gateway.Transport = nil
log.Printf("vpn gateway runtime skipped: vpn_connection_id=%s reason=fabric_packet_transport_unavailable", assignment.VPNConnectionID)
return nil
}
if err := gateway.EnsureStarted(ctx); err != nil {
return err
@@ -4236,29 +4269,17 @@ func ensureVPNGatewayRuntime(ctx context.Context, api *client.Client, identity s
return nil
}
func localGatewayTransportForAssignment(identity state.Identity, assignment client.NodeVPNAssignment, meshState *syntheticMeshState, api *client.Client) vpnruntime.PacketTransport {
func localGatewayTransportForAssignment(identity state.Identity, assignment client.NodeVPNAssignment, meshState *syntheticMeshState, _ *client.Client) vpnruntime.PacketTransport {
if meshState == nil || meshState.VPNFabricInbox == nil || assignment.VPNConnectionID == "" {
return nil
}
local := &vpnruntime.LocalPacketTransport{
return &vpnruntime.LocalPacketTransport{
Inbox: meshState.VPNFabricInbox,
VPNConnectionID: assignment.VPNConnectionID,
}
if api == nil {
return local
}
return &vpnruntime.AdaptivePacketTransport{
Primary: local,
Fallback: vpnruntime.BackendPacketTransport{
API: api,
ClusterID: identity.ClusterID,
VPNConnectionID: assignment.VPNConnectionID,
},
PrimaryTimeout: 50 * time.Millisecond,
}
}
func fabricGatewayTransportForAssignment(identity state.Identity, assignment client.NodeVPNAssignment, meshState *syntheticMeshState, api *client.Client) vpnruntime.PacketTransport {
func fabricGatewayTransportForAssignment(identity state.Identity, assignment client.NodeVPNAssignment, meshState *syntheticMeshState, _ *client.Client) vpnruntime.PacketTransport {
if meshState == nil || meshState.ProductionForwardTransport == nil || meshState.VPNFabricInbox == nil {
return nil
}
@@ -4266,7 +4287,7 @@ func fabricGatewayTransportForAssignment(identity state.Identity, assignment cli
if !ok {
return nil
}
fabric := &vpnruntime.FabricPacketTransport{
return &vpnruntime.FabricPacketTransport{
ForwardTransport: meshState.ProductionForwardTransport,
Inbox: meshState.VPNFabricInbox,
ClusterID: identity.ClusterID,
@@ -4279,18 +4300,6 @@ func fabricGatewayTransportForAssignment(identity state.Identity, assignment cli
SendDirection: vpnruntime.FabricDirectionGatewayToClient,
ReceiveDirection: vpnruntime.FabricDirectionClientToGateway,
}
if api == nil {
return fabric
}
return &vpnruntime.AdaptivePacketTransport{
Primary: fabric,
Fallback: vpnruntime.BackendPacketTransport{
API: api,
ClusterID: identity.ClusterID,
VPNConnectionID: assignment.VPNConnectionID,
},
PrimaryTimeout: 50 * time.Millisecond,
}
}
func selectVPNPacketRoute(routes []mesh.SyntheticRoute, clusterID string, localNodeID string) (mesh.SyntheticRoute, string, bool) {
agents/rap-node-agent/cmd/rap-node-agent/main_test.go
+44
View File
@@ -20,6 +20,7 @@ import (
"github.com/example/remote-access-platform/agents/rap-node-agent/internal/config"
"github.com/example/remote-access-platform/agents/rap-node-agent/internal/mesh"
"github.com/example/remote-access-platform/agents/rap-node-agent/internal/state"
"github.com/example/remote-access-platform/agents/rap-node-agent/internal/vpnruntime"
)
func TestLoadSyntheticMeshConfigPrefersScopedFile(t *testing.T) {
@@ -197,6 +198,49 @@ func TestRouteManagerDecisionsFromControlPlaneRejectsGuardedRemediationCommand(t
}
}
func TestGatewayTransportForAssignmentUsesFabricWithoutBackendFallback(t *testing.T) {
inbox := vpnruntime.NewFabricPacketInbox(4)
transport := fabricGatewayTransportForAssignment(
state.Identity{ClusterID: "cluster-1", NodeID: "exit-1"},
client.NodeVPNAssignment{VPNConnectionID: "vpn-1"},
&syntheticMeshState{
ProductionForwardTransport: noopProductionForwardTransport{},
VPNFabricInbox: inbox,
Routes: []mesh.SyntheticRoute{{
RouteID: "route-exit-entry",
ClusterID: "cluster-1",
SourceNodeID: "exit-1",
DestinationNodeID: "entry-1",
Hops: []string{"exit-1", "entry-1"},
AllowedChannels: []string{mesh.ProductionChannelVPNPacket},
ExpiresAt: time.Now().UTC().Add(time.Minute),
}},
},
nil,
)
if _, ok := transport.(*vpnruntime.FabricPacketTransport); !ok {
t.Fatalf("transport = %T, want fabric packet transport without backend fallback", transport)
}
}
func TestLocalGatewayTransportForAssignmentUsesLocalInboxWithoutBackendFallback(t *testing.T) {
transport := localGatewayTransportForAssignment(
state.Identity{ClusterID: "cluster-1", NodeID: "exit-1"},
client.NodeVPNAssignment{VPNConnectionID: "vpn-1"},
&syntheticMeshState{VPNFabricInbox: vpnruntime.NewFabricPacketInbox(4)},
nil,
)
if _, ok := transport.(*vpnruntime.LocalPacketTransport); !ok {
t.Fatalf("transport = %T, want local packet transport without backend fallback", transport)
}
}
type noopProductionForwardTransport struct{}
func (noopProductionForwardTransport) SendProduction(context.Context, string, mesh.ProductionEnvelope) (mesh.ProductionForwardResult, error) {
return mesh.ProductionForwardResult{}, nil
}
func TestRouteManagerDecisionsFromControlPlaneKeepsExplicitRemediationCommand(t *testing.T) {
now := time.Now().UTC()
report := &client.RoutePathDecisionReport{Decisions: []client.RoutePathDecision{{
agents/rap-node-agent/internal/agent/payload.go
+9 -3
View File
@@ -7,7 +7,7 @@ import (
"github.com/example/remote-access-platform/agents/rap-node-agent/internal/state"
)
const Version = "0.2.256-c18z82"
const Version = "0.2.267-vpnfarmonly"
func EnrollmentPayload(clusterID, joinToken string, identity state.Identity) client.EnrollRequest {
return client.EnrollRequest{
@@ -35,7 +35,10 @@ func EnrollmentPayload(clusterID, joinToken string, identity state.Identity) cli
"vpn_assignment_status": true,
"vpn_packet_forwarding": true,
"vpn_fabric_packet_transport": true,
"vpn_local_gateway_shortcut": true,
"vpn_local_gateway_shortcut": false,
"vpn_farm_owned_dataplane": true,
"vpn_backend_relay_fallback": false,
"fabric_service_channel_required": true,
"external_backend_entry_proxy": true,
},
ReportedFacts: map[string]any{
@@ -59,7 +62,10 @@ func HeartbeatPayload() client.HeartbeatRequest {
"vpn_assignment_status": true,
"vpn_packet_forwarding": true,
"vpn_fabric_packet_transport": true,
"vpn_local_gateway_shortcut": true,
"vpn_local_gateway_shortcut": false,
"vpn_farm_owned_dataplane": true,
"vpn_backend_relay_fallback": false,
"fabric_service_channel_required": true,
"external_backend_entry_proxy": true,
},
ServiceStates: map[string]any{
agents/rap-node-agent/internal/vpnruntime/fabric_transport.go
+41 -5
View File
@@ -48,10 +48,12 @@ type FabricClientPacketIngress struct {
Inbox *FabricPacketInbox
Routes func() []mesh.SyntheticRoute
LocalGateway func(vpnConnectionID string) bool
AllowLegacyLocalGatewayFallback bool
FlowScheduler *FabricFlowScheduler
MaxParallelFlowSends int
RecoveryPolicyFingerprint string
AdaptivePolicyFingerprint string
PreventLastRouteWithdrawal bool
ClusterID string
LocalNodeID string
@@ -1623,7 +1625,7 @@ func (i *FabricClientPacketIngress) ReceiveClientPacketBatch(ctx context.Context
}
func (i *FabricClientPacketIngress) localGatewayReady(vpnConnectionID string) bool {
if i == nil || i.inbox() == nil || vpnConnectionID == "" {
if i == nil || !i.AllowLegacyLocalGatewayFallback || i.inbox() == nil || vpnConnectionID == "" {
return false
}
localGateway := i.localGateway()
@@ -1669,6 +1671,7 @@ func (i *FabricClientPacketIngress) routeCandidatesWithPreference(clusterID stri
var preferred []fabricClientRouteCandidate
var alternates []fabricClientRouteCandidate
var deferred []fabricClientRouteCandidate
var withdrawn []fabricClientRouteCandidate
manager := i.routeManager()
if preferredRouteID != "" && manager.isWithdrawn(preferredRouteID) {
if replacementRouteID := manager.replacementRouteID(preferredRouteID); replacementRouteID != "" {
@@ -1684,9 +1687,6 @@ func (i *FabricClientPacketIngress) routeCandidatesWithPreference(clusterID stri
if route.ClusterID != clusterID || route.SourceNodeID != localNodeID || !containsString(route.AllowedChannels, mesh.ProductionChannelVPNPacket) {
continue
}
if manager.isWithdrawn(route.RouteID) {
continue
}
if !route.ExpiresAt.IsZero() && !route.ExpiresAt.After(now) {
continue
}
@@ -1695,6 +1695,10 @@ func (i *FabricClientPacketIngress) routeCandidatesWithPreference(clusterID stri
continue
}
candidate := fabricClientRouteCandidate{Route: route, NextHop: nextHop}
if manager.isWithdrawn(route.RouteID) {
withdrawn = append(withdrawn, candidate)
continue
}
if preferredRouteID != "" && route.RouteID == preferredRouteID {
preferred = append(preferred, candidate)
} else if avoidRouteID != "" && route.RouteID == avoidRouteID {
@@ -1703,9 +1707,32 @@ func (i *FabricClientPacketIngress) routeCandidatesWithPreference(clusterID stri
alternates = append(alternates, candidate)
}
}
if len(preferred) > 0 {
destinationNodeID := strings.TrimSpace(preferred[0].Route.DestinationNodeID)
alternates = filterRouteCandidatesByDestination(alternates, destinationNodeID)
deferred = filterRouteCandidatesByDestination(deferred, destinationNodeID)
}
out := append(preferred, alternates...)
out = i.applyRouteQualityPreferences(out, preferredRouteID)
return append(out, deferred...)
out = append(out, deferred...)
if len(out) == 0 && i.preventLastRouteWithdrawal() {
return withdrawn
}
return out
}
func filterRouteCandidatesByDestination(candidates []fabricClientRouteCandidate, destinationNodeID string) []fabricClientRouteCandidate {
destinationNodeID = strings.TrimSpace(destinationNodeID)
if destinationNodeID == "" || len(candidates) == 0 {
return candidates
}
out := candidates[:0]
for _, candidate := range candidates {
if strings.TrimSpace(candidate.Route.DestinationNodeID) == destinationNodeID {
out = append(out, candidate)
}
}
return out
}
func (i *FabricClientPacketIngress) applyRouteQualityPreferences(candidates []fabricClientRouteCandidate, preferredRouteID string) []fabricClientRouteCandidate {
@@ -1744,6 +1771,15 @@ func (i *FabricClientPacketIngress) applyRouteQualityPreferences(candidates []fa
return out
}
func (i *FabricClientPacketIngress) preventLastRouteWithdrawal() bool {
if i == nil {
return false
}
i.mu.Lock()
defer i.mu.Unlock()
return i.PreventLastRouteWithdrawal
}
func (t *FabricPacketTransport) ReceiveGatewayPacketBatch(ctx context.Context, timeout time.Duration) ([][]byte, error) {
if t == nil || t.Inbox == nil {
return nil, mesh.ErrForwardRuntimeUnavailable
agents/rap-node-agent/internal/vpnruntime/fabric_transport_test.go
+86
View File
@@ -524,6 +524,52 @@ func TestFabricClientPacketIngressTriesAlternateRouteBeforeBackendFallback(t *te
}
}
func TestFabricClientPacketIngressDoesNotFailOverPreferredRouteToDifferentDestination(t *testing.T) {
transport := &failoverProductionTransport{failNextHop: "relay-home"}
ingress := &FabricClientPacketIngress{
ForwardTransport: transport,
Inbox: NewFabricPacketInbox(4),
ClusterID: "cluster-1",
LocalNodeID: "entry-1",
Routes: func() []mesh.SyntheticRoute {
return []mesh.SyntheticRoute{
{
RouteID: "route-other",
ClusterID: "cluster-1",
SourceNodeID: "entry-1",
DestinationNodeID: "ifcm-1",
Hops: []string{"entry-1", "relay-ifcm", "ifcm-1"},
AllowedChannels: []string{mesh.ProductionChannelVPNPacket},
ExpiresAt: time.Now().UTC().Add(time.Minute),
MaxTTL: 8,
},
{
RouteID: "route-home",
ClusterID: "cluster-1",
SourceNodeID: "entry-1",
DestinationNodeID: "home-1",
Hops: []string{"entry-1", "relay-home", "home-1"},
AllowedChannels: []string{mesh.ProductionChannelVPNPacket},
ExpiresAt: time.Now().UTC().Add(time.Minute),
MaxTTL: 8,
},
}
},
}
ingress.PreferClientRoute("route-home")
err := ingress.SendClientPacketBatch(context.Background(), "cluster-1", "vpn-1", [][]byte{[]byte("packet")})
if err == nil {
t.Fatal("send client packet batch succeeded after preferred route failure; want failure without cross-destination fallback")
}
if len(transport.calls) != 1 || transport.calls[0] != "relay-home" {
t.Fatalf("route attempts = %#v, want only relay-home", transport.calls)
}
if transport.envelope.RouteID == "route-other" {
t.Fatalf("cross-destination route was used: %+v", transport.envelope)
}
}
func TestFabricClientPacketIngressAvoidsChannelFailedRouteOnNextSend(t *testing.T) {
transport := &captureManyProductionTransport{}
scheduler := NewFabricFlowScheduler(8, 16)
@@ -822,6 +868,44 @@ func TestFabricClientPacketIngressPendingDegradedFallbackWithdrawsRouteWithoutAl
}
}
func TestFabricClientPacketIngressKeepsLastRouteWhenWithdrawalPreventionEnabled(t *testing.T) {
transport := &captureManyProductionTransport{}
ingress := &FabricClientPacketIngress{
ForwardTransport: transport,
Inbox: NewFabricPacketInbox(4),
ClusterID: "cluster-1",
LocalNodeID: "entry-1",
PreventLastRouteWithdrawal: true,
Routes: func() []mesh.SyntheticRoute {
return []mesh.SyntheticRoute{{
RouteID: "route-only",
ClusterID: "cluster-1",
SourceNodeID: "entry-1",
DestinationNodeID: "exit-1",
Hops: []string{"entry-1", "exit-1"},
AllowedChannels: []string{mesh.ProductionChannelVPNPacket},
ExpiresAt: time.Now().UTC().Add(time.Minute),
MaxTTL: 8,
}}
},
}
ingress.UpdateRouteManager([]FabricServiceChannelRouteManagerDecision{{
RouteID: "route-only",
RebuildStatus: "pending_degraded_fallback",
DecisionSource: "service_channel_feedback_no_alternate",
}}, "config-v2", time.Now().UTC())
if err := ingress.SendClientPacketBatch(context.Background(), "cluster-1", "vpn-1", [][]byte{[]byte("packet")}); err != nil {
t.Fatalf("send client packet batch: %v", err)
}
if len(transport.envelopes) != 1 || transport.envelopes[0].RouteID != "route-only" {
t.Fatalf("envelopes = %+v, want preserved last route", transport.envelopes)
}
if snapshot := ingress.Snapshot("cluster-1"); snapshot.RouteCandidateCount != 1 {
t.Fatalf("route candidate count = %d, want last withdrawn route preserved", snapshot.RouteCandidateCount)
}
}
func TestFabricClientPacketIngressMarksChannelForRebuildAfterRepeatedRouteFailures(t *testing.T) {
transport := &failoverProductionTransport{failNextHop: "relay-bad"}
scheduler := NewFabricFlowScheduler(8, 16)
@@ -1933,6 +2017,7 @@ func TestFabricClientPacketIngressUsesLocalGatewayShortcutWithoutRoute(t *testin
Inbox: inbox,
ClusterID: "cluster-1",
LocalNodeID: "entry-1",
AllowLegacyLocalGatewayFallback: true,
LocalGateway: func(vpnConnectionID string) bool {
return vpnConnectionID == "vpn-1"
},
@@ -1957,6 +2042,7 @@ func TestFabricClientPacketIngressReceivesLocalGatewayReplyWithoutRoute(t *testi
Inbox: inbox,
ClusterID: "cluster-1",
LocalNodeID: "entry-1",
AllowLegacyLocalGatewayFallback: true,
LocalGateway: func(vpnConnectionID string) bool {
return vpnConnectionID == "vpn-1"
},