package grpcapi import ( "bytes" "context" "fmt" "galaxy/gateway/proto/galaxy/gateway/v1" "buf.build/go/protovalidate" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) const supportedProtocolVersion = "v1" // parsedEnvelope captures the authenticated transport fields extracted from a // request envelope after validation succeeds. Later wrappers may enrich this // structure without changing the raw gRPC request types. type parsedEnvelope struct { ProtocolVersion string DeviceSessionID string MessageType string TimestampMS int64 RequestID string TraceID string PayloadBytes []byte PayloadHash []byte Signature []byte } // parsedEnvelopeFromContext returns the parsed envelope previously attached to // ctx by the envelope-validating gRPC service wrapper. func parsedEnvelopeFromContext(ctx context.Context) (parsedEnvelope, bool) { if ctx == nil { return parsedEnvelope{}, false } envelope, ok := ctx.Value(parsedEnvelopeContextKey{}).(parsedEnvelope) if !ok { return parsedEnvelope{}, false } return envelope, true } // envelopeValidatingService applies envelope parsing and the protocol gate // before delegating to the configured service implementation. type envelopeValidatingService struct { gatewayv1.UnimplementedEdgeGatewayServer delegate gatewayv1.EdgeGatewayServer } // ExecuteCommand validates req and only then forwards it to the configured // delegate with the parsed envelope attached to ctx. func (s envelopeValidatingService) ExecuteCommand(ctx context.Context, req *gatewayv1.ExecuteCommandRequest) (*gatewayv1.ExecuteCommandResponse, error) { envelope, err := parseExecuteCommandRequest(req) if err != nil { return nil, err } return s.delegate.ExecuteCommand(context.WithValue(ctx, parsedEnvelopeContextKey{}, envelope), req) } // SubscribeEvents validates req and only then forwards it to the configured // delegate with the parsed envelope attached to the stream context. func (s envelopeValidatingService) SubscribeEvents(req *gatewayv1.SubscribeEventsRequest, stream grpc.ServerStreamingServer[gatewayv1.GatewayEvent]) error { envelope, err := parseSubscribeEventsRequest(req) if err != nil { return err } return s.delegate.SubscribeEvents(req, envelopeContextStream{ ServerStreamingServer: stream, ctx: context.WithValue(stream.Context(), parsedEnvelopeContextKey{}, envelope), }) } // parseExecuteCommandRequest validates req according to the request-envelope // rules and returns a cloned parsed envelope suitable for later auth steps. func parseExecuteCommandRequest(req *gatewayv1.ExecuteCommandRequest) (parsedEnvelope, error) { if req == nil { return parsedEnvelope{}, newMalformedEnvelopeError("request envelope must not be nil") } if err := protovalidate.Validate(req); err != nil { return parsedEnvelope{}, canonicalExecuteCommandValidationError(req) } if req.GetProtocolVersion() != supportedProtocolVersion { return parsedEnvelope{}, newUnsupportedProtocolVersionError(req.GetProtocolVersion()) } return parsedEnvelope{ ProtocolVersion: req.GetProtocolVersion(), DeviceSessionID: req.GetDeviceSessionId(), MessageType: req.GetMessageType(), TimestampMS: req.GetTimestampMs(), RequestID: req.GetRequestId(), TraceID: req.GetTraceId(), PayloadBytes: bytes.Clone(req.GetPayloadBytes()), PayloadHash: bytes.Clone(req.GetPayloadHash()), Signature: bytes.Clone(req.GetSignature()), }, nil } // parseSubscribeEventsRequest validates req according to the request-envelope // rules and returns a cloned parsed envelope suitable for later auth steps. func parseSubscribeEventsRequest(req *gatewayv1.SubscribeEventsRequest) (parsedEnvelope, error) { if req == nil { return parsedEnvelope{}, newMalformedEnvelopeError("request envelope must not be nil") } if err := protovalidate.Validate(req); err != nil { return parsedEnvelope{}, canonicalSubscribeEventsValidationError(req) } if req.GetProtocolVersion() != supportedProtocolVersion { return parsedEnvelope{}, newUnsupportedProtocolVersionError(req.GetProtocolVersion()) } return parsedEnvelope{ ProtocolVersion: req.GetProtocolVersion(), DeviceSessionID: req.GetDeviceSessionId(), MessageType: req.GetMessageType(), TimestampMS: req.GetTimestampMs(), RequestID: req.GetRequestId(), TraceID: req.GetTraceId(), PayloadBytes: bytes.Clone(req.GetPayloadBytes()), PayloadHash: bytes.Clone(req.GetPayloadHash()), Signature: bytes.Clone(req.GetSignature()), }, nil } // newEnvelopeValidatingService wraps delegate with the envelope-validation // gate. func newEnvelopeValidatingService(delegate gatewayv1.EdgeGatewayServer) gatewayv1.EdgeGatewayServer { return envelopeValidatingService{delegate: delegate} } // canonicalExecuteCommandValidationError maps any ExecuteCommand validation // failure into the stable canonical error chosen by field order. func canonicalExecuteCommandValidationError(req *gatewayv1.ExecuteCommandRequest) error { switch { case req.GetProtocolVersion() == "": return newMalformedEnvelopeError("protocol_version must not be empty") case req.GetDeviceSessionId() == "": return newMalformedEnvelopeError("device_session_id must not be empty") case req.GetMessageType() == "": return newMalformedEnvelopeError("message_type must not be empty") case req.GetTimestampMs() <= 0: return newMalformedEnvelopeError("timestamp_ms must be greater than zero") case req.GetRequestId() == "": return newMalformedEnvelopeError("request_id must not be empty") case len(req.GetPayloadBytes()) == 0: return newMalformedEnvelopeError("payload_bytes must not be empty") case len(req.GetPayloadHash()) == 0: return newMalformedEnvelopeError("payload_hash must not be empty") case len(req.GetSignature()) == 0: return newMalformedEnvelopeError("signature must not be empty") default: return newMalformedEnvelopeError("request envelope is invalid") } } // canonicalSubscribeEventsValidationError maps any SubscribeEvents validation // failure into the stable canonical error chosen by field order. func canonicalSubscribeEventsValidationError(req *gatewayv1.SubscribeEventsRequest) error { switch { case req.GetProtocolVersion() == "": return newMalformedEnvelopeError("protocol_version must not be empty") case req.GetDeviceSessionId() == "": return newMalformedEnvelopeError("device_session_id must not be empty") case req.GetMessageType() == "": return newMalformedEnvelopeError("message_type must not be empty") case req.GetTimestampMs() <= 0: return newMalformedEnvelopeError("timestamp_ms must be greater than zero") case req.GetRequestId() == "": return newMalformedEnvelopeError("request_id must not be empty") case len(req.GetPayloadHash()) == 0: return newMalformedEnvelopeError("payload_hash must not be empty") case len(req.GetSignature()) == 0: return newMalformedEnvelopeError("signature must not be empty") default: return newMalformedEnvelopeError("request envelope is invalid") } } // newMalformedEnvelopeError returns the stable malformed-envelope reject used // before the gateway performs any auth or routing work. func newMalformedEnvelopeError(message string) error { return status.Error(codes.InvalidArgument, message) } // newUnsupportedProtocolVersionError returns the stable reject for a non-empty // but unsupported protocol_version literal. func newUnsupportedProtocolVersionError(version string) error { return status.Error(codes.FailedPrecondition, fmt.Sprintf("unsupported protocol_version %q", version)) } type parsedEnvelopeContextKey struct{} type envelopeContextStream struct { grpc.ServerStreamingServer[gatewayv1.GatewayEvent] ctx context.Context } func (s envelopeContextStream) Context() context.Context { if s.ctx == nil { return context.Background() } return s.ctx } var _ gatewayv1.EdgeGatewayServer = envelopeValidatingService{}