galaxy-game/gateway/internal/grpcapi/envelope.go

package grpcapi

import (
	"bytes"
	"context"
	"fmt"
	edgev1 "galaxy/gateway/proto/edge/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 {
	edgev1.UnimplementedGatewayServer

	delegate edgev1.GatewayServer
}

// 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 *edgev1.ExecuteCommandRequest) (*edgev1.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 *edgev1.SubscribeEventsRequest, stream grpc.ServerStreamingServer[edgev1.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 *edgev1.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 *edgev1.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 edgev1.GatewayServer) edgev1.GatewayServer {
	return envelopeValidatingService{delegate: delegate}
}

// canonicalExecuteCommandValidationError maps any ExecuteCommand validation
// failure into the stable canonical error chosen by field order.
func canonicalExecuteCommandValidationError(req *edgev1.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 *edgev1.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[edgev1.GatewayEvent]
	ctx context.Context
}

func (s envelopeContextStream) Context() context.Context {
	if s.ctx == nil {
		return context.Background()
	}

	return s.ctx
}

var _ edgev1.GatewayServer = envelopeValidatingService{}