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

package grpcapi

import (
	"context"
	"errors"
	"time"

	"galaxy/gateway/internal/clock"
	"galaxy/gateway/internal/replay"
	edgev1 "galaxy/gateway/proto/edge/v1"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

const minimumReplayReservationTTL = time.Millisecond

// freshnessAndReplayService applies freshness and anti-replay checks after
// client-signature verification and before later policy or routing steps run.
type freshnessAndReplayService struct {
	edgev1.UnimplementedGatewayServer

	delegate        edgev1.GatewayServer
	clock           clock.Clock
	replayStore     replay.Store
	freshnessWindow time.Duration
}

// ExecuteCommand verifies request freshness and replay protection before
// delegating to the configured service implementation.
func (s freshnessAndReplayService) ExecuteCommand(ctx context.Context, req *edgev1.ExecuteCommandRequest) (*edgev1.ExecuteCommandResponse, error) {
	if err := s.verifyFreshnessAndReplay(ctx); err != nil {
		return nil, err
	}

	return s.delegate.ExecuteCommand(ctx, req)
}

// SubscribeEvents verifies request freshness and replay protection before
// delegating to the configured service implementation.
func (s freshnessAndReplayService) SubscribeEvents(req *edgev1.SubscribeEventsRequest, stream grpc.ServerStreamingServer[edgev1.GatewayEvent]) error {
	if err := s.verifyFreshnessAndReplay(stream.Context()); err != nil {
		return err
	}

	return s.delegate.SubscribeEvents(req, stream)
}

// newFreshnessAndReplayService wraps delegate with the freshness and replay
// gate.
func newFreshnessAndReplayService(delegate edgev1.GatewayServer, clk clock.Clock, replayStore replay.Store, freshnessWindow time.Duration) edgev1.GatewayServer {
	return freshnessAndReplayService{
		delegate:        delegate,
		clock:           clk,
		replayStore:     replayStore,
		freshnessWindow: freshnessWindow,
	}
}

func (s freshnessAndReplayService) verifyFreshnessAndReplay(ctx context.Context) error {
	envelope, ok := parsedEnvelopeFromContext(ctx)
	if !ok {
		return status.Error(codes.Internal, "authenticated request context is incomplete")
	}

	now := s.clock.Now().UTC()
	requestTime := time.UnixMilli(envelope.TimestampMS).UTC()
	if requestTime.Before(now.Add(-s.freshnessWindow)) || requestTime.After(now.Add(s.freshnessWindow)) {
		return status.Error(codes.FailedPrecondition, "request timestamp is outside the freshness window")
	}

	ttl := requestTime.Add(s.freshnessWindow).Sub(now)
	if ttl < minimumReplayReservationTTL {
		ttl = minimumReplayReservationTTL
	}

	err := s.replayStore.Reserve(ctx, envelope.DeviceSessionID, envelope.RequestID, ttl)
	switch {
	case err == nil:
		return nil
	case errors.Is(err, replay.ErrDuplicate):
		return status.Error(codes.FailedPrecondition, "request replay detected")
	default:
		return status.Error(codes.Unavailable, "replay store is unavailable")
	}
}

type unavailableReplayStore struct{}

func (unavailableReplayStore) Reserve(context.Context, string, string, time.Duration) error {
	return errors.New("replay store is unavailable")
}

var _ edgev1.GatewayServer = freshnessAndReplayService{}