Ilia Denisov
96 lines
3.1 KiB
Go
96 lines
3.1 KiB
Go
package grpcapi
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
"galaxy/gateway/internal/clock"
|
|
"galaxy/gateway/internal/replay"
|
|
gatewayv1 "galaxy/gateway/proto/galaxy/gateway/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 {
|
|
gatewayv1.UnimplementedEdgeGatewayServer
|
|
|
|
delegate gatewayv1.EdgeGatewayServer
|
|
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 *gatewayv1.ExecuteCommandRequest) (*gatewayv1.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 *gatewayv1.SubscribeEventsRequest, stream grpc.ServerStreamingServer[gatewayv1.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 gatewayv1.EdgeGatewayServer, clk clock.Clock, replayStore replay.Store, freshnessWindow time.Duration) gatewayv1.EdgeGatewayServer {
|
|
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 _ gatewayv1.EdgeGatewayServer = freshnessAndReplayService{}
|