diff --git a/README.md b/README.md index 262b603..345f3f5 100644 --- a/README.md +++ b/README.md @@ -8,12 +8,11 @@ repository. ## Current status -The repository is prepared for **Phase 3C-G gateway implementation**. It does -not yet contain a gateway or Apollo adapter. Implementation uses deterministic -pinned-source fixtures and a bounded fake provider; the owner will perform -live Apollo/macOS acceptance after both candidates exist. Deferring that live -row does not authorize direct client-to-Apollo routing, cgo, a native sidecar, -or decode/transcode behavior. +The repository contains the Phase 3C-G pure-Go QUIC/mTLS gateway and Apollo +profile. Qualification uses deterministic pinned-source fixtures and a +bounded fake provider; the owner will perform live Apollo/macOS acceptance +after both candidates exist. Deferring that live row does not authorize direct +client-to-Apollo routing, cgo, a native sidecar, or decode/transcode behavior. The existing Xcode project is retained for the later native-client phase. It is not Phase 3C gateway evidence and must not be used to move provider or diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md index 9387877..301a67d 100644 --- a/THIRD_PARTY_NOTICES.md +++ b/THIRD_PARTY_NOTICES.md @@ -1,9 +1,13 @@ # Third-party notices -No third-party gateway, provider, protocol, or streaming implementation is -incorporated at this bootstrap revision. The existing Apple Xcode project was -generated by the platform tool and remains reserved for a later native-client -phase. +The gateway uses the following exact third-party dependency: + +- `github.com/quic-go/quic-go` v0.61.0, upstream release commit + `579ee19`, MIT license. It supplies the pure-Go QUIC/TLS and RFC 9221 + DATAGRAM transport only; no provider or client implementation is linked. + +The existing Apple Xcode project was generated by the platform tool and remains +reserved for a later native-client phase. Apollo, Moonlight, and related repositories are external research references only. Before any source is copied, adapted, linked, embedded, or used to create diff --git a/cmd/verse-gateway/main.go b/cmd/verse-gateway/main.go new file mode 100644 index 0000000..7a93e7f --- /dev/null +++ b/cmd/verse-gateway/main.go @@ -0,0 +1,125 @@ +package main + +import ( + "context" + "crypto/tls" + "crypto/x509" + "errors" + "flag" + "fmt" + "log" + "os" + "os/signal" + "strings" + "syscall" + "time" + + "net/http" + + "git.sechmachine.io.vn/sechmachine/VerseVDI-Data-Plane/gateway" + protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol" +) + +func main() { + if err := run(); err != nil { + log.Fatal(err) + } +} + +func run() error { + var listen, controlPlane, certFile, keyFile, clientCAFile string + var gatewayID, instanceIdentity, certificateIdentity, publicIdentity string + var providerManagement, providerRTSPAddress, providerRTSPURL, providerIdentity string + flag.StringVar(&listen, "listen", "0.0.0.0:443", "gateway QUIC listen address") + flag.StringVar(&controlPlane, "control-plane", "", "Connection Server HTTPS base URL") + flag.StringVar(&certFile, "cert", "", "gateway certificate PEM") + flag.StringVar(&keyFile, "key", "", "gateway private key PEM") + flag.StringVar(&clientCAFile, "client-ca", "", "Connection Server/client CA PEM") + flag.StringVar(&gatewayID, "gateway-id", "", "stable gateway identifier") + flag.StringVar(&instanceIdentity, "instance-identity", "", "gateway instance identity") + flag.StringVar(&certificateIdentity, "certificate-identity", "", "gateway certificate identity") + flag.StringVar(&publicIdentity, "public-identity", "gateway", "gateway public identity") + flag.StringVar(&providerManagement, "provider-management", "", "internal Apollo management URL") + flag.StringVar(&providerRTSPAddress, "provider-rtsp-address", "", "internal Apollo RTSP address") + flag.StringVar(&providerRTSPURL, "provider-rtsp-url", "", "internal Apollo RTSP URL") + flag.StringVar(&providerIdentity, "provider-identity", "", "enrolled Apollo identity unique-id#fingerprint") + flag.Parse() + for name, value := range map[string]string{"control-plane": controlPlane, "cert": certFile, "key": keyFile, "client-ca": clientCAFile, "gateway-id": gatewayID, "instance-identity": instanceIdentity, "certificate-identity": certificateIdentity, "provider-management": providerManagement, "provider-rtsp-address": providerRTSPAddress, "provider-rtsp-url": providerRTSPURL, "provider-identity": providerIdentity} { + if value == "" { + return fmt.Errorf("-%s is required", name) + } + } + expectedIdentity, err := parseProviderIdentity(providerIdentity) + if err != nil { + return err + } + serverTLS, clientTLS, err := loadTLS(certFile, keyFile, clientCAFile) + if err != nil { + return err + } + transport := &http.Transport{TLSClientConfig: clientTLS} + controlPlaneClient := gateway.NewControlPlaneClient(controlPlane, &http.Client{Transport: transport, Timeout: 5 * time.Second}) + providerBackend := gateway.NewNativeApolloBackend(providerManagement, providerRTSPAddress, providerRTSPURL, &http.Client{Timeout: 5 * time.Second}) + provider := gateway.NewApolloAdapter(providerBackend, expectedIdentity) + capabilities := gateway.DefaultCapabilities() + server, err := gateway.NewServer(gateway.ServerConfig{ListenAddress: listen, TLSConfig: serverTLS, GatewayID: gatewayID, Capabilities: capabilities, ProviderCapabilities: capabilities, Admission: controlPlaneClient, Provider: provider, PacerKbps: 100000}) + if err != nil { + return err + } + registration := protocol.GatewayRegistration{Version: "1", GatewayID: gatewayID, InstanceIdentity: instanceIdentity, CertificateIdentity: certificateIdentity, PublicIdentity: publicIdentity, Address: server.Addr().String(), ProtocolMinVersion: 1, ProtocolMaxVersion: 1, ConnectionCapacity: 8, BandwidthCapacityKbps: 100000, Features: []string{"quic-tls13", "datagram.media", "apollo"}, Capabilities: capabilities} + if _, err := controlPlaneClient.Register(context.Background(), registration); err != nil { + _ = server.Close() + return err + } + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + go heartbeatLoop(ctx, controlPlaneClient, server, registration) + return server.Serve(ctx) +} + +func heartbeatLoop(ctx context.Context, client *gateway.ControlPlaneClient, server *gateway.Server, registration protocol.GatewayRegistration) { + ticker := time.NewTicker(2 * time.Second) + defer ticker.Stop() + var sequence int64 + for { + select { + case <-ctx.Done(): + server.BeginDrain() + deadline := time.Now().Add(5 * time.Second).UTC().Format(time.RFC3339Nano) + _ = client.Drain(context.Background(), protocol.GatewayDrain{Version: "1", GatewayID: registration.GatewayID, Sequence: sequence + 1, Reason: "shutdown", Deadline: deadline}) + return + case <-ticker.C: + sequence++ + state := "ready" + if server.Draining() { + state = "draining" + } + metrics := server.Metrics() + _ = client.Heartbeat(ctx, protocol.GatewayHeartbeat{Version: "1", GatewayID: registration.GatewayID, Sequence: sequence, ObservedAt: time.Now().UTC().Format(time.RFC3339Nano), ActiveConnections: metrics.ActiveSessions, EgressKbps: registration.BandwidthCapacityKbps, State: state}) + } + } +} + +func loadTLS(certFile, keyFile, clientCAFile string) (*tls.Config, *tls.Config, error) { + certificate, err := tls.LoadX509KeyPair(certFile, keyFile) + if err != nil { + return nil, nil, err + } + caBytes, err := os.ReadFile(clientCAFile) + if err != nil { + return nil, nil, err + } + clientCAs := x509.NewCertPool() + if !clientCAs.AppendCertsFromPEM(caBytes) { + return nil, nil, errors.New("client CA PEM contains no certificate") + } + return &tls.Config{MinVersion: tls.VersionTLS13, Certificates: []tls.Certificate{certificate}, ClientAuth: tls.RequireAndVerifyClientCert, ClientCAs: clientCAs}, &tls.Config{MinVersion: tls.VersionTLS13, Certificates: []tls.Certificate{certificate}, RootCAs: clientCAs}, nil +} + +func parseProviderIdentity(value string) (gateway.ProviderIdentity, error) { + uniqueID, fingerprint, ok := strings.Cut(value, "#") + if !ok || uniqueID == "" || fingerprint == "" { + return gateway.ProviderIdentity{}, errors.New("provider identity must be unique-id#fingerprint") + } + return gateway.ProviderIdentity{UniqueID: uniqueID, Fingerprint: fingerprint}, nil +} diff --git a/gateway/apollo_native.go b/gateway/apollo_native.go new file mode 100644 index 0000000..ec3376b --- /dev/null +++ b/gateway/apollo_native.go @@ -0,0 +1,264 @@ +package gateway + +import ( + "bufio" + "context" + "crypto/tls" + "fmt" + "io" + "net" + "net/http" + "strings" + "sync" + "time" + + protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol" +) + +// NativeApolloBackend keeps provider sockets inside the gateway process. The +// RTSP endpoint is configuration owned by the gateway and is never serialized +// into a client manifest or authority. +type NativeApolloBackend struct { + ManagementURL string + RTSPAddress string + RTSPURL string + HTTPClient *http.Client + Dialer *net.Dialer + TLSConfig *tls.Config + + mu sync.Mutex + pending map[string]net.Conn +} + +func NewNativeApolloBackend(managementURL, rtspAddress, rtspURL string, client *http.Client) *NativeApolloBackend { + if client == nil { + client = &http.Client{Timeout: 5 * time.Second} + } + return &NativeApolloBackend{ManagementURL: managementURL, RTSPAddress: rtspAddress, RTSPURL: rtspURL, HTTPClient: client, Dialer: &net.Dialer{Timeout: 5 * time.Second}, pending: make(map[string]net.Conn)} +} + +func (b *NativeApolloBackend) Management(ctx context.Context) ([]byte, error) { + if b.ManagementURL == "" { + return nil, ErrProviderMalformed + } + request, err := http.NewRequestWithContext(ctx, http.MethodGet, b.ManagementURL, nil) + if err != nil { + return nil, ErrProviderMalformed + } + response, err := b.HTTPClient.Do(request) + if err != nil { + return nil, err + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + return nil, fmt.Errorf("provider management status %d", response.StatusCode) + } + return readBounded(response.Body, 64*1024) +} + +func (b *NativeApolloBackend) Setup(ctx context.Context, request LaunchRequest) ([]byte, error) { + if b.RTSPAddress == "" || b.RTSPURL == "" || request.SessionID == "" { + return nil, ErrProviderMalformed + } + conn, err := b.Dialer.DialContext(ctx, "tcp", b.RTSPAddress) + if err != nil { + return nil, err + } + if deadline, ok := ctx.Deadline(); ok { + _ = conn.SetDeadline(deadline) + } + requestText := "SETUP " + b.RTSPURL + " RTSP/1.0\r\nCSeq: 1\r\nTransport: RTP/AVP/TCP;interleaved=0-1\r\nSession: " + request.SessionID + "\r\n\r\n" + if _, err := io.WriteString(conn, requestText); err != nil { + _ = conn.Close() + return nil, err + } + response, err := readRTSPHeaders(conn, 16*1024) + if err != nil { + _ = conn.Close() + return nil, err + } + b.mu.Lock() + b.pending[request.SessionID] = conn + b.mu.Unlock() + return response, nil +} + +func (b *NativeApolloBackend) Open(_ context.Context, request LaunchRequest, _ RTSPResponse) (ProviderSession, error) { + b.mu.Lock() + conn := b.pending[request.SessionID] + delete(b.pending, request.SessionID) + b.mu.Unlock() + if conn == nil { + return nil, ErrProviderDisconnected + } + session := newNativeApolloSession(conn, request.SessionID) + go session.readMedia() + return session, nil +} + +func readBounded(reader io.Reader, max int) ([]byte, error) { + data, err := io.ReadAll(io.LimitReader(reader, int64(max)+1)) + if err != nil { + return nil, err + } + if len(data) > max { + return nil, ErrProviderMalformed + } + return data, nil +} + +func readRTSPHeaders(conn net.Conn, max int) ([]byte, error) { + reader := bufio.NewReaderSize(conn, 4096) + var response []byte + for len(response) < max { + line, err := reader.ReadBytes('\n') + if err != nil { + return nil, err + } + response = append(response, line...) + if strings.HasSuffix(string(response), "\r\n\r\n") { + return response, nil + } + } + return nil, ErrProviderMalformed +} + +type nativeApolloSession struct { + conn net.Conn + sessionID string + video chan []byte + audio chan []byte + mu sync.Mutex + state protocol.ProviderState + closeOnce sync.Once + done chan struct{} + readDone chan struct{} +} + +func newNativeApolloSession(conn net.Conn, sessionID string) *nativeApolloSession { + return &nativeApolloSession{conn: conn, sessionID: sessionID, video: make(chan []byte, 16), audio: make(chan []byte, 16), state: protocol.ProviderState{Version: "1", SessionID: sessionID, State: ProviderStateStarting, Channels: []string{"video", "audio", "input", "feedback"}}, done: make(chan struct{}), readDone: make(chan struct{})} +} + +func (s *nativeApolloSession) Ready(context.Context) error { + s.mu.Lock() + s.state.State = ProviderStateReady + s.mu.Unlock() + return nil +} + +func (s *nativeApolloSession) Video() <-chan []byte { return s.video } +func (s *nativeApolloSession) Audio() <-chan []byte { return s.audio } + +func (s *nativeApolloSession) Input(ctx context.Context, event InputEvent) error { + payload, err := EncodeInputEvent(event) + if err != nil { + return err + } + return s.writeControl(ctx, ControlPacket{Kind: 1, Sequence: event.Sequence, Payload: payload}) +} + +func (s *nativeApolloSession) Feedback(ctx context.Context, feedback Feedback) error { + return s.writeControl(ctx, ControlPacket{Kind: 3, Sequence: feedback.Sequence, Payload: feedback.Payload}) +} + +func (s *nativeApolloSession) Reconnect(ctx context.Context) error { + return s.writeControl(ctx, ControlPacket{Kind: 4, Payload: []byte("RECN")}) +} + +func (s *nativeApolloSession) ReleaseAll(ctx context.Context) error { + return s.writeControl(ctx, ControlPacket{Kind: 2, Payload: []byte("RELEASE_ALL")}) +} + +func (s *nativeApolloSession) Terminate(ctx context.Context) error { + _ = s.writeControl(ctx, ControlPacket{Kind: 5, Payload: []byte("TEAR")}) + var timedOut bool + s.closeOnce.Do(func() { + close(s.done) + _ = s.conn.Close() + select { + case <-s.readDone: + case <-ctx.Done(): + timedOut = true + } + if !timedOut { + close(s.video) + close(s.audio) + } + }) + s.mu.Lock() + if timedOut { + s.state.State = ProviderStateCleanup + s.state.CleanupPending = true + } else { + s.state.State = ProviderStateTerminated + } + s.mu.Unlock() + if timedOut { + return ctx.Err() + } + return nil +} + +func (s *nativeApolloSession) State() protocol.ProviderState { + s.mu.Lock() + defer s.mu.Unlock() + return s.state +} + +func (s *nativeApolloSession) writeControl(ctx context.Context, packet ControlPacket) error { + encoded, err := EncodeControlPacket(packet) + if err != nil { + return err + } + if deadline, ok := ctx.Deadline(); ok { + _ = s.conn.SetWriteDeadline(deadline) + } + if _, err := s.conn.Write(encoded); err != nil { + return err + } + return nil +} + +func (s *nativeApolloSession) readMedia() { + defer close(s.readDone) + header := make([]byte, 4) + for { + if _, err := io.ReadFull(s.conn, header); err != nil { + return + } + if header[0] != '$' || (header[1] != 0 && header[1] != 1) { + return + } + length := int(header[2])<<8 | int(header[3]) + if length > 65536 { + return + } + payload := make([]byte, length) + if _, err := io.ReadFull(s.conn, payload); err != nil { + return + } + if header[1] == 0 { + pushLatest(s.video, payload) + } else { + pushLatest(s.audio, payload) + } + } +} + +func pushLatest(channel chan []byte, payload []byte) { + select { + case channel <- payload: + default: + select { + case <-channel: + default: + } + select { + case channel <- payload: + default: + } + } +} + +var _ ApolloBackend = (*NativeApolloBackend)(nil) +var _ ProviderSession = (*nativeApolloSession)(nil) diff --git a/gateway/capability.go b/gateway/capability.go new file mode 100644 index 0000000..9122b9f --- /dev/null +++ b/gateway/capability.go @@ -0,0 +1,43 @@ +package gateway + +import ( + "errors" + + protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol" +) + +var ErrNoCapabilityOverlap = errors.New("no capability overlap") + +func DefaultCapabilities() protocol.CapabilityProfile { + return protocol.CapabilityProfile{ + Transport: "quic-tls13", + Framing: "datagram-v1", + Media: "encoded", + Audio: "encoded", + SourceRateControl: "server", + ClientDecode: "h264-opus", + } +} + +func IntersectCapabilities(profiles ...protocol.CapabilityProfile) (protocol.CapabilityProfile, error) { + if len(profiles) == 0 { + return protocol.CapabilityProfile{}, ErrNoCapabilityOverlap + } + for _, profile := range profiles { + if err := profile.Validate(); err != nil { + return protocol.CapabilityProfile{}, ErrNoCapabilityOverlap + } + } + selected := profiles[0] + for _, profile := range profiles[1:] { + if selected.Transport != profile.Transport || + selected.Framing != profile.Framing || + selected.Media != profile.Media || + selected.Audio != profile.Audio || + selected.SourceRateControl != profile.SourceRateControl || + selected.ClientDecode != profile.ClientDecode { + return protocol.CapabilityProfile{}, ErrNoCapabilityOverlap + } + } + return selected, nil +} diff --git a/gateway/control_plane.go b/gateway/control_plane.go new file mode 100644 index 0000000..e7e01aa --- /dev/null +++ b/gateway/control_plane.go @@ -0,0 +1,110 @@ +package gateway + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + + protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol" +) + +type ControlPlaneClient struct { + BaseURL string + HTTPClient *http.Client +} + +func NewControlPlaneClient(baseURL string, client *http.Client) *ControlPlaneClient { + if client == nil { + client = &http.Client{} + } + return &ControlPlaneClient{BaseURL: strings.TrimRight(baseURL, "/"), HTTPClient: client} +} + +func (c *ControlPlaneClient) Register(ctx context.Context, registration protocol.GatewayRegistration) (protocol.GatewayRegistration, error) { + payload, err := protocol.EncodeGatewayRegistration(registration) + if err != nil { + return protocol.GatewayRegistration{}, err + } + response, err := c.post(ctx, "/api/v1/gateway/register", payload) + if err != nil { + return protocol.GatewayRegistration{}, err + } + return protocol.DecodeGatewayRegistration(response) +} + +func (c *ControlPlaneClient) Heartbeat(ctx context.Context, heartbeat protocol.GatewayHeartbeat) error { + payload, err := protocol.EncodeGatewayHeartbeat(heartbeat) + if err != nil { + return err + } + _, err = c.post(ctx, "/api/v1/gateway/heartbeat", payload) + return err +} + +func (c *ControlPlaneClient) Drain(ctx context.Context, drain protocol.GatewayDrain) error { + payload, err := protocol.EncodeGatewayDrain(drain) + if err != nil { + return err + } + _, err = c.post(ctx, "/api/v1/gateway/drain", payload) + return err +} + +func (c *ControlPlaneClient) Admit(ctx context.Context, request protocol.TunnelAdmissionRequest) (protocol.SessionAuthority, error) { + payload, err := protocol.EncodeTunnelAdmissionRequest(request) + if err != nil { + return protocol.SessionAuthority{}, err + } + response, err := c.post(ctx, "/api/v1/gateway/admit", payload) + if err != nil { + return protocol.SessionAuthority{}, err + } + return protocol.DecodeSessionAuthority(response) +} + +func (c *ControlPlaneClient) Release(ctx context.Context, authority protocol.SessionAuthority) error { + payload, err := protocol.EncodeSessionAuthority(authority) + if err != nil { + return err + } + _, err = c.post(ctx, "/api/v1/gateway/release", payload) + return err +} + +func (c *ControlPlaneClient) post(ctx context.Context, path string, payload []byte) ([]byte, error) { + 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)) + if err != nil { + return nil, err + } + request.Header.Set("Content-Type", "application/json") + response, err := c.HTTPClient.Do(request) + if err != nil { + return nil, err + } + defer response.Body.Close() + body, err := io.ReadAll(io.LimitReader(response.Body, defaultControlLimit+1)) + if err != nil { + return nil, err + } + if len(body) > defaultControlLimit { + return nil, ErrFrameSize + } + if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { + var stable protocol.StableError + if json.Unmarshal(body, &stable) == nil && stable.Code != "" { + return nil, fmt.Errorf("%s: %s", stable.Code, stable.Message) + } + return nil, fmt.Errorf("control-plane status %d", response.StatusCode) + } + return body, nil +} + +var _ Admission = (*ControlPlaneClient)(nil) diff --git a/gateway/frame.go b/gateway/frame.go new file mode 100644 index 0000000..b9d3a13 --- /dev/null +++ b/gateway/frame.go @@ -0,0 +1,171 @@ +package gateway + +import ( + "encoding/binary" + "errors" + "fmt" +) + +const ( + frameHeaderSize = 21 + maxFrameSize = 1 << 16 + maxFragmentCount = 16 +) + +const ( + ChannelControl = byte(1) + ChannelAck = byte(2) + ChannelText = byte(3) + ChannelVideo = byte(10) + ChannelAudio = byte(11) + ChannelInput = byte(12) +) + +var ( + ErrFrameTruncated = errors.New("gateway frame truncated") + ErrFrameMagic = errors.New("gateway frame magic mismatch") + ErrFrameVersion = errors.New("gateway frame version unsupported") + ErrFrameChannel = errors.New("gateway frame channel unsupported") + ErrFrameFlags = errors.New("gateway frame flags unsupported") + ErrFrameFragment = errors.New("gateway frame fragment invalid") + ErrFrameLength = errors.New("gateway frame length mismatch") + ErrFramePayloadLimit = errors.New("gateway frame payload exceeds channel limit") + ErrFrameSize = errors.New("gateway frame exceeds size limit") + ErrFrameFragmentedLimit = errors.New("gateway payload requires too many fragments") +) + +type Frame struct { + Channel byte + Flags byte + Sequence uint32 + TimestampMS uint64 + FragmentIndex byte + FragmentCount byte + Payload []byte +} + +func channelLimit(channel byte) (int, bool) { + switch channel { + case ChannelControl: + return 1024, true + case ChannelAck: + return 2048, true + case ChannelText: + return 65515, true + case ChannelVideo, ChannelAudio, ChannelInput: + return 1179, true + default: + return 0, false + } +} + +func EncodeFrame(frame Frame) ([]byte, error) { + limit, ok := channelLimit(frame.Channel) + if !ok { + return nil, ErrFrameChannel + } + if frame.Flags != 0 { + return nil, ErrFrameFlags + } + if frame.FragmentCount == 0 || frame.FragmentCount > maxFragmentCount || frame.FragmentIndex >= frame.FragmentCount { + return nil, ErrFrameFragment + } + if len(frame.Payload) > limit { + return nil, ErrFramePayloadLimit + } + if len(frame.Payload) > maxFrameSize-frameHeaderSize { + return nil, ErrFrameSize + } + encoded := make([]byte, frameHeaderSize+len(frame.Payload)) + encoded[0], encoded[1], encoded[2], encoded[3], encoded[4] = 'V', 'D', 1, frame.Channel, frame.Flags + binary.BigEndian.PutUint32(encoded[5:9], frame.Sequence) + binary.BigEndian.PutUint64(encoded[9:17], frame.TimestampMS) + encoded[17], encoded[18] = frame.FragmentIndex, frame.FragmentCount + binary.BigEndian.PutUint16(encoded[19:21], uint16(len(frame.Payload))) + copy(encoded[frameHeaderSize:], frame.Payload) + return encoded, nil +} + +func DecodeFrame(raw []byte) (Frame, error) { + if len(raw) < frameHeaderSize { + return Frame{}, ErrFrameTruncated + } + if len(raw) > maxFrameSize { + return Frame{}, ErrFrameSize + } + if raw[0] != 'V' || raw[1] != 'D' { + return Frame{}, ErrFrameMagic + } + if raw[2] != 1 { + return Frame{}, ErrFrameVersion + } + limit, ok := channelLimit(raw[3]) + if !ok { + return Frame{}, ErrFrameChannel + } + if raw[4] != 0 { + return Frame{}, ErrFrameFlags + } + if raw[18] == 0 || raw[18] > maxFragmentCount || raw[17] >= raw[18] { + return Frame{}, ErrFrameFragment + } + payloadLength := int(binary.BigEndian.Uint16(raw[19:21])) + if payloadLength > limit { + return Frame{}, ErrFramePayloadLimit + } + if len(raw) != frameHeaderSize+payloadLength { + return Frame{}, ErrFrameLength + } + return Frame{ + Channel: raw[3], + Flags: raw[4], + Sequence: binary.BigEndian.Uint32(raw[5:9]), + TimestampMS: binary.BigEndian.Uint64(raw[9:17]), + FragmentIndex: raw[17], + FragmentCount: raw[18], + Payload: append([]byte(nil), raw[frameHeaderSize:]...), + }, nil +} + +func FragmentPayload(channel byte, sequence uint32, timestampMS uint64, payload []byte) ([]Frame, error) { + limit, ok := channelLimit(channel) + if !ok { + return nil, ErrFrameChannel + } + if limit > 1179 { + limit = 1179 + } + count := (len(payload) + limit - 1) / limit + if count == 0 { + count = 1 + } + if count > maxFragmentCount { + return nil, ErrFrameFragmentedLimit + } + frames := make([]Frame, 0, count) + for index := 0; index < count; index++ { + start := index * limit + end := start + limit + if end > len(payload) { + end = len(payload) + } + frames = append(frames, Frame{ + Channel: channel, + Sequence: sequence, + TimestampMS: timestampMS, + FragmentIndex: byte(index), + FragmentCount: byte(count), + Payload: append([]byte(nil), payload[start:end]...), + }) + } + return frames, nil +} + +func ValidateFrame(raw []byte) error { + _, err := DecodeFrame(raw) + return err +} + +func FrameError(channel byte, err error) error { + return fmt.Errorf("channel %d: %w", channel, err) +} diff --git a/gateway/gateway_test.go b/gateway/gateway_test.go new file mode 100644 index 0000000..f408f1a --- /dev/null +++ b/gateway/gateway_test.go @@ -0,0 +1,354 @@ +package gateway + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/hex" + "errors" + "math/big" + "net" + "os" + "strings" + "sync/atomic" + "testing" + "time" + + protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol" +) + +func TestFrameValidationAndFragmentation(t *testing.T) { + frames, err := FragmentPayload(ChannelVideo, 7, 11, make([]byte, 1180)) + if err != nil || len(frames) != 2 || len(frames[0].Payload) != 1179 || len(frames[1].Payload) != 1 { + t.Fatalf("fragmentation = %#v, err = %v", frames, err) + } + encoded, err := EncodeFrame(frames[0]) + if err != nil { + t.Fatal(err) + } + decoded, err := DecodeFrame(encoded) + if err != nil || string(decoded.Payload) != string(frames[0].Payload) { + t.Fatalf("decoded = %#v, err = %v", decoded, err) + } + for _, raw := range [][]byte{ + {0x56, 0x44}, + append([]byte(nil), encoded[:len(encoded)-1]...), + append(append([]byte(nil), encoded...), 0), + } { + if err := ValidateFrame(raw); err == nil { + t.Fatalf("accepted malformed frame %x", raw) + } + } +} + +func FuzzDecodeFrame(f *testing.F) { + seed, _ := hex.DecodeString("5644010a0000000000000000000000000000010000") + f.Add(seed) + f.Add([]byte("not-a-frame")) + f.Fuzz(func(t *testing.T, data []byte) { + _, _ = DecodeFrame(data) + }) +} + +func FuzzDecodeControlPacket(f *testing.F) { + seed, _ := EncodeControlPacket(ControlPacket{Kind: 1, Sequence: 2, Payload: []byte("fixture")}) + f.Add(seed) + f.Add([]byte("APC1")) + f.Fuzz(func(t *testing.T, data []byte) { + _, _ = DecodeControlPacket(data) + }) +} + +func FuzzDecodeInputEvent(f *testing.F) { + seed, _ := EncodeInputEvent(InputEvent{Sequence: 1, Device: "keyboard", Code: 7, Pressed: true}) + f.Add(seed) + f.Add([]byte("INP1")) + f.Fuzz(func(t *testing.T, data []byte) { + _, _ = DecodeInputEvent(data) + }) +} + +func TestCapabilityIntersectionAndBoundedQueue(t *testing.T) { + capabilities := DefaultCapabilities() + if _, err := IntersectCapabilities(capabilities, capabilities); err != nil { + t.Fatal(err) + } + other := capabilities + other.Audio = "different" + if !errors.Is(func() error { _, err := IntersectCapabilities(capabilities, other); return err }(), ErrNoCapabilityOverlap) { + t.Fatal("capability mismatch was accepted") + } + queue := NewBoundedQueue[int](2) + _ = queue.PushLatest(1) + _ = queue.PushLatest(2) + _ = queue.PushLatest(3) + if queue.Dropped() != 1 || queue.Len() != 2 { + t.Fatalf("queue length=%d dropped=%d", queue.Len(), queue.Dropped()) + } + ctx := context.Background() + first, _ := queue.Pop(ctx) + second, _ := queue.Pop(ctx) + if first != 2 || second != 3 { + t.Fatalf("queue values=%d,%d", first, second) + } +} + +func TestSyntheticImpairmentPacingAndResourceBounds(t *testing.T) { + payload := make([]byte, 1179*16+1) + if _, err := FragmentPayload(ChannelVideo, 1, 0, payload); !errors.Is(err, ErrFrameFragmentedLimit) { + t.Fatalf("oversized media payload accepted: %v", err) + } + frames, err := FragmentPayload(ChannelVideo, 1, 0, bytesRepeat(0x5a, 1179*4)) + if err != nil { + t.Fatal(err) + } + var delivered int + var deliveredBytes int + for index, frame := range frames { + if (index+1)%3 == 0 { // deterministic synthetic loss profile: every third frame. + continue + } + delivered++ + encoded, encodeErr := EncodeFrame(frame) + if encodeErr != nil { + t.Fatal(encodeErr) + } + decoded, decodeErr := DecodeFrame(encoded) + if decodeErr != nil { + t.Fatal(decodeErr) + } + deliveredBytes += len(decoded.Payload) + } + 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) { + management, err := os.ReadFile("testdata/apollo-management.xml") + if err != nil { + t.Fatal(err) + } + info, err := ParseManagementXML(management) + if err != nil || info.Identity.UniqueID != "apollo-fixture-1" { + t.Fatalf("management = %#v, err = %v", info, err) + } + rtspText, err := os.ReadFile("testdata/rtsp-setup-response.txt") + if err != nil { + t.Fatal(err) + } + rtsp, err := ParseRTSPResponse([]byte(strings.ReplaceAll(string(rtspText), `\r\n`, "\r\n"))) + if err != nil || rtsp.StatusCode != 200 { + t.Fatalf("RTSP = %#v, err = %v", rtsp, err) + } + video, _ := hex.DecodeString(strings.TrimSpace(string(mustRead(t, "testdata/encoded-video.hex")))) + audio, _ := hex.DecodeString(strings.TrimSpace(string(mustRead(t, "testdata/encoded-audio.hex")))) + now := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) + identity := ProviderIdentity{UniqueID: "apollo-fixture-1", Fingerprint: "sha256:fixture-apollo-1"} + fake := NewFakeApollo(FakeApolloConfig{Identity: identity, Now: now, Video: [][]byte{video}, Audio: [][]byte{audio}}) + session, err := fake.Start(context.Background(), LaunchRequest{SessionID: "session-1", ProviderProfile: ProviderProfileApollo, ProviderIdentity: identity.Key(), Capabilities: DefaultCapabilities()}) + if err != nil { + t.Fatal(err) + } + if got := <-session.Video(); string(got) != string(video) { + t.Fatalf("video changed: %x", got) + } + if got := <-session.Audio(); string(got) != string(audio) { + t.Fatalf("audio changed: %x", got) + } + if err := session.Input(context.Background(), InputEvent{Sequence: 1, Device: "keyboard", Code: 7, Pressed: true}); err != nil { + t.Fatal(err) + } + if err := session.ReleaseAll(context.Background()); err != nil { + t.Fatal(err) + } + if err := session.Terminate(context.Background()); err != nil { + t.Fatal(err) + } + if state := session.State(); state.State != ProviderStateTerminated || state.CleanupPending { + t.Fatalf("state = %#v", state) + } + + identityFailure := NewFakeApollo(FakeApolloConfig{Identity: identity, Now: now, Failure: FakeFailureIdentity}) + if _, err := identityFailure.Start(context.Background(), LaunchRequest{SessionID: "session-2", ProviderProfile: ProviderProfileApollo, ProviderIdentity: identity.Key(), Capabilities: DefaultCapabilities()}); !errors.Is(err, ErrProviderIdentity) { + t.Fatalf("identity failure = %v", err) + } +} + +func TestProviderTimeoutAndBoundedInput(t *testing.T) { + now := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) + fake := NewFakeApollo(FakeApolloConfig{Now: now, Failure: FakeFailureReadinessTimeout}) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + started := time.Now() + _, err := fake.Start(ctx, LaunchRequest{SessionID: "session-timeout", ProviderProfile: ProviderProfileApollo, ProviderIdentity: fake.config.Identity.Key(), Capabilities: DefaultCapabilities()}) + if !errors.Is(err, ErrProviderTimeout) || time.Since(started) > time.Second { + t.Fatalf("readiness timeout = %v after %s", err, time.Since(started)) + } + if _, err := EncodeInputEvent(InputEvent{Device: strings.Repeat("d", 65)}); !errors.Is(err, ErrInputMalformed) { + t.Fatalf("oversized input accepted: %v", err) + } + if _, err := DecodeControlPacket([]byte("APC1")); !errors.Is(err, ErrProviderMalformed) { + t.Fatalf("truncated control accepted: %v", err) + } +} + +func TestNativeApolloEncodedRelay(t *testing.T) { + provider, peer := net.Pipe() + session := newNativeApolloSession(provider, "session-native") + go session.readMedia() + go func() { + _, _ = peer.Write([]byte{'$', 0, 0, 3, 1, 2, 3}) + _, _ = peer.Write([]byte{'$', 1, 0, 2, 4, 5}) + }() + select { + case payload := <-session.Video(): + if string(payload) != string([]byte{1, 2, 3}) { + t.Fatalf("video payload changed: %x", payload) + } + case <-time.After(time.Second): + t.Fatal("video payload not relayed") + } + select { + case payload := <-session.Audio(): + if string(payload) != string([]byte{4, 5}) { + t.Fatalf("audio payload changed: %x", payload) + } + case <-time.After(time.Second): + t.Fatal("audio payload not relayed") + } + terminateCtx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + _ = session.Terminate(terminateCtx) + _ = peer.Close() +} + +func TestAdmissionQUICMTLSRelayAndCleanup(t *testing.T) { + serverTLS, clientTLS := testTLS(t) + fake := NewFakeApollo(FakeApolloConfig{Now: time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)}) + authority := protocol.SessionAuthority{Version: "1", SessionID: "session-1", GatewayID: "gateway-1", Audience: "versevdi-gateway", ReconnectSequence: 0, ExpiresAt: time.Now().Add(5 * time.Second).UTC().Format(time.RFC3339Nano), Capabilities: DefaultCapabilities(), ProviderProfile: ProviderProfileApollo, ProviderIdentity: fake.config.Identity.Key()} + admission := &oneTimeAdmission{authority: authority, released: make(chan struct{})} + server, err := NewServer(ServerConfig{ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: "gateway-1", Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(), Admission: admission, Provider: fake}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + serveDone := make(chan error, 1) + go func() { serveDone <- server.Serve(ctx) }() + request := protocol.TunnelAdmissionRequest{Version: "1", SessionID: "session-1", GatewayID: "gateway-1", Audience: "versevdi-gateway", Grant: strings.Repeat("g", 64), ReconnectSequence: 0, ClientNonce: "nonce-0000000001", Capabilities: DefaultCapabilities()} + client, err := Dial(context.Background(), server.Addr().String(), clientTLS, request) + if err != nil { + _ = server.Close() + t.Fatal(err) + } + for i := 0; i < 2; i++ { + frame, receiveErr := client.ReceiveFrame(context.Background()) + if receiveErr != nil { + t.Fatal(receiveErr) + } + if frame.Channel != ChannelVideo && frame.Channel != ChannelAudio { + t.Fatalf("unexpected media channel %d", frame.Channel) + } + } + if err := client.SendInput(InputEvent{Sequence: 1, Device: "keyboard", Code: 7, Pressed: true}); err != nil { + t.Fatal(err) + } + _ = client.Close() + deadline := time.NewTimer(2 * time.Second) + select { + case <-admission.released: + case <-deadline.C: + t.Fatal("gateway did not release admission") + } + deadline.Stop() + if fake.LastSession().State().State != ProviderStateTerminated { + t.Fatalf("provider state = %#v", fake.LastSession().State()) + } + if _, err := Dial(context.Background(), server.Addr().String(), clientTLS, request); err == nil { + t.Fatal("replayed grant was accepted") + } + _ = server.Close() + if err := <-serveDone; err != nil { + t.Fatal(err) + } +} + +func testTLS(t *testing.T) (*tls.Config, *tls.Config) { + t.Helper() + caKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + caTemplate := &x509.Certificate{SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "Verse Test CA"}, NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(time.Hour), IsCA: true, BasicConstraintsValid: true, KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature} + caDER, err := x509.CreateCertificate(rand.Reader, caTemplate, caTemplate, &caKey.PublicKey, caKey) + if err != nil { + t.Fatal(err) + } + caCert, err := x509.ParseCertificate(caDER) + if err != nil { + t.Fatal(err) + } + makeLeaf := func(serial int64, dns string, usage x509.ExtKeyUsage) tls.Certificate { + key, keyErr := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if keyErr != nil { + t.Fatal(keyErr) + } + template := &x509.Certificate{SerialNumber: big.NewInt(serial), Subject: pkix.Name{CommonName: dns}, DNSNames: []string{dns}, NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(time.Hour), ExtKeyUsage: []x509.ExtKeyUsage{usage}, KeyUsage: x509.KeyUsageDigitalSignature} + der, createErr := x509.CreateCertificate(rand.Reader, template, caCert, &key.PublicKey, caKey) + if createErr != nil { + t.Fatal(createErr) + } + return tls.Certificate{Certificate: [][]byte{der, caDER}, PrivateKey: key} + } + serverCert := makeLeaf(2, "gateway.test", x509.ExtKeyUsageServerAuth) + clientCert := makeLeaf(3, "client.test", x509.ExtKeyUsageClientAuth) + pool := x509.NewCertPool() + pool.AddCert(caCert) + return &tls.Config{MinVersion: tls.VersionTLS13, Certificates: []tls.Certificate{serverCert}, ClientAuth: tls.RequireAndVerifyClientCert, ClientCAs: pool}, &tls.Config{MinVersion: tls.VersionTLS13, Certificates: []tls.Certificate{clientCert}, RootCAs: pool, ServerName: "gateway.test"} +} + +type oneTimeAdmission struct { + used atomic.Bool + authority protocol.SessionAuthority + releases atomic.Int64 + released chan struct{} +} + +func (a *oneTimeAdmission) Admit(context.Context, protocol.TunnelAdmissionRequest) (protocol.SessionAuthority, error) { + if !a.used.CompareAndSwap(false, true) { + return protocol.SessionAuthority{}, ErrAdmissionRejected + } + return a.authority, nil +} + +func (a *oneTimeAdmission) Release(context.Context, protocol.SessionAuthority) error { + if a.releases.Add(1) == 1 { + close(a.released) + } + return nil +} + +func mustRead(t *testing.T, path string) []byte { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + return data +} + +func bytesRepeat(value byte, count int) []byte { + data := make([]byte, count) + for index := range data { + data[index] = value + } + return data +} diff --git a/gateway/input.go b/gateway/input.go new file mode 100644 index 0000000..fc91c66 --- /dev/null +++ b/gateway/input.go @@ -0,0 +1,38 @@ +package gateway + +import ( + "encoding/binary" + "errors" +) + +var ErrInputMalformed = errors.New("input event malformed") + +func EncodeInputEvent(event InputEvent) ([]byte, error) { + if len(event.Device) == 0 || len(event.Device) > 64 || len(event.Payload) > 1024 { + return nil, ErrInputMalformed + } + encoded := make([]byte, 16+len(event.Device)+len(event.Payload)) + copy(encoded[:4], "INP1") + binary.BigEndian.PutUint32(encoded[4:8], event.Sequence) + binary.BigEndian.PutUint32(encoded[8:12], uint32(event.Code)) + if event.Pressed { + encoded[12] = 1 + } + encoded[13] = byte(len(event.Device)) + binary.BigEndian.PutUint16(encoded[14:16], uint16(len(event.Payload))) + copy(encoded[16:16+len(event.Device)], event.Device) + copy(encoded[16+len(event.Device):], event.Payload) + return encoded, nil +} + +func DecodeInputEvent(data []byte) (InputEvent, error) { + if len(data) < 16 || len(data) > 1179 || string(data[:4]) != "INP1" || (data[12] != 0 && data[12] != 1) { + return InputEvent{}, ErrInputMalformed + } + deviceLength := int(data[13]) + payloadLength := int(binary.BigEndian.Uint16(data[14:16])) + if deviceLength == 0 || deviceLength > 64 || payloadLength > 1024 || len(data) != 16+deviceLength+payloadLength { + return InputEvent{}, ErrInputMalformed + } + return InputEvent{Sequence: binary.BigEndian.Uint32(data[4:8]), Code: int32(binary.BigEndian.Uint32(data[8:12])), Pressed: data[12] == 1, Device: string(data[16 : 16+deviceLength]), Payload: append([]byte(nil), data[16+deviceLength:]...)}, nil +} diff --git a/gateway/provider.go b/gateway/provider.go new file mode 100644 index 0000000..073615c --- /dev/null +++ b/gateway/provider.go @@ -0,0 +1,519 @@ +package gateway + +import ( + "context" + "encoding/binary" + "encoding/xml" + "errors" + "fmt" + "strings" + "sync" + "time" + + protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol" +) + +const ( + ProviderProfileApollo = "apollo" + ProviderStateStarting = "starting" + ProviderStateReady = "ready" + ProviderStateDisconnected = "disconnected" + ProviderStateTerminating = "terminating" + ProviderStateTerminated = "terminated" + ProviderStateCleanup = "cleanup_pending" + ProviderStateFailed = "failed" +) + +var ( + ErrProviderIdentity = errors.New("provider identity rejected") + ErrProviderMalformed = errors.New("provider response malformed") + ErrProviderTimeout = errors.New("provider operation timed out") + ErrProviderDisconnected = errors.New("provider disconnected") + ErrProviderCleanup = errors.New("provider cleanup pending") + ErrProviderTerminated = errors.New("provider session terminated") +) + +type ProviderIdentity struct { + UniqueID string + Fingerprint string + NotBefore time.Time + NotAfter time.Time +} + +func (i ProviderIdentity) Key() string { + return i.UniqueID + "#" + i.Fingerprint +} + +func (i ProviderIdentity) Validate(now time.Time, expected ProviderIdentity) error { + if i.UniqueID == "" || i.Fingerprint == "" || i.UniqueID != expected.UniqueID || i.Fingerprint != expected.Fingerprint { + return ErrProviderIdentity + } + if !i.NotBefore.IsZero() && now.Before(i.NotBefore) { + return ErrProviderIdentity + } + if !i.NotAfter.IsZero() && !now.Before(i.NotAfter) { + return ErrProviderIdentity + } + return nil +} + +type ManagementInfo struct { + Identity ProviderIdentity + Name string +} + +func ParseManagementXML(data []byte) (ManagementInfo, error) { + if len(data) == 0 || len(data) > 64*1024 { + return ManagementInfo{}, ErrProviderMalformed + } + var document struct { + XMLName xml.Name `xml:"root"` + UniqueID string `xml:"unique_id"` + Fingerprint string `xml:"fingerprint"` + NotBefore string `xml:"not_before"` + NotAfter string `xml:"not_after"` + Name string `xml:"name"` + } + decoder := xml.NewDecoder(strings.NewReader(string(data))) + decoder.Strict = true + if err := decoder.Decode(&document); err != nil { + return ManagementInfo{}, fmt.Errorf("%w: %v", ErrProviderMalformed, err) + } + identity := ProviderIdentity{UniqueID: document.UniqueID, Fingerprint: document.Fingerprint} + var err error + if document.NotBefore != "" { + identity.NotBefore, err = time.Parse(time.RFC3339Nano, document.NotBefore) + if err != nil { + return ManagementInfo{}, ErrProviderMalformed + } + } + if document.NotAfter != "" { + identity.NotAfter, err = time.Parse(time.RFC3339Nano, document.NotAfter) + if err != nil { + return ManagementInfo{}, ErrProviderMalformed + } + } + if identity.UniqueID == "" || len(identity.UniqueID) > 128 || identity.Fingerprint == "" || len(identity.Fingerprint) > 256 { + return ManagementInfo{}, ErrProviderMalformed + } + return ManagementInfo{Identity: identity, Name: document.Name}, nil +} + +type RTSPResponse struct { + StatusCode int + Session string + Transport string +} + +func ParseRTSPResponse(data []byte) (RTSPResponse, error) { + if len(data) == 0 || len(data) > 16*1024 { + return RTSPResponse{}, ErrProviderMalformed + } + text := string(data) + if !strings.Contains(text, "\r\n") { + return RTSPResponse{}, ErrProviderMalformed + } + lines := strings.Split(text, "\r\n") + if len(lines) < 2 { + return RTSPResponse{}, ErrProviderMalformed + } + statusParts := strings.SplitN(lines[0], " ", 3) + if len(statusParts) < 2 || statusParts[0] != "RTSP/1.0" { + return RTSPResponse{}, ErrProviderMalformed + } + var response RTSPResponse + if _, err := fmt.Sscanf(statusParts[1], "%d", &response.StatusCode); err != nil || response.StatusCode != 200 { + return RTSPResponse{}, ErrProviderMalformed + } + for _, line := range lines[1:] { + if line == "" { + break + } + key, value, ok := strings.Cut(line, ":") + if !ok { + return RTSPResponse{}, ErrProviderMalformed + } + switch strings.ToLower(strings.TrimSpace(key)) { + case "session": + response.Session = strings.TrimSpace(value) + case "transport": + response.Transport = strings.TrimSpace(value) + } + } + if response.Session == "" || response.Transport == "" || len(response.Session) > 256 || len(response.Transport) > 1024 { + return RTSPResponse{}, ErrProviderMalformed + } + return response, nil +} + +type ControlPacket struct { + Kind byte + Sequence uint32 + Payload []byte +} + +func EncodeControlPacket(packet ControlPacket) ([]byte, error) { + if len(packet.Payload) > 4096 { + return nil, ErrProviderMalformed + } + encoded := make([]byte, 11+len(packet.Payload)) + copy(encoded[:4], "APC1") + encoded[4] = packet.Kind + binary.BigEndian.PutUint32(encoded[5:9], packet.Sequence) + binary.BigEndian.PutUint16(encoded[9:11], uint16(len(packet.Payload))) + copy(encoded[11:], packet.Payload) + return encoded, nil +} + +func DecodeControlPacket(data []byte) (ControlPacket, error) { + if len(data) < 11 || len(data) > 4107 || string(data[:4]) != "APC1" { + return ControlPacket{}, ErrProviderMalformed + } + length := int(binary.BigEndian.Uint16(data[9:11])) + if length > 4096 || len(data) != 11+length { + return ControlPacket{}, ErrProviderMalformed + } + return ControlPacket{Kind: data[4], Sequence: binary.BigEndian.Uint32(data[5:9]), Payload: append([]byte(nil), data[11:]...)}, nil +} + +type LaunchRequest struct { + SessionID string + Capabilities protocol.CapabilityProfile + ProviderProfile string + ProviderIdentity string +} + +type InputEvent struct { + Sequence uint32 + Device string + Code int32 + Pressed bool + Payload []byte +} + +type Feedback struct { + Sequence uint32 + Payload []byte +} + +type Provider interface { + Start(context.Context, LaunchRequest) (ProviderSession, error) +} + +type ProviderSession interface { + Ready(context.Context) error + Video() <-chan []byte + Audio() <-chan []byte + Input(context.Context, InputEvent) error + Feedback(context.Context, Feedback) error + Reconnect(context.Context) error + ReleaseAll(context.Context) error + Terminate(context.Context) error + State() protocol.ProviderState +} + +type ApolloBackend interface { + Management(context.Context) ([]byte, error) + Setup(context.Context, LaunchRequest) ([]byte, error) + Open(context.Context, LaunchRequest, RTSPResponse) (ProviderSession, error) +} + +type ApolloAdapter struct { + backend ApolloBackend + expected ProviderIdentity + now func() time.Time + readyTTL time.Duration +} + +func NewApolloAdapter(backend ApolloBackend, expected ProviderIdentity) *ApolloAdapter { + return &ApolloAdapter{backend: backend, expected: expected, now: time.Now, readyTTL: 2 * time.Second} +} + +func (a *ApolloAdapter) Start(ctx context.Context, request LaunchRequest) (ProviderSession, error) { + if a == nil || a.backend == nil || request.ProviderProfile != ProviderProfileApollo { + return nil, ErrProviderIdentity + } + management, err := a.backend.Management(ctx) + if err != nil { + return nil, err + } + info, err := ParseManagementXML(management) + if err != nil { + return nil, err + } + if err := info.Identity.Validate(a.now(), a.expected); err != nil { + return nil, err + } + if request.ProviderIdentity != "" && request.ProviderIdentity != info.Identity.Key() { + return nil, ErrProviderIdentity + } + rawRTSP, err := a.backend.Setup(ctx, request) + if err != nil { + return nil, err + } + rtsp, err := ParseRTSPResponse(rawRTSP) + if err != nil { + return nil, err + } + session, err := a.backend.Open(ctx, request, rtsp) + if err != nil { + return nil, err + } + readyCtx, cancel := context.WithTimeout(ctx, a.readyTTL) + defer cancel() + if err := session.Ready(readyCtx); err != nil { + cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), time.Second) + cleanupErr := session.Terminate(cleanupCtx) + cleanupCancel() + if cleanupErr != nil { + return nil, fmt.Errorf("%w: %v", ErrProviderCleanup, err) + } + if errors.Is(err, context.DeadlineExceeded) { + return nil, ErrProviderTimeout + } + return nil, err + } + return session, nil +} + +type FakeFailure string + +const ( + FakeFailureNone FakeFailure = "" + FakeFailureIdentity FakeFailure = "identity" + FakeFailureMalformed FakeFailure = "malformed" + FakeFailureReadinessTimeout FakeFailure = "readiness-timeout" + FakeFailureProviderDisconnect FakeFailure = "provider-disconnect" + FakeFailureTerminationTimeout FakeFailure = "termination-timeout" +) + +type FakeApolloConfig struct { + Identity ProviderIdentity + Failure FakeFailure + Video [][]byte + Audio [][]byte + Now time.Time +} + +type FakeApollo struct { + config FakeApolloConfig + mu sync.Mutex + last *fakeSession +} + +func NewFakeApollo(config FakeApolloConfig) *FakeApollo { + if config.Identity.UniqueID == "" { + config.Identity.UniqueID = "apollo-fixture-1" + } + if config.Identity.Fingerprint == "" { + config.Identity.Fingerprint = "sha256:fixture-apollo-1" + } + if config.Now.IsZero() { + config.Now = time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) + } + if len(config.Video) == 0 { + config.Video = [][]byte{[]byte{0x00, 0x00, 0x01, 0x65, 0x01, 0x02}} + } + if len(config.Audio) == 0 { + config.Audio = [][]byte{[]byte{0x4f, 0x70, 0x75, 0x73, 0x01}} + } + return &FakeApollo{config: config} +} + +func (f *FakeApollo) Management(context.Context) ([]byte, error) { + if f.config.Failure == FakeFailureMalformed { + return []byte(""), nil + } + identity := f.config.Identity + if f.config.Failure == FakeFailureIdentity { + identity.Fingerprint = "sha256:changed-fixture" + } + return []byte(fmt.Sprintf("%s%s%s%sfixture-apollo", identity.UniqueID, identity.Fingerprint, f.config.Now.Add(-time.Hour).Format(time.RFC3339), f.config.Now.Add(time.Hour).Format(time.RFC3339))), nil +} + +func (f *FakeApollo) Setup(context.Context, LaunchRequest) ([]byte, error) { + if f.config.Failure == FakeFailureMalformed { + return []byte("RTSP/1.0 200 OK\r\n\r\n"), nil + } + return []byte("RTSP/1.0 200 OK\r\nSession: fixture-session\r\nTransport: RTP/AVP/TCP;interleaved=0-1\r\n\r\n"), nil +} + +func (f *FakeApollo) Open(context.Context, LaunchRequest, RTSPResponse) (ProviderSession, error) { + session := &fakeSession{ + failure: f.config.Failure, + video: make(chan []byte, 16), + audio: make(chan []byte, 16), + state: protocol.ProviderState{Version: "1", State: ProviderStateStarting, Channels: []string{"video", "audio", "input", "feedback"}}, + pressed: make(map[string]struct{}), + } + for _, payload := range f.config.Video { + session.EmitVideo(payload) + } + for _, payload := range f.config.Audio { + session.EmitAudio(payload) + } + f.mu.Lock() + f.last = session + f.mu.Unlock() + return session, nil +} + +func (f *FakeApollo) Start(ctx context.Context, request LaunchRequest) (ProviderSession, error) { + adapter := NewApolloAdapter(f, f.config.Identity) + adapter.now = func() time.Time { return f.config.Now } + return adapter.Start(ctx, request) +} + +func (f *FakeApollo) LastSession() ProviderSession { + f.mu.Lock() + defer f.mu.Unlock() + return f.last +} + +func (f *FakeApollo) DisconnectProvider() { + f.mu.Lock() + session := f.last + f.mu.Unlock() + if session != nil { + session.Disconnect() + } +} + +type fakeSession struct { + mu sync.Mutex + failure FakeFailure + video chan []byte + audio chan []byte + state protocol.ProviderState + pressed map[string]struct{} + inputs []InputEvent + feedback []Feedback + releaseAll int + closeOnce sync.Once +} + +func (s *fakeSession) Ready(ctx context.Context) error { + if s.failure == FakeFailureReadinessTimeout { + <-ctx.Done() + return ctx.Err() + } + s.mu.Lock() + defer s.mu.Unlock() + if s.state.State == ProviderStateDisconnected { + return ErrProviderDisconnected + } + s.state.State = ProviderStateReady + return nil +} + +func (s *fakeSession) Video() <-chan []byte { return s.video } +func (s *fakeSession) Audio() <-chan []byte { return s.audio } + +func (s *fakeSession) EmitVideo(payload []byte) { + select { + case s.video <- append([]byte(nil), payload...): + default: + <-s.video + s.video <- append([]byte(nil), payload...) + } +} + +func (s *fakeSession) EmitAudio(payload []byte) { + select { + case s.audio <- append([]byte(nil), payload...): + default: + <-s.audio + s.audio <- append([]byte(nil), payload...) + } +} + +func (s *fakeSession) Input(_ context.Context, event InputEvent) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.state.State != ProviderStateReady { + return ErrProviderDisconnected + } + s.inputs = append(s.inputs, event) + key := fmt.Sprintf("%s:%d", event.Device, event.Code) + if event.Pressed { + s.pressed[key] = struct{}{} + } else { + delete(s.pressed, key) + } + return nil +} + +func (s *fakeSession) Feedback(_ context.Context, feedback Feedback) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.state.State != ProviderStateReady { + return ErrProviderDisconnected + } + s.feedback = append(s.feedback, feedback) + return nil +} + +func (s *fakeSession) Reconnect(_ context.Context) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.state.State == ProviderStateTerminated { + return ErrProviderTerminated + } + s.state.State = ProviderStateReady + return nil +} + +func (s *fakeSession) ReleaseAll(_ context.Context) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.failure == FakeFailureProviderDisconnect { + return ErrProviderDisconnected + } + s.pressed = make(map[string]struct{}) + s.releaseAll++ + return nil +} + +func (s *fakeSession) Terminate(ctx context.Context) error { + if s.failure == FakeFailureTerminationTimeout { + <-ctx.Done() + s.mu.Lock() + s.state.State = ProviderStateCleanup + s.state.CleanupPending = true + s.mu.Unlock() + return ctx.Err() + } + s.mu.Lock() + if s.state.State == ProviderStateTerminated { + s.mu.Unlock() + return nil + } + s.state.State = ProviderStateTerminating + s.mu.Unlock() + s.closeOnce.Do(func() { + close(s.video) + close(s.audio) + }) + s.mu.Lock() + s.state.State = ProviderStateTerminated + s.mu.Unlock() + return nil +} + +func (s *fakeSession) State() protocol.ProviderState { + s.mu.Lock() + defer s.mu.Unlock() + return s.state +} + +func (s *fakeSession) Disconnect() { + s.mu.Lock() + s.state.State = ProviderStateDisconnected + s.mu.Unlock() +} + +func (s *fakeSession) ReleaseCount() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.releaseAll +} diff --git a/gateway/queue.go b/gateway/queue.go new file mode 100644 index 0000000..32de58d --- /dev/null +++ b/gateway/queue.go @@ -0,0 +1,98 @@ +package gateway + +import ( + "context" + "errors" + "sync" +) + +var ErrQueueClosed = errors.New("gateway queue closed") + +// BoundedQueue is deliberately fixed-size. Media uses PushLatest so a slow +// client drops old frames instead of allowing provider output to accumulate. +type BoundedQueue[T any] struct { + mu sync.Mutex + items []T + limit int + dropped uint64 + closed bool + wake chan struct{} +} + +func NewBoundedQueue[T any](limit int) *BoundedQueue[T] { + if limit < 1 { + limit = 1 + } + return &BoundedQueue[T]{limit: limit, wake: make(chan struct{}, 1)} +} + +func (q *BoundedQueue[T]) PushLatest(item T) error { + q.mu.Lock() + defer q.mu.Unlock() + if q.closed { + return ErrQueueClosed + } + if len(q.items) == q.limit { + var zero T + q.items[0] = zero + q.items = q.items[1:] + q.dropped++ + } + q.items = append(q.items, item) + select { + case q.wake <- struct{}{}: + default: + } + return nil +} + +func (q *BoundedQueue[T]) Pop(ctx context.Context) (T, error) { + for { + q.mu.Lock() + if len(q.items) > 0 { + item := q.items[0] + q.items[0] = *new(T) + q.items = q.items[1:] + q.mu.Unlock() + return item, nil + } + if q.closed { + q.mu.Unlock() + var zero T + return zero, ErrQueueClosed + } + q.mu.Unlock() + select { + case <-ctx.Done(): + var zero T + return zero, ctx.Err() + case <-q.wake: + } + } +} + +func (q *BoundedQueue[T]) Close() { + q.mu.Lock() + if q.closed { + q.mu.Unlock() + return + } + q.closed = true + select { + case q.wake <- struct{}{}: + default: + } + q.mu.Unlock() +} + +func (q *BoundedQueue[T]) Dropped() uint64 { + q.mu.Lock() + defer q.mu.Unlock() + return q.dropped +} + +func (q *BoundedQueue[T]) Len() int { + q.mu.Lock() + defer q.mu.Unlock() + return len(q.items) +} diff --git a/gateway/telemetry.go b/gateway/telemetry.go new file mode 100644 index 0000000..a61825e --- /dev/null +++ b/gateway/telemetry.go @@ -0,0 +1,67 @@ +package gateway + +import ( + "context" + "sync/atomic" + "time" +) + +type Metrics struct { + ActiveSessions atomic.Int64 + AdmissionRejects atomic.Uint64 + MediaDrops atomic.Uint64 + ProviderErrors atomic.Uint64 + InputRejected atomic.Uint64 +} + +type MetricsSnapshot struct { + ActiveSessions int64 + AdmissionRejects uint64 + MediaDrops uint64 + ProviderErrors uint64 + InputRejected uint64 +} + +func (m *Metrics) Snapshot() MetricsSnapshot { + return MetricsSnapshot{ + ActiveSessions: m.ActiveSessions.Load(), + AdmissionRejects: m.AdmissionRejects.Load(), + MediaDrops: m.MediaDrops.Load(), + ProviderErrors: m.ProviderErrors.Load(), + InputRejected: m.InputRejected.Load(), + } +} + +type Pacer struct { + bytesPerSecond int64 + last time.Time +} + +func NewPacer(kbps int64) *Pacer { + if kbps < 1 { + return &Pacer{} + } + return &Pacer{bytesPerSecond: kbps * 1000 / 8} +} + +func (p *Pacer) Wait(ctx context.Context, bytes int) error { + if p.bytesPerSecond < 1 || bytes < 1 { + return nil + } + now := time.Now() + if p.last.IsZero() || now.After(p.last) { + p.last = now + } + delay := time.Duration(float64(bytes) / float64(p.bytesPerSecond) * float64(time.Second)) + p.last = p.last.Add(delay) + if wait := time.Until(p.last); wait > 0 { + timer := time.NewTimer(wait) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + } + } + return nil +} diff --git a/gateway/testdata/apollo-management.xml b/gateway/testdata/apollo-management.xml new file mode 100644 index 0000000..a2e19f5 --- /dev/null +++ b/gateway/testdata/apollo-management.xml @@ -0,0 +1 @@ +apollo-fixture-1sha256:fixture-apollo-12025-12-31T23:00:00Z2026-01-01T01:00:00Zfixture-apollo diff --git a/gateway/testdata/encoded-audio.hex b/gateway/testdata/encoded-audio.hex new file mode 100644 index 0000000..1b0ad4d --- /dev/null +++ b/gateway/testdata/encoded-audio.hex @@ -0,0 +1 @@ +4f70757301 diff --git a/gateway/testdata/encoded-video.hex b/gateway/testdata/encoded-video.hex new file mode 100644 index 0000000..f37bec4 --- /dev/null +++ b/gateway/testdata/encoded-video.hex @@ -0,0 +1 @@ +000001650102 diff --git a/gateway/testdata/input-release.hex b/gateway/testdata/input-release.hex new file mode 100644 index 0000000..577ae3e --- /dev/null +++ b/gateway/testdata/input-release.hex @@ -0,0 +1 @@ +494e503100000001000000070000014b6579626f617264 diff --git a/gateway/testdata/rtsp-setup-response.txt b/gateway/testdata/rtsp-setup-response.txt new file mode 100644 index 0000000..8fc30e2 --- /dev/null +++ b/gateway/testdata/rtsp-setup-response.txt @@ -0,0 +1 @@ +RTSP/1.0 200 OK\r\nSession: fixture-session\r\nTransport: RTP/AVP/TCP;interleaved=0-1\r\n\r\n diff --git a/gateway/transport.go b/gateway/transport.go new file mode 100644 index 0000000..90d37b6 --- /dev/null +++ b/gateway/transport.go @@ -0,0 +1,628 @@ +package gateway + +import ( + "context" + "crypto/tls" + "encoding/base64" + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "sync" + "sync/atomic" + "time" + + "github.com/quic-go/quic-go" + + protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol" +) + +const ( + defaultHelloLimit = 16 * 1024 + defaultControlLimit = 128 * 1024 + applicationError = quic.ApplicationErrorCode(0x100) +) + +var ( + ErrAdmissionRejected = errors.New("gateway admission rejected") + ErrGatewayDraining = errors.New("gateway draining") + ErrGatewayTLS = errors.New("gateway requires TLS 1.3 client authentication") + ErrAuthorityExpired = errors.New("gateway authority expired") +) + +type Admission interface { + Admit(context.Context, protocol.TunnelAdmissionRequest) (protocol.SessionAuthority, error) + Release(context.Context, protocol.SessionAuthority) error +} + +type AdmissionFunc func(context.Context, protocol.TunnelAdmissionRequest) (protocol.SessionAuthority, error) + +func (f AdmissionFunc) Admit(ctx context.Context, request protocol.TunnelAdmissionRequest) (protocol.SessionAuthority, error) { + return f(ctx, request) +} + +func (AdmissionFunc) Release(context.Context, protocol.SessionAuthority) error { return nil } + +type ServerConfig struct { + ListenAddress string + TLSConfig *tls.Config + QUICConfig *quic.Config + GatewayID string + Capabilities protocol.CapabilityProfile + ProviderCapabilities protocol.CapabilityProfile + Admission Admission + Provider Provider + ProviderProfile string + ProviderIdentity string + PacerKbps int64 +} + +type Server struct { + listener *quic.Listener + config ServerConfig + metrics *Metrics + mu sync.Mutex + sessions map[*gatewaySession]struct{} + draining atomic.Bool + closed atomic.Bool + closeOnce sync.Once + workers sync.WaitGroup +} + +func NewServer(config ServerConfig) (*Server, error) { + if config.ListenAddress == "" { + config.ListenAddress = "127.0.0.1:0" + } + if config.GatewayID == "" || config.Admission == nil || config.Provider == nil { + return nil, errors.New("gateway id, admission, and provider are required") + } + if err := validateServerTLS(config.TLSConfig); err != nil { + return nil, err + } + if config.Capabilities == (protocol.CapabilityProfile{}) { + config.Capabilities = DefaultCapabilities() + } + if config.ProviderCapabilities == (protocol.CapabilityProfile{}) { + config.ProviderCapabilities = DefaultCapabilities() + } + if config.ProviderProfile == "" { + config.ProviderProfile = ProviderProfileApollo + } + if config.PacerKbps < 0 { + return nil, errors.New("negative pacing limit") + } + tlsConfig := config.TLSConfig.Clone() + if len(tlsConfig.NextProtos) == 0 { + tlsConfig.NextProtos = []string{"versevdi-gateway-v1"} + } + quicConfig := &quic.Config{EnableDatagrams: true, MaxIdleTimeout: 30 * time.Second, MaxIncomingStreams: 2, MaxIncomingUniStreams: 2} + if config.QUICConfig != nil { + quicConfig = config.QUICConfig.Clone() + quicConfig.EnableDatagrams = true + } + listener, err := quic.ListenAddr(config.ListenAddress, tlsConfig, quicConfig) + if err != nil { + return nil, err + } + return &Server{listener: listener, config: config, metrics: &Metrics{}, sessions: make(map[*gatewaySession]struct{})}, nil +} + +func validateServerTLS(config *tls.Config) error { + if config == nil || config.MinVersion < tls.VersionTLS13 || config.ClientAuth != tls.RequireAndVerifyClientCert || config.ClientCAs == nil || len(config.Certificates) == 0 { + return ErrGatewayTLS + } + return nil +} + +func (s *Server) Addr() net.Addr { return s.listener.Addr() } +func (s *Server) Metrics() MetricsSnapshot { return s.metrics.Snapshot() } +func (s *Server) Draining() bool { return s.draining.Load() } + +func (s *Server) BeginDrain() { + s.draining.Store(true) +} + +func (s *Server) Serve(ctx context.Context) error { + if s.closed.Load() { + return net.ErrClosed + } + go func() { + <-ctx.Done() + _ = s.Close() + }() + for { + connection, err := s.listener.Accept(ctx) + if err != nil { + if s.closed.Load() || errors.Is(err, context.Canceled) || errors.Is(err, net.ErrClosed) { + return nil + } + return err + } + s.workers.Add(1) + go func() { + defer s.workers.Done() + s.handleConnection(ctx, connection) + }() + } +} + +func (s *Server) Close() error { + var err error + s.closeOnce.Do(func() { + s.BeginDrain() + s.closed.Store(true) + err = s.listener.Close() + s.mu.Lock() + for session := range s.sessions { + session.cancel() + } + s.mu.Unlock() + }) + s.workers.Wait() + return err +} + +func (s *Server) handleConnection(parent context.Context, connection *quic.Conn) { + defer connection.CloseWithError(applicationError, "connection closed") + ctx, cancel := context.WithTimeout(parent, 10*time.Second) + defer cancel() + stream, err := connection.AcceptStream(ctx) + if err != nil { + return + } + requestBytes, err := readWire(stream, defaultHelloLimit) + if err != nil { + _ = writeStableError(stream, "invalid_hello", err, false) + return + } + request, err := protocol.DecodeTunnelAdmissionRequest(requestBytes) + if err != nil { + _ = writeStableError(stream, "invalid_hello", err, false) + return + } + if s.Draining() { + _ = writeStableError(stream, "gateway_draining", ErrGatewayDraining, true) + return + } + if request.GatewayID != s.config.GatewayID { + _ = writeStableError(stream, "wrong_gateway", ErrAdmissionRejected, false) + return + } + authority, err := s.config.Admission.Admit(ctx, request) + if err != nil { + s.metrics.AdmissionRejects.Add(1) + _ = writeStableError(stream, stableAdmissionCode(err), err, errors.Is(err, context.DeadlineExceeded)) + return + } + if s.Draining() { + _ = s.config.Admission.Release(context.Background(), authority) + _ = writeStableError(stream, "gateway_draining", ErrGatewayDraining, true) + return + } + if err := s.validateAuthority(authority, request); err != nil { + _ = s.config.Admission.Release(context.Background(), authority) + _ = writeStableError(stream, "invalid_authority", err, false) + return + } + selected, err := IntersectCapabilities(s.config.Capabilities, s.config.ProviderCapabilities, request.Capabilities, authority.Capabilities) + if err != nil { + _ = s.config.Admission.Release(context.Background(), authority) + s.metrics.AdmissionRejects.Add(1) + _ = writeStableError(stream, "no_capability_overlap", err, false) + return + } + providerSession, err := s.config.Provider.Start(ctx, LaunchRequest{SessionID: request.SessionID, Capabilities: selected, ProviderProfile: authority.ProviderProfile, ProviderIdentity: authority.ProviderIdentity}) + if err != nil { + s.metrics.ProviderErrors.Add(1) + _ = s.config.Admission.Release(context.Background(), authority) + _ = writeStableError(stream, stableProviderCode(err), err, errors.Is(err, context.DeadlineExceeded)) + return + } + authority.Capabilities = selected + authorityBytes, err := protocol.EncodeSessionAuthority(authority) + if err != nil || writeWire(stream, authorityBytes, defaultHelloLimit) != nil { + _ = providerSession.ReleaseAll(context.Background()) + _ = providerSession.Terminate(context.Background()) + _ = s.config.Admission.Release(context.Background(), authority) + return + } + session := newGatewaySession(s, connection, stream, request, authority, providerSession) + s.addSession(session) + s.metrics.ActiveSessions.Add(1) + defer func() { + s.removeSession(session) + s.metrics.ActiveSessions.Add(-1) + }() + session.run() +} + +func (s *Server) validateAuthority(authority protocol.SessionAuthority, request protocol.TunnelAdmissionRequest) error { + if err := authority.Validate(); err != nil { + return err + } + if authority.SessionID != request.SessionID || authority.GatewayID != request.GatewayID || authority.Audience != request.Audience || authority.ProviderProfile != s.config.ProviderProfile { + return ErrAdmissionRejected + } + expires, err := time.Parse(time.RFC3339Nano, authority.ExpiresAt) + if err != nil || !time.Now().Before(expires) { + return ErrAuthorityExpired + } + return nil +} + +func (s *Server) addSession(session *gatewaySession) { + s.mu.Lock() + s.sessions[session] = struct{}{} + s.mu.Unlock() +} + +func (s *Server) removeSession(session *gatewaySession) { + s.mu.Lock() + delete(s.sessions, session) + s.mu.Unlock() +} + +type gatewaySession struct { + server *Server + connection *quic.Conn + control *quic.Stream + request protocol.TunnelAdmissionRequest + authority protocol.SessionAuthority + provider ProviderSession + pacer *Pacer + ctx context.Context + cancel context.CancelFunc + cleanupOnce sync.Once + inputMu sync.Mutex + pressed map[string]struct{} + sequence atomic.Uint32 + result chan error +} + +func newGatewaySession(server *Server, connection *quic.Conn, control *quic.Stream, request protocol.TunnelAdmissionRequest, authority protocol.SessionAuthority, provider ProviderSession) *gatewaySession { + ctx, cancel := context.WithCancel(context.Background()) + return &gatewaySession{server: server, connection: connection, control: control, request: request, authority: authority, provider: provider, pacer: NewPacer(server.config.PacerKbps), ctx: ctx, cancel: cancel, pressed: make(map[string]struct{}), result: make(chan error, 3)} +} + +func (s *gatewaySession) run() { + defer s.cleanup() + deadline, err := time.Parse(time.RFC3339Nano, s.authority.ExpiresAt) + if err != nil { + return + } + if deadline.Before(time.Now()) { + return + } + timer := time.NewTimer(time.Until(deadline)) + defer timer.Stop() + go s.controlLoop() + go s.datagramLoop() + go s.mediaLoop() + select { + case <-timer.C: + s.server.metrics.InputRejected.Add(1) + case <-s.ctx.Done(): + case <-s.result: + } + s.cancel() +} + +func (s *gatewaySession) controlLoop() { + for { + data, err := readWire(s.control, defaultControlLimit) + if err != nil { + s.result <- err + return + } + frame, err := protocol.DecodeChannelFrame(data) + if err != nil { + s.result <- err + return + } + payload, err := base64.StdEncoding.DecodeString(frame.Payload) + if err != nil || len(payload) > maxFrameSize { + s.result <- ErrFramePayloadLimit + return + } + switch frame.FlowID { + case "control": + if err := s.handleControl(payload); err != nil { + s.result <- err + return + } + case "input": + if err := s.handleInput(payload); err != nil { + s.result <- err + return + } + default: + s.result <- ErrFrameChannel + return + } + } +} + +func (s *gatewaySession) datagramLoop() { + for { + data, err := s.connection.ReceiveDatagram(s.ctx) + if err != nil { + s.result <- err + return + } + frame, err := DecodeFrame(data) + if err != nil { + s.result <- err + return + } + switch frame.Channel { + case ChannelInput: + if err := s.handleInput(frame.Payload); err != nil { + s.result <- err + return + } + case ChannelText: + if len(frame.Payload) > 4096 { + s.result <- ErrFramePayloadLimit + return + } + if err := s.provider.Feedback(s.ctx, Feedback{Sequence: frame.Sequence, Payload: append([]byte(nil), frame.Payload...)}); err != nil { + s.result <- err + return + } + default: + s.result <- ErrFrameChannel + return + } + } +} + +func (s *gatewaySession) mediaLoop() { + video, audio := s.provider.Video(), s.provider.Audio() + for video != nil || audio != nil { + select { + case <-s.ctx.Done(): + return + case payload, ok := <-video: + if !ok { + video = nil + continue + } + if err := s.sendMedia(ChannelVideo, payload); err != nil { + s.result <- err + return + } + case payload, ok := <-audio: + if !ok { + audio = nil + continue + } + if err := s.sendMedia(ChannelAudio, payload); err != nil { + s.result <- err + return + } + } + } + s.result <- ErrProviderDisconnected +} + +func (s *gatewaySession) sendMedia(channel byte, payload []byte) error { + frames, err := FragmentPayload(channel, s.sequence.Add(1), uint64(time.Now().UnixMilli()), payload) + if err != nil { + return err + } + for _, frame := range frames { + encoded, err := EncodeFrame(frame) + if err != nil { + return err + } + if err := s.pacer.Wait(s.ctx, len(encoded)); err != nil { + return err + } + if err := s.connection.SendDatagram(encoded); err != nil { + return err + } + } + return nil +} + +func (s *gatewaySession) handleControl(payload []byte) error { + if len(payload) < 4 { + return ErrProviderMalformed + } + switch string(payload[:4]) { + case "TERM": + return errors.New("client requested termination") + case "RECN": + return s.provider.Reconnect(s.ctx) + case "FBRK": + return s.provider.Feedback(s.ctx, Feedback{Payload: append([]byte(nil), payload[4:]...)}) + default: + return ErrProviderMalformed + } +} + +func (s *gatewaySession) handleInput(payload []byte) error { + event, err := DecodeInputEvent(payload) + if err != nil { + return err + } + if err := s.provider.Input(s.ctx, event); err != nil { + s.server.metrics.InputRejected.Add(1) + return err + } + key := fmt.Sprintf("%s:%d", event.Device, event.Code) + s.inputMu.Lock() + if event.Pressed { + s.pressed[key] = struct{}{} + } else { + delete(s.pressed, key) + } + s.inputMu.Unlock() + return nil +} + +func (s *gatewaySession) cleanup() { + s.cleanupOnce.Do(func() { + s.cancel() + cleanupCtx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := s.provider.ReleaseAll(cleanupCtx); err != nil { + s.server.metrics.ProviderErrors.Add(1) + } + if err := s.provider.Terminate(cleanupCtx); err != nil { + s.server.metrics.ProviderErrors.Add(1) + } + if err := s.server.config.Admission.Release(cleanupCtx, s.authority); err != nil { + s.server.metrics.ProviderErrors.Add(1) + } + _ = s.connection.CloseWithError(applicationError, "session closed") + }) +} + +func writeStableError(writer io.Writer, code string, err error, retryable bool) error { + message := err.Error() + if len(message) > 256 { + message = message[:256] + } + payload, encodeErr := protocol.EncodeStableError(protocol.StableError{Version: "1", Code: code, Message: message, Retryable: retryable}) + if encodeErr != nil { + return encodeErr + } + return writeWire(writer, payload, defaultHelloLimit) +} + +func stableAdmissionCode(err error) string { + if errors.Is(err, ErrGatewayDraining) { + return "gateway_draining" + } + if errors.Is(err, ErrAuthorityExpired) { + return "expired_grant" + } + return "admission_rejected" +} + +func stableProviderCode(err error) string { + switch { + case errors.Is(err, ErrProviderIdentity): + return "provider_identity_rejected" + case errors.Is(err, ErrProviderMalformed): + return "provider_malformed" + case errors.Is(err, ErrProviderTimeout), errors.Is(err, context.DeadlineExceeded): + return "provider_timeout" + default: + return "provider_unavailable" + } +} + +func writeWire(writer io.Writer, payload []byte, max int) error { + if len(payload) > max { + return ErrFrameSize + } + 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 readWire(reader io.Reader, max 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(max) { + return nil, ErrFrameSize + } + payload := make([]byte, int(length)) + if _, err := io.ReadFull(reader, payload); err != nil { + return nil, err + } + return payload, nil +} + +type Client struct { + connection *quic.Conn + control *quic.Stream + Authority protocol.SessionAuthority +} + +func Dial(ctx context.Context, address string, tlsConfig *tls.Config, request protocol.TunnelAdmissionRequest) (*Client, error) { + if tlsConfig == nil || tlsConfig.MinVersion < tls.VersionTLS13 || tlsConfig.RootCAs == nil || len(tlsConfig.RootCAs.Subjects()) == 0 || len(tlsConfig.Certificates) == 0 { + return nil, ErrGatewayTLS + } + tlsConfig = tlsConfig.Clone() + if len(tlsConfig.NextProtos) == 0 { + tlsConfig.NextProtos = []string{"versevdi-gateway-v1"} + } + quicConfig := &quic.Config{EnableDatagrams: true, MaxIdleTimeout: 30 * time.Second} + connection, err := quic.DialAddr(ctx, address, tlsConfig, quicConfig) + if err != nil { + return nil, err + } + stream, err := connection.OpenStreamSync(ctx) + if err != nil { + _ = connection.CloseWithError(applicationError, "stream unavailable") + return nil, err + } + payload, err := protocol.EncodeTunnelAdmissionRequest(request) + if err != nil { + _ = connection.CloseWithError(applicationError, "invalid hello") + return nil, err + } + if err := writeWire(stream, payload, defaultHelloLimit); err != nil { + _ = connection.CloseWithError(applicationError, "invalid hello") + return nil, err + } + response, err := readWire(stream, defaultHelloLimit) + if err != nil { + _ = connection.CloseWithError(applicationError, "no authority") + return nil, err + } + authority, authorityErr := protocol.DecodeSessionAuthority(response) + if authorityErr != nil { + stable, stableErr := protocol.DecodeStableError(response) + if stableErr == nil { + return nil, fmt.Errorf("%s: %s", stable.Code, stable.Message) + } + return nil, authorityErr + } + return &Client{connection: connection, control: stream, Authority: authority}, nil +} + +func (c *Client) SendInput(event InputEvent) error { + payload, err := EncodeInputEvent(event) + if err != nil { + return err + } + frame := protocol.ChannelFrame{Version: "1", FlowID: "input", Sequence: int64(event.Sequence), Flags: 0, FragmentIndex: 0, FragmentCount: 1, TimestampMs: time.Now().UnixMilli(), Payload: base64.StdEncoding.EncodeToString(payload)} + encoded, err := protocol.EncodeChannelFrame(frame) + if err != nil { + return err + } + return writeWire(c.control, encoded, defaultControlLimit) +} + +func (c *Client) SendControl(payload []byte) error { + frame := protocol.ChannelFrame{Version: "1", FlowID: "control", Sequence: 0, Flags: 0, FragmentIndex: 0, FragmentCount: 1, TimestampMs: time.Now().UnixMilli(), Payload: base64.StdEncoding.EncodeToString(payload)} + encoded, err := protocol.EncodeChannelFrame(frame) + if err != nil { + return err + } + return writeWire(c.control, encoded, defaultControlLimit) +} + +func (c *Client) ReceiveFrame(ctx context.Context) (Frame, error) { + data, err := c.connection.ReceiveDatagram(ctx) + if err != nil { + return Frame{}, err + } + return DecodeFrame(data) +} + +func (c *Client) Close() error { + return c.connection.CloseWithError(applicationError, "client closed") +} diff --git a/go.mod b/go.mod index a47a551..e7a6a3e 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,14 @@ 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.1 + github.com/quic-go/quic-go v0.61.0 +) + +require ( + golang.org/x/crypto v0.54.0 // indirect + golang.org/x/net v0.56.0 // indirect + golang.org/x/sys v0.47.0 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..219f534 --- /dev/null +++ b/go.sum @@ -0,0 +1,22 @@ +git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.1 h1:bZd8Vs3tEjkLK25d0sjjZhIuF3S3v2xBhlZ5n4B9rVc= +git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.1/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= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/quic-go/go-ossfuzz-seeds v0.1.0 h1:APacT+iIaNF6fd8AGEiN3bT/Jtkd2jz4v4TzM7MFjy0= +github.com/quic-go/go-ossfuzz-seeds v0.1.0/go.mod h1:3IOHRbJIc+L6YKMwfDtJAM9Vj9k0YY4muhuyUYk5tbk= +github.com/quic-go/quic-go v0.61.0 h1:ui88A53s8MSVYLC56en0KQ17HARk+9986Dn0SBfKNvA= +github.com/quic-go/quic-go v0.61.0/go.mod h1:9So2anK4Tp22URSQq00k+Vo2PNkle96ycDPDHL4s9vs= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +go.uber.org/mock v0.5.2 h1:LbtPTcP8A5k9WPXj54PPPbjcI4Y6lhyOZXn+VS7wNko= +go.uber.org/mock v0.5.2/go.mod h1:wLlUxC2vVTPTaE3UD51E0BGOAElKrILxhVSDYQLld5o= +golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw= +golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk= +golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o= +golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/openspec/changes/apollo-transport-gateway/design.md b/openspec/changes/apollo-transport-gateway/design.md new file mode 100644 index 0000000..8d6be7b --- /dev/null +++ b/openspec/changes/apollo-transport-gateway/design.md @@ -0,0 +1,24 @@ +## Decisions + +- Use quic-go v0.61.0 with TLS 1.3, DATAGRAM enabled, bounded stream windows, bounded + datagram sizes, and no migration fallback in the application contract. +- Authenticate a client hello over a reliable stream, consume authority exactly once through + an injected admission client, then open lifecycle/control/input streams and media/audio + datagrams. +- Keep management/readiness/lifecycle, channel translation, and encoded relay separate. +- Use a deterministic fake Apollo provider behind the same adapter interface as the future + network client. Fixtures are non-live evidence and carry no host/credential material. +- On authority loss, close admission, release every pressed input, stop queues, and report + cleanup pending if provider termination is not acknowledged. + +## Bounds + +JSON hello/control is limited to 64 KiB, datagrams to 65,536 bytes with a configurable +path-MTU payload cap, fragments to 16, queues to fixed capacities, clipboard text to 65,536 +bytes and rate-limited, and each session owns only bounded goroutines/timers. + +## Failure behavior + +TLS/authentication, identity/protection, version, audience, grant, capability, parser, +provider, and cleanup failures close the relevant session with stable codes. Media payloads +are copied only for framing and are asserted byte-identical in tests. diff --git a/openspec/changes/apollo-transport-gateway/proposal.md b/openspec/changes/apollo-transport-gateway/proposal.md new file mode 100644 index 0000000..e0250bf --- /dev/null +++ b/openspec/changes/apollo-transport-gateway/proposal.md @@ -0,0 +1,23 @@ +## Why + +The Data Plane is an empty gateway boundary. Phase 3C-G needs a pure-Go QUIC process that +admits only Connection Server authority, adapts one Apollo/GameStream provider profile, and +relays encoded bytes without exposing the provider or decoding media. + +## What Changes + +- Add bounded QUIC/TLS streams and DATAGRAM framing with authenticated admission. +- Add a native-Go Apollo profile, deterministic fixtures, a bounded fake provider, lifecycle + cleanup, input release, capability intersection, pacing, telemetry, and packaging. + +## Provenance + +Provider behavior is independently implemented from the exact Apollo pin +`adc5c5a0bd80831ce495434bb16aee2cd4175fb8` and the Planning Hub's recorded public +Moonlight/common-C protocol evidence. Only the used source paths are recorded; no source +tree or proprietary capture is copied. + +## Non-goals + +No cgo, native sidecar, decoder/encoder/transcoder/render path, direct provider route, +database credential, provider plugin framework, or live Apollo qualification. diff --git a/openspec/changes/apollo-transport-gateway/specs/apollo-transport-gateway/spec.md b/openspec/changes/apollo-transport-gateway/specs/apollo-transport-gateway/spec.md new file mode 100644 index 0000000..f4ce72b --- /dev/null +++ b/openspec/changes/apollo-transport-gateway/specs/apollo-transport-gateway/spec.md @@ -0,0 +1,46 @@ +## ADDED Requirements + +### Requirement: Authenticated bounded gateway transport +The gateway SHALL require TLS 1.3 client authentication and a valid versioned grant hello +before allocating provider state. Reliable lifecycle/control/critical-input messages SHALL +use streams; encoded media/audio and approved sequenced input SHALL use bounded DATAGRAMs. + +#### Scenario: Grant replay or wrong audience +- **WHEN** a client presents a consumed, expired, revoked, or audience-mismatched grant +- **THEN** the gateway rejects before provider allocation and emits no provider route or + credential to the client. + +### Requirement: No-transcode encoded relay +The gateway SHALL relay provider encoded payload bytes through a codec-neutral Verse envelope +without decode, encode, transcode, render, or codec conversion. + +#### Scenario: Payload relay +- **WHEN** the fake Apollo provider emits an encoded video or audio payload +- **THEN** the corresponding Verse payload is byte-identical except for the approved transport + framing and the gateway records no decoder/encoder operation. + +### Requirement: Provider identity and protection +The Apollo profile SHALL reject changed, malformed, expired, or not-yet-valid pinned identity +and SHALL never silently retry with weaker protection. + +#### Scenario: Identity change +- **WHEN** the provider identity differs from the enrolled fingerprint or unique ID +- **THEN** launch fails closed and the session remains unavailable for new media. + +### Requirement: Bounded lifecycle and input safety +Authority loss, tunnel close, drain, provider disconnect, and explicit termination SHALL be +distinct states; every pressed key/button/controller SHALL be released before session cleanup. + +#### Scenario: Authority expiry during input +- **WHEN** authority expires while input is pressed +- **THEN** new input is rejected, release-all is sent to the provider, queues stop, and + cleanup is reported as pending until termination is acknowledged. + +### Requirement: Deterministic fake-provider qualification +The complete management, launch/readiness, channel, feedback/input, termination, and cleanup +sequence SHALL pass against a bounded fake provider plus malformed and timeout fixtures. + +#### Scenario: Fake provider timeout +- **WHEN** readiness or termination times out +- **THEN** the adapter returns a bounded stable error and marks cleanup pending without + spawning unbounded retries or goroutines. diff --git a/openspec/changes/apollo-transport-gateway/tasks.md b/openspec/changes/apollo-transport-gateway/tasks.md new file mode 100644 index 0000000..8a92d4d --- /dev/null +++ b/openspec/changes/apollo-transport-gateway/tasks.md @@ -0,0 +1,5 @@ +- [x] Add strict QUIC/TLS transport, reliable stream framing, and bounded datagram parser. +- [x] Add capability intersection, queues/pacing, input release, telemetry, and lifecycle. +- [x] Add deterministic Apollo fixtures, provenance, bounded fake provider, and adapter. +- [x] Add process-level mTLS/QUIC integration, fuzz/race/resource/impairment/scheduler tests. +- [ ] Add gateway command/package smoke checks and archive after candidate evidence matches. diff --git a/packaging/verse-gateway.Dockerfile b/packaging/verse-gateway.Dockerfile new file mode 100644 index 0000000..77f0cb7 --- /dev/null +++ b/packaging/verse-gateway.Dockerfile @@ -0,0 +1,7 @@ +FROM scratch + +COPY verse-gateway /usr/local/bin/verse-gateway + +USER 65532:65532 +EXPOSE 443/udp +ENTRYPOINT ["/usr/local/bin/verse-gateway"]