From baf4073f687399af72ca6fa2099d5d6739a0769b Mon Sep 17 00:00:00 2001 From: sechmachine <97589681+sechmachine727@users.noreply.github.com> Date: Thu, 30 Jul 2026 11:15:48 +0700 Subject: [PATCH] fix(gateway): secure control and terminal ownership --- cmd/verse-gateway/main_test.go | 2 +- gateway/apollo_native.go | 30 +++- gateway/apollo_native_test.go | 28 +++ gateway/control_plane.go | 18 +- gateway/control_plane_test.go | 94 ++++++++++ gateway/feedback.go | 12 ++ gateway/gateway_test.go | 170 +++++++++++++++++- gateway/transport.go | 54 ++++-- go.mod | 2 +- go.sum | 2 + .../gateway-terminal-receipt/.openspec.yaml | 2 + .../gateway-terminal-receipt/design.md | 29 +++ .../gateway-terminal-receipt/proposal.md | 24 +++ .../specs/provider-session-lifecycle/spec.md | 23 +++ .../changes/gateway-terminal-receipt/tasks.md | 16 ++ .../specs/provider-session-lifecycle/spec.md | 18 +- 16 files changed, 488 insertions(+), 36 deletions(-) create mode 100644 gateway/control_plane_test.go create mode 100644 openspec/changes/gateway-terminal-receipt/.openspec.yaml create mode 100644 openspec/changes/gateway-terminal-receipt/design.md create mode 100644 openspec/changes/gateway-terminal-receipt/proposal.md create mode 100644 openspec/changes/gateway-terminal-receipt/specs/provider-session-lifecycle/spec.md create mode 100644 openspec/changes/gateway-terminal-receipt/tasks.md diff --git a/cmd/verse-gateway/main_test.go b/cmd/verse-gateway/main_test.go index 4715dc7..46c6e22 100644 --- a/cmd/verse-gateway/main_test.go +++ b/cmd/verse-gateway/main_test.go @@ -21,7 +21,7 @@ import ( func TestHeartbeatReportsMeasuredEgressInsteadOfConfiguredCapacity(t *testing.T) { heartbeats := make(chan protocol.GatewayHeartbeat, 1) - control := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + control := httptest.NewTLSServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { if request.URL.Path == "/api/v1/gateway/heartbeat" { heartbeat, err := protocol.DecodeGatewayHeartbeat(mustReadBody(t, request)) if err != nil { diff --git a/gateway/apollo_native.go b/gateway/apollo_native.go index 3c6609c..39d04a8 100644 --- a/gateway/apollo_native.go +++ b/gateway/apollo_native.go @@ -289,6 +289,7 @@ type nativeApolloSession struct { audio chan ProviderMedia events chan ProviderEvent mu sync.Mutex + eventMu sync.Mutex mediaMu sync.Mutex controlMu sync.Mutex state protocol.ProviderState @@ -686,10 +687,32 @@ func (s *nativeApolloSession) handleApolloControlPayload(_ uint8, _ bool, payloa } func (s *nativeApolloSession) emitProviderEvent(event ProviderEvent) { + if !s.enqueueProviderEvent(event) { + s.handleApolloDisconnect(ErrProviderMalformed) + } +} + +func (s *nativeApolloSession) enqueueProviderEvent(event ProviderEvent) bool { + s.eventMu.Lock() + defer s.eventMu.Unlock() select { case s.events <- event: + return true default: - s.handleApolloDisconnect(ErrProviderMalformed) + } + if event.Kind != ProviderEventTerminated && event.Kind != ProviderEventDisconnected { + return false + } + select { + case <-s.events: + default: + return false + } + select { + case s.events <- event: + return true + default: + return false } } @@ -706,10 +729,7 @@ func (s *nativeApolloSession) handleApolloDisconnect(err error) { } s.state.State = ProviderStateDisconnected s.mu.Unlock() - select { - case s.events <- ProviderEvent{Kind: ProviderEventDisconnected}: - default: - } + s.enqueueProviderEvent(ProviderEvent{Kind: ProviderEventDisconnected}) }) } diff --git a/gateway/apollo_native_test.go b/gateway/apollo_native_test.go index 2ec1e73..c45ed24 100644 --- a/gateway/apollo_native_test.go +++ b/gateway/apollo_native_test.go @@ -858,6 +858,34 @@ func TestNativeApolloSessionForwardsEncryptedHostFeedback(t *testing.T) { } } +func TestApolloTerminalEventSurvivesFullFeedbackQueue(t *testing.T) { + session := newNativeApolloSession("session-full-feedback") + key := []byte("0123456789abcdef") + codec, err := newApolloControlCodec(key) + if err != nil { + t.Fatal(err) + } + session.control = codec + for range cap(session.events) { + session.events <- ProviderEvent{Kind: ProviderEventRumble, Payload: []byte{1, 0, 0, 0, 0}} + } + + session.handleApolloControlPayload( + apolloChannelGeneric, + true, + sourceSealHostControl(t, key, 0, apolloControlTypeTerm, []byte{1, 2, 3, 4}), + ) + + foundTerminal := false + for range cap(session.events) { + event := <-session.events + foundTerminal = foundTerminal || event.Kind == ProviderEventTerminated + } + if !foundTerminal { + t.Fatal("encrypted host termination was lost behind a full feedback queue") + } +} + func TestPushLatestDropsExactlyOneOldPayload(t *testing.T) { queue := make(chan []byte, 1) if dropped := pushLatest(queue, []byte("old")); dropped { diff --git a/gateway/control_plane.go b/gateway/control_plane.go index ecdf5f1..8cb6ea3 100644 --- a/gateway/control_plane.go +++ b/gateway/control_plane.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "net/http" + "net/url" "strings" protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol" @@ -110,12 +111,25 @@ func (c *ControlPlaneClient) post(ctx context.Context, path string, payload []by if c == nil || c.HTTPClient == nil || c.BaseURL == "" { return nil, errors.New("control-plane client is not configured") } - request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.BaseURL+path, bytes.NewReader(payload)) + baseURL, err := url.Parse(c.BaseURL) + if err != nil || baseURL.Scheme != "https" || baseURL.Host == "" { + return nil, errors.New("control-plane base URL must be absolute HTTPS") + } + requestURL := *baseURL + requestURL.Path = strings.TrimRight(baseURL.Path, "/") + path + requestURL.RawPath = "" + requestURL.RawQuery = "" + requestURL.Fragment = "" + request, err := http.NewRequestWithContext(ctx, http.MethodPost, requestURL.String(), bytes.NewReader(payload)) if err != nil { return nil, err } request.Header.Set("Content-Type", "application/json") - response, err := c.HTTPClient.Do(request) + client := *c.HTTPClient + client.CheckRedirect = func(*http.Request, []*http.Request) error { + return errors.New("control-plane redirects are not permitted") + } + response, err := client.Do(request) if err != nil { return nil, err } diff --git a/gateway/control_plane_test.go b/gateway/control_plane_test.go new file mode 100644 index 0000000..6c0ae31 --- /dev/null +++ b/gateway/control_plane_test.go @@ -0,0 +1,94 @@ +package gateway + +import ( + "context" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol" +) + +func TestControlPlaneRejectsHTTPBeforeSending(t *testing.T) { + var requests atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + requests.Add(1) + })) + defer server.Close() + + err := NewControlPlaneClient(server.URL, server.Client()).Heartbeat( + context.Background(), + validControlPlaneHeartbeat("gateway-1"), + ) + if err == nil { + t.Fatal("HTTP control-plane URL was accepted") + } + if got := requests.Load(); got != 0 { + t.Fatalf("HTTP control-plane received %d requests, want 0", got) + } +} + +func TestControlPlaneRejectsHTTPSRedirectToHTTPWithoutDisclosure(t *testing.T) { + var downgradeRequests atomic.Int64 + downgrade := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + downgradeRequests.Add(1) + })) + defer downgrade.Close() + + serverTLS, clientTLS := testTLS(t) + source := httptest.NewUnstartedServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + if len(request.TLS.PeerCertificates) == 0 { + t.Error("source did not authenticate the client certificate") + } + http.Redirect(response, request, downgrade.URL+"/capture", http.StatusTemporaryRedirect) + })) + source.TLS = serverTLS + source.StartTLS() + defer source.Close() + + client := &http.Client{Transport: &http.Transport{TLSClientConfig: clientTLS}} + err := NewControlPlaneClient(source.URL, client).Heartbeat( + context.Background(), + validControlPlaneHeartbeat("gateway-secret"), + ) + if err == nil { + t.Fatal("HTTPS-to-HTTP redirect was accepted") + } + if got := downgradeRequests.Load(); got != 0 { + t.Fatalf("downgrade target received %d requests, want 0", got) + } +} + +func TestControlPlanePinnedMTLSRemainsFunctional(t *testing.T) { + serverTLS, clientTLS := testTLS(t) + var authenticated atomic.Bool + server := httptest.NewUnstartedServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + authenticated.Store(request.TLS != nil && len(request.TLS.PeerCertificates) > 0) + response.WriteHeader(http.StatusNoContent) + })) + server.TLS = serverTLS + server.StartTLS() + defer server.Close() + + client := &http.Client{Transport: &http.Transport{TLSClientConfig: clientTLS}} + if err := NewControlPlaneClient(server.URL, client).Heartbeat( + context.Background(), + validControlPlaneHeartbeat("gateway-1"), + ); err != nil { + t.Fatalf("pinned mTLS heartbeat: %v", err) + } + if !authenticated.Load() { + t.Fatal("server did not authenticate the client certificate") + } +} + +func validControlPlaneHeartbeat(gatewayID string) protocol.GatewayHeartbeat { + return protocol.GatewayHeartbeat{ + Version: "1", GatewayID: gatewayID, Sequence: 1, + ObservedAt: time.Unix(1, 0).UTC().Format(time.RFC3339Nano), + State: "ready", + Telemetry: protocol.GatewayTelemetry{MediaPackets: 1, ProviderState: "ready"}, + } +} diff --git a/gateway/feedback.go b/gateway/feedback.go index 71d882e..20f3d83 100644 --- a/gateway/feedback.go +++ b/gateway/feedback.go @@ -7,6 +7,7 @@ type FeedbackKind uint8 const ( FeedbackIDR FeedbackKind = iota + 1 FeedbackFEC + FeedbackTerminalReceipt ) const ( @@ -15,6 +16,7 @@ const ( gatewayFeedbackGateway = 1 gatewayFeedbackIDR = 1 gatewayFeedbackFEC = 2 + gatewayFeedbackTerminalAck = 3 gatewayFeedbackTerminated = 0x10 gatewayFeedbackRumble = 0x11 gatewayFeedbackHDR = 0x12 @@ -67,6 +69,11 @@ func EncodeClientFeedback(feedback Feedback) ([]byte, error) { if !validGatewayFECStatus(feedback.Payload) { return nil, ErrProviderMalformed } + case FeedbackTerminalReceipt: + kind = gatewayFeedbackTerminalAck + if len(feedback.Payload) != 0 { + return nil, ErrProviderMalformed + } default: return nil, ErrProviderMalformed } @@ -89,6 +96,11 @@ func DecodeClientFeedback(data []byte) (Feedback, error) { return Feedback{}, ErrProviderMalformed } return Feedback{Kind: FeedbackFEC, Payload: message.payload}, nil + case gatewayFeedbackTerminalAck: + if len(message.payload) != 0 { + return Feedback{}, ErrProviderMalformed + } + return Feedback{Kind: FeedbackTerminalReceipt}, nil default: return Feedback{}, ErrProviderMalformed } diff --git a/gateway/gateway_test.go b/gateway/gateway_test.go index f429815..1a3e7de 100644 --- a/gateway/gateway_test.go +++ b/gateway/gateway_test.go @@ -9,8 +9,10 @@ import ( "crypto/x509" "crypto/x509/pkix" "encoding/base64" + "encoding/binary" "encoding/hex" "errors" + "io" "math/big" "net" "os" @@ -21,6 +23,8 @@ import ( "testing" "time" + "github.com/quic-go/quic-go" + protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol" ) @@ -105,6 +109,10 @@ func TestClientFeedbackUsesFixedProtocolVGFVector(t *testing.T) { if event, err := DecodeProviderEvent(disconnected); err != nil || event.Kind != ProviderEventDisconnected { t.Fatalf("decoded disconnected event = %#v, %v", event, err) } + receipt, err := EncodeClientFeedback(Feedback{Kind: FeedbackTerminalReceipt}) + if err != nil || hex.EncodeToString(receipt) != "5647463100030000" { + t.Fatalf("terminal receipt vector = %x, %v", receipt, err) + } } func TestCapabilityIntersectionAndBoundedQueue(t *testing.T) { @@ -179,12 +187,6 @@ func TestSyntheticImpairmentPacingAndResourceBounds(t *testing.T) { if delivered != 3 || deliveredBytes != 1179*3 { t.Fatalf("synthetic impairment delivered=%d bytes=%d", delivered, deliveredBytes) } - pacer := NewPacer(1) - ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond) - defer cancel() - if err := pacer.Wait(ctx, 100); !errors.Is(err, context.DeadlineExceeded) { - t.Fatalf("pacer ignored bounded context: %v", err) - } } func TestApolloFixturesAndLifecycle(t *testing.T) { @@ -701,6 +703,11 @@ func TestProviderTerminationEndsPublicGatewaySession(t *testing.T) { if states := h.reporter.States(); len(states) == 0 || states[len(states)-1].State != ProviderStateTerminated { t.Fatalf("provider states = %#v", states) } + h.session.mu.Lock() + if len(h.session.feedback) != 0 { + t.Fatalf("terminal receipt reached provider: %#v", h.session.feedback) + } + h.session.mu.Unlock() h.assertMediaClosed(t) } @@ -716,6 +723,7 @@ func TestEncryptedNativeHostTerminationEndsPublicGatewaySession(t *testing.T) { if err != nil || event.Kind != ProviderEventTerminated { t.Fatalf("native provider termination = %#v, %v", event, err) } + h.client.waitClosed(t) tryQueueNativeMedia(h.native, h.native.video, []byte("new-video")) tryQueueNativeMedia(h.native, h.native.audio, []byte("new-audio")) h.assertNoMedia(t) @@ -737,6 +745,7 @@ func TestNativeENetDisconnectQuiescesPublicGatewaySession(t *testing.T) { if err != nil || event.Kind != ProviderEventDisconnected { t.Fatalf("native provider disconnect = %#v, %v", event, err) } + h.client.waitClosed(t) tryQueueNativeMedia(h.native, h.native.video, []byte("new-video")) tryQueueNativeMedia(h.native, h.native.audio, []byte("new-audio")) h.assertNoMedia(t) @@ -746,10 +755,24 @@ func TestNativeENetDisconnectQuiescesPublicGatewaySession(t *testing.T) { } } +func TestTerminalTunnelClosesWhenIndependentClientDoesNotAcknowledge(t *testing.T) { + h := newNativeGatewayLifecycleHarness(t, "session-native-no-receipt") + h.native.handleApolloDisconnect(ErrProviderDisconnected) + + eventCtx, eventCancel := context.WithTimeout(context.Background(), time.Second) + event, err := h.client.receiveProviderEvent(eventCtx, false) + eventCancel() + if err != nil || event.Kind != ProviderEventDisconnected { + t.Fatalf("native provider disconnect = %#v, %v", event, err) + } + h.client.waitClosed(t) + h.waitReleased(t) +} + type nativeGatewayLifecycleHarness struct { native *nativeApolloSession key []byte - client *Client + client *independentGatewayClient admission *oneTimeAdmission reporter *recordingProviderStateReporter } @@ -795,7 +818,7 @@ func newNativeGatewayLifecycleHarness(t *testing.T, sessionID string) nativeGate Grant: strings.Repeat("g", 64), ClientNonce: "nonce-0000000001", DeviceSignature: strings.Repeat("s", 86), Capabilities: DefaultCapabilities(), } - client, err := Dial(context.Background(), server.Addr().String(), clientTLS, request) + client, err := dialIndependentGateway(context.Background(), server.Addr().String(), clientTLS, request) if err != nil { t.Fatal(err) } @@ -810,6 +833,127 @@ func newNativeGatewayLifecycleHarness(t *testing.T, sessionID string) nativeGate return nativeGatewayLifecycleHarness{native: native, key: key, client: client, admission: admission, reporter: reporter} } +type independentGatewayClient struct { + connection *quic.Conn + control *quic.Stream +} + +func dialIndependentGateway(ctx context.Context, address string, tlsConfig *tls.Config, request protocol.TunnelAdmissionRequest) (*independentGatewayClient, error) { + config := tlsConfig.Clone() + config.NextProtos = []string{"versevdi-gateway-v1"} + connection, err := quic.DialAddr(ctx, address, config, &quic.Config{EnableDatagrams: true, MaxIdleTimeout: 30 * time.Second}) + if err != nil { + return nil, err + } + stream, err := connection.OpenStreamSync(ctx) + if err != nil { + _ = connection.CloseWithError(applicationError, "independent client setup failed") + return nil, err + } + encoded, err := protocol.EncodeTunnelAdmissionRequest(request) + if err == nil { + err = independentWriteWire(stream, encoded) + } + if err == nil { + encoded, err = independentReadWire(stream, defaultHelloLimit) + } + if err == nil { + _, err = protocol.DecodeSessionAuthority(encoded) + } + if err != nil { + _ = connection.CloseWithError(applicationError, "independent client admission failed") + return nil, err + } + return &independentGatewayClient{connection: connection, control: stream}, nil +} + +func (c *independentGatewayClient) ReceiveProviderEvent(ctx context.Context) (ProviderEvent, error) { + return c.receiveProviderEvent(ctx, true) +} + +func (c *independentGatewayClient) receiveProviderEvent(ctx context.Context, acknowledge bool) (ProviderEvent, error) { + if deadline, ok := ctx.Deadline(); ok { + if err := c.control.SetReadDeadline(deadline); err != nil { + return ProviderEvent{}, err + } + defer c.control.SetReadDeadline(time.Time{}) + } + for { + encoded, err := independentReadWire(c.control, defaultControlLimit) + if err != nil { + return ProviderEvent{}, err + } + frame, err := protocol.DecodeChannelFrame(encoded) + if err != nil || frame.FlowID != "control.ack.v1" { + return ProviderEvent{}, ErrProviderMalformed + } + payload, err := base64.StdEncoding.DecodeString(frame.Payload) + if err != nil { + return ProviderEvent{}, err + } + event, err := DecodeProviderEvent(payload) + if acknowledge && err == nil && (event.Kind == ProviderEventTerminated || event.Kind == ProviderEventDisconnected) { + receipt, encodeErr := EncodeClientFeedback(Feedback{Kind: FeedbackTerminalReceipt}) + if encodeErr != nil { + return ProviderEvent{}, encodeErr + } + ack, encodeErr := protocol.EncodeChannelFrame(testChannelFrame("control.ack.v1", 0, receipt)) + if encodeErr != nil { + return ProviderEvent{}, encodeErr + } + if encodeErr = independentWriteWire(c.control, ack); encodeErr != nil { + return ProviderEvent{}, encodeErr + } + } + return event, err + } +} + +func (c *independentGatewayClient) ReceiveFrame(ctx context.Context) (Frame, error) { + data, err := c.connection.ReceiveDatagram(ctx) + if err != nil { + return Frame{}, err + } + return DecodeFrame(data) +} + +func (c *independentGatewayClient) waitClosed(t *testing.T) { + t.Helper() + select { + case <-c.connection.Context().Done(): + case <-time.After(terminalAckTimeout + time.Second): + t.Fatal("gateway did not close the terminal QUIC connection") + } +} + +func (c *independentGatewayClient) Close() error { + return c.connection.CloseWithError(applicationError, "independent client closed") +} + +func independentWriteWire(writer io.Writer, payload []byte) error { + var header [4]byte + binary.BigEndian.PutUint32(header[:], uint32(len(payload))) + if _, err := writer.Write(header[:]); err != nil { + return err + } + _, err := writer.Write(payload) + return err +} + +func independentReadWire(reader io.Reader, limit int) ([]byte, error) { + var header [4]byte + if _, err := io.ReadFull(reader, header[:]); err != nil { + return nil, err + } + length := binary.BigEndian.Uint32(header[:]) + if length > uint32(limit) { + return nil, ErrFrameSize + } + payload := make([]byte, length) + _, err := io.ReadFull(reader, payload) + return payload, err +} + func (h nativeGatewayLifecycleHarness) waitReleased(t *testing.T) { t.Helper() select { @@ -837,6 +981,11 @@ func TestProviderDisconnectEndsPublicGatewaySessionReconnectable(t *testing.T) { h.drainInitialMedia(t) h.session.Disconnect() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + if event, err := h.client.ReceiveProviderEvent(ctx); err != nil || event.Kind != ProviderEventDisconnected { + t.Fatalf("provider disconnect event = %#v, %v", event, err) + } + cancel() h.waitReleased(t) if states := h.reporter.States(); len(states) == 0 || states[len(states)-1].State != ProviderStateDisconnected || states[len(states)-1].CleanupPending { t.Fatalf("provider states = %#v", states) @@ -852,6 +1001,11 @@ func TestProviderTerminalCleanupFailureReportsCleanupPending(t *testing.T) { h.session.mu.Unlock() h.session.EmitEvent(ProviderEvent{Kind: ProviderEventTerminated, Payload: []byte{1, 2, 3, 4}}) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + if event, err := h.client.ReceiveProviderEvent(ctx); err != nil || event.Kind != ProviderEventTerminated { + t.Fatalf("provider termination event = %#v, %v", event, err) + } + cancel() h.waitReleased(t) if states := h.reporter.States(); len(states) == 0 || states[len(states)-1].State != ProviderStateCleanup || !states[len(states)-1].CleanupPending { t.Fatalf("provider states = %#v", states) diff --git a/gateway/transport.go b/gateway/transport.go index ddbb067..1ca14e8 100644 --- a/gateway/transport.go +++ b/gateway/transport.go @@ -23,6 +23,7 @@ const ( defaultHelloLimit = 16 * 1024 defaultControlLimit = 128 * 1024 clientControlBacklog = 64 + terminalAckTimeout = 2 * time.Second applicationError = quic.ApplicationErrorCode(0x100) controlFlowID = "control.ack.v1" inputFlowID = "input.sequenced.v1" @@ -394,18 +395,21 @@ type gatewaySession struct { inputMu sync.Mutex controlWriteMu sync.Mutex outputMu sync.Mutex + terminalMu sync.Mutex pressed map[string]struct{} sequence atomic.Uint32 mediaDrops uint64 mediaQuiesced bool terminalSent atomic.Bool + terminalAwait bool + terminalAck chan struct{} endReason error result chan error } func newGatewaySession(server *Server, connection *quic.Conn, control *quic.Stream, request protocol.TunnelAdmissionRequest, authority protocol.SessionAuthority, provider ProviderSession, clipboard *clipboardGate) *gatewaySession { ctx, cancel := context.WithCancel(context.Background()) - return &gatewaySession{server: server, connection: connection, control: control, request: request, authority: authority, provider: provider, clipboard: clipboard, ctx: ctx, cancel: cancel, pressed: make(map[string]struct{}), result: make(chan error, 3)} + return &gatewaySession{server: server, connection: connection, control: control, request: request, authority: authority, provider: provider, clipboard: clipboard, ctx: ctx, cancel: cancel, pressed: make(map[string]struct{}), terminalAck: make(chan struct{}, 1), result: make(chan error, 3)} } func (s *gatewaySession) run() { @@ -436,10 +440,6 @@ func (s *gatewaySession) run() { s.cancel() if s.terminalSent.Load() { s.cleanup() - select { - case <-s.connection.Context().Done(): - case <-timer.C: - } } } @@ -457,12 +457,15 @@ func (s *gatewaySession) providerEventLoop() { if terminal { s.outputMu.Lock() s.mediaQuiesced = true + s.terminalMu.Lock() } payload, err := EncodeProviderEvent(event) if err == nil { err = s.sendControl(s.sequence.Add(1), payload) } if terminal { + s.terminalAwait = err == nil + s.terminalMu.Unlock() s.outputMu.Unlock() } if err != nil { @@ -471,6 +474,23 @@ func (s *gatewaySession) providerEventLoop() { } if terminal { s.terminalSent.Store(true) + timer := time.NewTimer(terminalAckTimeout) + select { + case <-s.terminalAck: + case <-timer.C: + s.terminalMu.Lock() + s.terminalAwait = false + s.terminalMu.Unlock() + s.result <- context.DeadlineExceeded + return + case <-s.ctx.Done(): + timer.Stop() + s.terminalMu.Lock() + s.terminalAwait = false + s.terminalMu.Unlock() + return + } + timer.Stop() } switch event.Kind { case ProviderEventTerminated: @@ -747,6 +767,20 @@ func (s *gatewaySession) handleControl(payload []byte, sequence uint32) error { if err != nil { return err } + if feedback.Kind == FeedbackTerminalReceipt { + s.terminalMu.Lock() + defer s.terminalMu.Unlock() + if !s.terminalAwait { + return ErrProviderMalformed + } + s.terminalAwait = false + select { + case s.terminalAck <- struct{}{}: + return nil + default: + return ErrProviderMalformed + } + } feedback.Sequence = sequence return s.provider.Feedback(s.ctx, feedback) } @@ -857,9 +891,7 @@ func (s *gatewaySession) cleanup() { } else if err := s.server.config.Admission.Release(cleanupCtx, s.authority); err != nil { s.server.metrics.ProviderErrors.Add(1) } - if !s.terminalSent.Load() { - _ = s.connection.CloseWithError(applicationError, "session closed") - } + _ = s.connection.CloseWithError(applicationError, "session closed") return } if errors.Is(s.endReason, ErrProviderDisconnected) { @@ -872,9 +904,7 @@ func (s *gatewaySession) cleanup() { if err := s.server.reportProviderState(cleanupCtx, state); err != nil { s.server.metrics.ProviderErrors.Add(1) } - if !s.terminalSent.Load() { - _ = s.connection.CloseWithError(applicationError, "session closed") - } + _ = s.connection.CloseWithError(applicationError, "session closed") }) } @@ -1065,7 +1095,7 @@ func (c *Client) ReceiveProviderEvent(ctx context.Context) (ProviderEvent, error } event, err := DecodeProviderEvent(payload) if err == nil && (event.Kind == ProviderEventTerminated || event.Kind == ProviderEventDisconnected) { - _ = c.Close() + err = c.SendFeedback(Feedback{Kind: FeedbackTerminalReceipt}) } return event, err } diff --git a/go.mod b/go.mod index 7465018..0910bbd 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module git.sechmachine.io.vn/sechmachine/VerseVDI-Data-Plane go 1.26.5 require ( - git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.7 + git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.8 github.com/quic-go/quic-go v0.61.0 ) diff --git a/go.sum b/go.sum index ab40b5e..befa647 100644 --- a/go.sum +++ b/go.sum @@ -8,6 +8,8 @@ git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.6 git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.6/go.mod h1:7PhFIDhjtr20btWoEb2GqB+7dBpzJt43olrnHVutWoc= git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.7 h1:vQWELUD8bTjEI9rsJYinH2PegOlGwOkyndcD+jgLMBQ= git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.7/go.mod h1:7PhFIDhjtr20btWoEb2GqB+7dBpzJt43olrnHVutWoc= +git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.8 h1:DqD2I3bjiVt+mr741o7hw4wDUp2vYZx32CNDSkqADwY= +git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.8/go.mod h1:7PhFIDhjtr20btWoEb2GqB+7dBpzJt43olrnHVutWoc= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= diff --git a/openspec/changes/gateway-terminal-receipt/.openspec.yaml b/openspec/changes/gateway-terminal-receipt/.openspec.yaml new file mode 100644 index 0000000..ab39675 --- /dev/null +++ b/openspec/changes/gateway-terminal-receipt/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-07-30 diff --git a/openspec/changes/gateway-terminal-receipt/design.md b/openspec/changes/gateway-terminal-receipt/design.md new file mode 100644 index 0000000..a492d17 --- /dev/null +++ b/openspec/changes/gateway-terminal-receipt/design.md @@ -0,0 +1,29 @@ +## Context + +The native Apollo session and production QUIC gateway already quiesce media before terminal delivery. A public independent-client test proved that immediate `CloseWithError` can overtake the queued reliable stream frame, while waiting for client connection closure leaves tunnel ownership unbounded. + +## Goals / Non-Goals + +**Goals:** + +- Deliver one terminal event before gateway-owned closure. +- Bound closure when a client remains open or omits the receipt. +- Preserve cleanup, input release, reservation, and durable state behavior. + +**Non-Goals:** + +- A generic acknowledgement or lifecycle framework. +- Any Apollo protocol, media, Server authority, or dependency change. + +## Decisions + +- Reuse Protocol `control.ack.v1` type `0x03` as an empty terminal receipt. +- Hold the receipt-state lock across the terminal write, arm one receipt slot only after a successful write, and consume it in the gateway rather than provider feedback. +- Wait at most two seconds for receipt, then close and clean up regardless. +- Serialize the bounded native event queue and evict one older feedback item only when necessary to retain a terminal event. + +## Risks / Trade-offs + +- [Client omits receipt] → Close at the two-second bound and retain durable cleanup behavior. +- [Feedback queue is saturated] → Sacrifice one older nonterminal feedback event rather than lose terminal ownership. +- [Receipt is malformed, duplicate, or early] → Fail the session closed without provider mutation. diff --git a/openspec/changes/gateway-terminal-receipt/proposal.md b/openspec/changes/gateway-terminal-receipt/proposal.md new file mode 100644 index 0000000..cfbd93b --- /dev/null +++ b/openspec/changes/gateway-terminal-receipt/proposal.md @@ -0,0 +1,24 @@ +## Why + +P3C-018, P3C-019, P3C-021, and P3C-027 require terminal feedback, bounded cleanup, durable state, and explicit input release. A public independent-client regression proved that immediate QUIC closure loses the terminal event, while the old behavior left the tunnel open until the client closed it. + +## What Changes + +- Consume the Protocol-owned terminal receipt on `control.ack.v1` inside the gateway rather than forwarding it to Apollo. +- Quiesce media before terminal delivery and close the gateway-owned tunnel after receipt or a bounded receipt deadline. +- Guarantee a terminal event survives saturation of the bounded native feedback queue. +- Preserve provider cleanup, reservation release, reconnectable disconnect reporting, and `cleanup_pending`. + +## Capabilities + +### New Capabilities + +None. + +### Modified Capabilities + +- `provider-session-lifecycle`: Make terminal delivery and gateway-owned bounded closure executable rather than dependent on client connection closure. + +## Impact + +The pure-Go GPLv3 gateway control and native Apollo session paths change. The Protocol repository remains the wire-contract owner; the Server remains the durable authority. No cgo, sidecar, direct provider route, decode/transcode path, dependency, or proprietary source is introduced. Failure to receive a valid receipt before the bound is a hard session close, not a fallback. diff --git a/openspec/changes/gateway-terminal-receipt/specs/provider-session-lifecycle/spec.md b/openspec/changes/gateway-terminal-receipt/specs/provider-session-lifecycle/spec.md new file mode 100644 index 0000000..3a848f5 --- /dev/null +++ b/openspec/changes/gateway-terminal-receipt/specs/provider-session-lifecycle/spec.md @@ -0,0 +1,23 @@ +## MODIFIED Requirements + +### Requirement: Provider terminal events end forwarding +Encrypted provider termination and unexpected provider disconnect SHALL quiesce provider ingestion and queued/new media forwarding before the existing reliable typed terminal event is delivered. The client SHALL return the Protocol-owned terminal receipt after decoding that event. The gateway SHALL close the Verse tunnel after that receipt or a bounded receipt deadline even when the client keeps the connection open, release the session reservation, and report the appropriate durable provider/session state. The receipt MUST be consumed by the gateway and MUST NOT be forwarded to the provider. A fixed drain delay MUST NOT stand in for reliable control delivery. + +#### Scenario: Host termination closes the tunnel +- **WHEN** the native provider emits an authenticated termination event and an independent client returns its terminal receipt +- **THEN** queued and newly injected media cannot cross the Verse transport after observation, and the gateway closes the tunnel, releases the reservation, and completes the durable lifecycle transition + +#### Scenario: Unexpected provider disconnect is reconnectable +- **WHEN** required provider transport disconnects without acknowledged provider termination +- **THEN** forwarding stops, the final typed disconnect reaches the client, and the Server receives the existing reconnectable lifecycle state rather than a termination claim + +#### Scenario: Client omits terminal receipt +- **WHEN** the terminal event is written but the client remains open without returning a valid receipt +- **THEN** the gateway closes the tunnel at the bounded receipt deadline and continues cleanup without restoring media forwarding + +### Requirement: Cleanup failure remains durable +Gateway cleanup MUST preserve `cleanup_pending` when provider input release, transport cleanup, authorized cancellation, or durable reporting fails. + +#### Scenario: Terminal cleanup fails +- **WHEN** a provider terminal event is handled but required cleanup cannot complete +- **THEN** the session is not reported reusable and durable state remains cleanup pending diff --git a/openspec/changes/gateway-terminal-receipt/tasks.md b/openspec/changes/gateway-terminal-receipt/tasks.md new file mode 100644 index 0000000..58b7d74 --- /dev/null +++ b/openspec/changes/gateway-terminal-receipt/tasks.md @@ -0,0 +1,16 @@ +## 1. Regressions + +- [x] 1.1 Reproduce terminal event loss with an independent QUIC client and immediate gateway closure +- [x] 1.2 Reproduce terminal loss behind a saturated native feedback queue + +## 2. Lifecycle repair + +- [x] 2.1 Consume the scoped terminal receipt without provider forwarding +- [x] 2.2 Close acknowledged and non-acknowledged terminal tunnels within bounds +- [x] 2.3 Preserve media quiescence, reservation release, durable state, and cleanup-pending + +## 3. Verification + +- [x] 3.1 Pin the final immutable Protocol version and pass focused lifecycle/race/resource checks +- [ ] 3.2 Pass complete Data Plane verification and the frozen normative Section 7 qualification +- [ ] 3.3 Record that deterministic fixtures do not prove live Apollo, macOS-client, or physical-firewall interoperability diff --git a/openspec/specs/provider-session-lifecycle/spec.md b/openspec/specs/provider-session-lifecycle/spec.md index a6d96a9..4bb76ef 100644 --- a/openspec/specs/provider-session-lifecycle/spec.md +++ b/openspec/specs/provider-session-lifecycle/spec.md @@ -1,22 +1,26 @@ # provider-session-lifecycle Specification ## Purpose -TBD - created by archiving change phase3c-gateway-audit-remediation. Update Purpose after archive. +Define terminal media quiescence, bounded Verse tunnel ownership, provider cleanup, reservation release, and durable lifecycle outcomes. ## Requirements ### Requirement: Provider terminal events end forwarding -Encrypted provider termination and unexpected provider disconnect SHALL quiesce provider ingestion and queued/new media forwarding before the existing reliable typed terminal event is delivered, close the Verse tunnel within a bounded interval, release the session reservation, and report the appropriate durable provider/session state. A fixed drain delay MUST NOT stand in for reliable control delivery. +Encrypted provider termination and unexpected provider disconnect SHALL quiesce provider ingestion and queued/new media forwarding before the existing reliable typed terminal event is delivered. The client SHALL return the Protocol-owned terminal receipt after decoding that event. The gateway SHALL close the Verse tunnel after that receipt or a bounded receipt deadline even when the client keeps the connection open, release the session reservation, and report the appropriate durable provider/session state. The receipt MUST be consumed by the gateway and MUST NOT be forwarded to the provider. A fixed drain delay MUST NOT stand in for reliable control delivery. #### Scenario: Host termination closes the tunnel -- **WHEN** the native provider emits an authenticated termination event -- **THEN** queued and newly injected media cannot cross the Verse transport after observation, and the client tunnel, reservation, and durable lifecycle transition complete +- **WHEN** the native provider emits an authenticated termination event and an independent client returns its terminal receipt +- **THEN** queued and newly injected media cannot cross the Verse transport after observation, and the gateway closes the tunnel, releases the reservation, and completes the durable lifecycle transition #### Scenario: Unexpected provider disconnect is reconnectable -- **WHEN** required provider transport disconnects without acknowledged termination -- **THEN** forwarding stops and the Server receives the existing reconnectable lifecycle state rather than a termination claim +- **WHEN** required provider transport disconnects without acknowledged provider termination +- **THEN** forwarding stops, the final typed disconnect reaches the client, and the Server receives the existing reconnectable lifecycle state rather than a termination claim + +#### Scenario: Client omits terminal receipt +- **WHEN** the terminal event is written but the client remains open without returning a valid receipt +- **THEN** the gateway closes the tunnel at the bounded receipt deadline and continues cleanup without restoring media forwarding ### Requirement: Cleanup failure remains durable Gateway cleanup MUST preserve `cleanup_pending` when provider input release, transport cleanup, authorized cancellation, or durable reporting fails. #### Scenario: Terminal cleanup fails - **WHEN** a provider terminal event is handled but required cleanup cannot complete -- **THEN** the session is not reported released or reusable and durable state remains cleanup pending +- **THEN** the session is not reported reusable and durable state remains cleanup pending