diff --git a/internal/agent/recording_client.go b/internal/agent/recording_client.go index 24b7b33..42219fc 100644 --- a/internal/agent/recording_client.go +++ b/internal/agent/recording_client.go @@ -5,9 +5,14 @@ import ( "context" "errors" "fmt" + "log" + "time" agentpb "git.ipao.vip/rogee/go-sip/gen/agent" + "git.ipao.vip/rogee/go-sip/internal/session" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" ) @@ -53,16 +58,19 @@ func (c RecordingClient) RequestUpload(ctx context.Context, asset *agentpb.Asset asset.GetExecutionId() != c.SourceEventID || asset.GetCallId() != c.SourceEventID || asset.GetAssetId() == "" || asset.GetSizeBytes() <= 0 || asset.GetChecksumSha256() == "" { return nil, errors.New("recording token request does not describe the approved call and original asset") } - response, err := c.Client.RequestRecordingUpload(ctx, &agentpb.RequestRecordingUploadRequest{ - Meta: meta, DispatcherId: c.DispatcherID, TenantId: c.TenantID, SourceEventId: c.SourceEventID, - UploadId: uploadID, Asset: proto.Clone(asset).(*agentpb.AssetDescriptor), + originalAsset := proto.Clone(asset).(*agentpb.AssetDescriptor) + response, err := reReportVerifiedSessionCutover(ctx, c, meta, "upload/"+uploadID, func(callCtx context.Context, current *agentpb.RequestMeta) (*agentpb.RequestRecordingUploadResponse, error) { + return c.Client.RequestRecordingUpload(callCtx, &agentpb.RequestRecordingUploadRequest{ + Meta: current, DispatcherId: c.DispatcherID, TenantId: c.TenantID, SourceEventId: c.SourceEventID, + UploadId: uploadID, Asset: proto.Clone(originalAsset).(*agentpb.AssetDescriptor), + }) }) if err != nil { return nil, fmt.Errorf("request original recording upload token: %w", err) } grant := response.GetGrant() if grant == nil || grant.GetUploadId() != uploadID || grant.GetBucket() == "" || grant.GetObjectKey() == "" || - grant.GetMaxBytes() != asset.GetSizeBytes() || grant.GetRequiredChecksumSha256() != asset.GetChecksumSha256() { + grant.GetMaxBytes() != originalAsset.GetSizeBytes() || grant.GetRequiredChecksumSha256() != originalAsset.GetChecksumSha256() { return nil, fmt.Errorf("%w: Dispatcher returned a different recording asset", ErrUploadGrantInvalid) } return grant, nil @@ -73,8 +81,10 @@ func (c RecordingClient) ReportEnded(ctx context.Context) error { if err != nil { return err } - response, err := c.Client.ReportCallEnded(ctx, &agentpb.ReportCallEndedRequest{ - Meta: meta, DispatcherId: c.DispatcherID, TenantId: c.TenantID, SourceEventId: c.SourceEventID, + response, err := reReportVerifiedSessionCutover(ctx, c, meta, "ended", func(callCtx context.Context, current *agentpb.RequestMeta) (*agentpb.ReportCallEndedResponse, error) { + return c.Client.ReportCallEnded(callCtx, &agentpb.ReportCallEndedRequest{ + Meta: current, DispatcherId: c.DispatcherID, TenantId: c.TenantID, SourceEventId: c.SourceEventID, + }) }) if err != nil { return fmt.Errorf("report confirmed call end: %w", err) @@ -93,14 +103,21 @@ func (c RecordingClient) ReportFinal(ctx context.Context, payload []byte, upload if len(payload) == 0 { return "", errors.New("final call result payload is required") } - request := &agentpb.ReportCallResultRequest{ - Meta: meta, DispatcherId: c.DispatcherID, TenantId: c.TenantID, SourceEventId: c.SourceEventID, - ResultPayloadJson: bytes.Clone(payload), - } + originalPayload := bytes.Clone(payload) + var originalUpload *agentpb.UploadObservation if upload != nil { - request.Upload = proto.Clone(upload).(*agentpb.UploadObservation) + originalUpload = proto.Clone(upload).(*agentpb.UploadObservation) } - response, err := c.Client.ReportCallResult(ctx, request) + response, err := reReportVerifiedSessionCutover(ctx, c, meta, "result", func(callCtx context.Context, current *agentpb.RequestMeta) (*agentpb.ReportCallResultResponse, error) { + request := &agentpb.ReportCallResultRequest{ + Meta: current, DispatcherId: c.DispatcherID, TenantId: c.TenantID, SourceEventId: c.SourceEventID, + ResultPayloadJson: bytes.Clone(originalPayload), + } + if originalUpload != nil { + request.Upload = proto.Clone(originalUpload).(*agentpb.UploadObservation) + } + return c.Client.ReportCallResult(callCtx, request) + }) if err != nil { return "", fmt.Errorf("persist unique call result: %w", err) } @@ -109,3 +126,44 @@ func (c RecordingClient) ReportFinal(ctx context.Context, payload []byte, upload } return "", errors.New("Dispatcher did not persist the unique call result") } + +func reReportVerifiedSessionCutover[T any](ctx context.Context, c RecordingClient, original *agentpb.RequestMeta, action string, send func(context.Context, *agentpb.RequestMeta) (T, error)) (T, error) { + response, err := send(ctx, original) + if !isVerifiedSessionTransition(err) { + return response, err + } + log.Printf("Agent fact fenced during session transition event=%s agent=%s generation=%d action=%s", c.SourceEventID, original.AgentId, original.SessionGeneration, action) + retryCtx, cancel := context.WithTimeout(ctx, 6*time.Second) + defer cancel() + attempts := 1 + for { + timer := time.NewTimer(100 * time.Millisecond) + select { + case <-retryCtx.Done(): + timer.Stop() + log.Printf("Agent fact re-report deadline event=%s agent=%s action=%s attempts=%d", c.SourceEventID, original.AgentId, action, attempts) + var zero T + return zero, fmt.Errorf("Agent fact re-report not confirmed: %w", errors.Join(err, retryCtx.Err())) + case <-timer.C: + } + current, sessionErr := c.requestMeta(retryCtx, action) + if sessionErr != nil { + var zero T + return zero, fmt.Errorf("obtain renewed Agent session for original fact: %w", sessionErr) + } + if current.AgentId != original.AgentId || current.CellId != original.CellId || current.BootId != original.BootId || current.DispatcherEpoch != original.DispatcherEpoch { + var zero T + return zero, fmt.Errorf("%w: Agent identity changed during session transition", ErrRecordingClientUnavailable) + } + response, err = send(retryCtx, current) + attempts++ + if !isVerifiedSessionTransition(err) { + return response, err + } + } +} + +func isVerifiedSessionTransition(err error) bool { + st, ok := status.FromError(err) + return ok && st.Code() == codes.Unauthenticated && st.Message() == session.GenerationTransitionMessage +} diff --git a/internal/agent/recording_client_transition_test.go b/internal/agent/recording_client_transition_test.go new file mode 100644 index 0000000..acb07b6 --- /dev/null +++ b/internal/agent/recording_client_transition_test.go @@ -0,0 +1,202 @@ +package agent + +import ( + "bytes" + "context" + "errors" + "testing" + "time" + + agentpb "git.ipao.vip/rogee/go-sip/gen/agent" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" +) + +const verifiedSessionCutover = "verified Agent session generation changed" + +type cutoverRecordingRPC struct { + agentpb.AgentControlServiceClient + failure error + lateFailure error + alwaysReject bool + sameGeneration bool + changeIdentity string + generation uint64 + agentID string + cellID string + boot string + epoch string + metas []*agentpb.RequestMeta + assets []*agentpb.AssetDescriptor + uploadIDs []string + payloads [][]byte +} + +func (s *cutoverRecordingRPC) receive(meta *agentpb.RequestMeta) error { + s.metas = append(s.metas, proto.Clone(meta).(*agentpb.RequestMeta)) + if len(s.metas) == 1 { + if !s.sameGeneration { + s.generation++ + } + switch s.changeIdentity { + case "agent": + s.agentID = "different-agent" + case "cell": + s.cellID = "different-cell" + case "boot": + s.boot = "different-boot" + case "epoch": + s.epoch = "different-epoch" + } + return s.failure + } + if s.lateFailure != nil { + return s.lateFailure + } + if s.alwaysReject { + return s.failure + } + return nil +} + +func (s *cutoverRecordingRPC) RequestRecordingUpload(_ context.Context, req *agentpb.RequestRecordingUploadRequest, _ ...grpc.CallOption) (*agentpb.RequestRecordingUploadResponse, error) { + s.uploadIDs = append(s.uploadIDs, req.UploadId) + s.assets = append(s.assets, proto.Clone(req.Asset).(*agentpb.AssetDescriptor)) + if err := s.receive(req.Meta); err != nil { + return nil, err + } + return &agentpb.RequestRecordingUploadResponse{Grant: &agentpb.UploadGrant{ + UploadId: req.UploadId, Bucket: "mock-bucket", ObjectKey: "mock-key", + MaxBytes: req.Asset.SizeBytes, RequiredChecksumSha256: req.Asset.ChecksumSha256, + }}, nil +} + +func (s *cutoverRecordingRPC) ReportCallEnded(_ context.Context, req *agentpb.ReportCallEndedRequest, _ ...grpc.CallOption) (*agentpb.ReportCallEndedResponse, error) { + if err := s.receive(req.Meta); err != nil { + return nil, err + } + return &agentpb.ReportCallEndedResponse{Receipt: &agentpb.OperationReceipt{FactId: req.SourceEventId, Result: agentpb.ResultCode_RESULT_CODE_APPLIED}}, nil +} + +func (s *cutoverRecordingRPC) ReportCallResult(_ context.Context, req *agentpb.ReportCallResultRequest, _ ...grpc.CallOption) (*agentpb.ReportCallResultResponse, error) { + s.payloads = append(s.payloads, bytes.Clone(req.ResultPayloadJson)) + if err := s.receive(req.Meta); err != nil { + return nil, err + } + return &agentpb.ReportCallResultResponse{Receipt: &agentpb.OperationReceipt{FactId: "result-fact", Result: agentpb.ResultCode_RESULT_CODE_ACCEPTED}}, nil +} + +func newCutoverRecordingClient(s *cutoverRecordingRPC) RecordingClient { + s.generation = 1 + s.agentID = "agent-1" + s.cellID = "cell-1" + s.boot = "same-boot" + s.epoch = "epoch-1" + return RecordingClient{ + Client: s, DispatcherID: "dispatcher-1", TenantID: 1, SourceEventID: "call-1", + Session: func(context.Context) (*agentpb.RequestMeta, error) { + return &agentpb.RequestMeta{AgentId: s.agentID, CellId: s.cellID, BootId: s.boot, DispatcherEpoch: s.epoch, SessionGeneration: s.generation}, nil + }, + } +} + +func TestRecordingClientReReportsOnlyOriginalFactAfterVerifiedSessionCutover(t *testing.T) { + for _, tc := range []struct { + name string + operation string + report func(RecordingClient) error + }{ + {"upload grant", "call-1/upload/up-1", func(c RecordingClient) error { + _, err := c.RequestUpload(context.Background(), &agentpb.AssetDescriptor{ + Kind: agentpb.AssetKind_ASSET_KIND_RECORDING, ExecutionId: "call-1", CallId: "call-1", AssetId: "recording-1", SizeBytes: 4, ChecksumSha256: "digest", + }, "up-1") + return err + }}, + {"confirmed end", "call-1/ended", func(c RecordingClient) error { return c.ReportEnded(context.Background()) }}, + {"final result", "call-1/result", func(c RecordingClient) error { + _, err := c.ReportFinal(context.Background(), []byte(`{"call_id":"call-1"}`), nil) + return err + }}, + } { + t.Run(tc.name, func(t *testing.T) { + transport := &cutoverRecordingRPC{failure: status.Error(codes.Unauthenticated, verifiedSessionCutover)} + if err := tc.report(newCutoverRecordingClient(transport)); err != nil { + t.Fatalf("original fact not delivered after session cutover: %v", err) + } + if len(transport.metas) != 2 { + t.Fatalf("want exactly one bounded re-report, got %d requests", len(transport.metas)) + } + old, current := transport.metas[0], transport.metas[1] + if old.SessionGeneration != 1 || current.SessionGeneration != 2 || + old.AgentId != current.AgentId || old.CellId != current.CellId || old.BootId != current.BootId || old.DispatcherEpoch != current.DispatcherEpoch || + old.OperationId != tc.operation || current.OperationId != tc.operation || old.IdempotencyKey != tc.operation || current.IdempotencyKey != tc.operation { + t.Fatalf("session cutover changed fact identity or accepted an unbound Agent: first=%v second=%v", old, current) + } + if tc.name == "upload grant" && (len(transport.assets) != 2 || !proto.Equal(transport.assets[0], transport.assets[1]) || transport.uploadIDs[0] != "up-1" || transport.uploadIDs[1] != "up-1") { + t.Fatal("upload grant request changed original asset/upload identity") + } + if tc.name == "final result" && (len(transport.payloads) != 2 || !bytes.Equal(transport.payloads[0], transport.payloads[1])) { + t.Fatal("re-report changed original final result payload") + } + }) + } +} + +func TestRecordingClientDoesNotRetryUnverifiedOrUnknownOutcomes(t *testing.T) { + for _, tc := range []struct { + name string + failure error + identity string + }{ + {"permission denied", status.Error(codes.PermissionDenied, "inactive Agent"), ""}, + {"timeout with unknown result", status.Error(codes.DeadlineExceeded, "result unknown"), ""}, + {"unavailable with unknown result", status.Error(codes.Unavailable, "result unknown"), ""}, + {"unverified unauthenticated", status.Error(codes.Unauthenticated, "mTLS peer missing"), ""}, + {"different Agent after cutover", status.Error(codes.Unauthenticated, verifiedSessionCutover), "agent"}, + {"different Cell after cutover", status.Error(codes.Unauthenticated, verifiedSessionCutover), "cell"}, + {"different boot after cutover", status.Error(codes.Unauthenticated, verifiedSessionCutover), "boot"}, + {"different epoch after cutover", status.Error(codes.Unauthenticated, verifiedSessionCutover), "epoch"}, + } { + t.Run(tc.name, func(t *testing.T) { + transport := &cutoverRecordingRPC{failure: tc.failure, changeIdentity: tc.identity} + err := newCutoverRecordingClient(transport).ReportEnded(context.Background()) + if err == nil || len(transport.metas) != 1 { + t.Fatalf("unauthorized or uncertain call end was re-reported: requests=%d error=%v", len(transport.metas), err) + } + }) + } +} + +func TestRecordingClientStopsWhenSecondOutcomeIsUnknown(t *testing.T) { + transport := &cutoverRecordingRPC{ + failure: status.Error(codes.Unauthenticated, verifiedSessionCutover), + lateFailure: status.Error(codes.DeadlineExceeded, "result unknown after re-report"), + } + err := newCutoverRecordingClient(transport).ReportEnded(context.Background()) + if status.Code(err) != codes.DeadlineExceeded || len(transport.metas) != 2 { + t.Fatalf("unknown second outcome caused an unsafe retry: requests=%d error=%v", len(transport.metas), err) + } +} + +func TestRecordingClientAllowsOriginalGenerationUntilDispatcherCatchesUp(t *testing.T) { + transport := &cutoverRecordingRPC{failure: status.Error(codes.Unauthenticated, verifiedSessionCutover), sameGeneration: true} + if err := newCutoverRecordingClient(transport).ReportEnded(context.Background()); err != nil { + t.Fatalf("original fact was not delivered after Dispatcher session caught up: %v", err) + } + if len(transport.metas) != 2 || transport.metas[0].SessionGeneration != transport.metas[1].SessionGeneration { + t.Fatalf("re-report changed an otherwise still-active Agent generation: %d requests", len(transport.metas)) + } +} + +func TestRecordingClientSessionCutoverReReportIsBounded(t *testing.T) { + transport := &cutoverRecordingRPC{failure: status.Error(codes.Unauthenticated, verifiedSessionCutover), alwaysReject: true} + ctx, cancel := context.WithTimeout(context.Background(), 280*time.Millisecond) + defer cancel() + err := newCutoverRecordingClient(transport).ReportEnded(ctx) + if !errors.Is(err, context.DeadlineExceeded) || len(transport.metas) < 2 || len(transport.metas) > 4 { + t.Fatalf("re-report did not stop at caller deadline: requests=%d error=%v", len(transport.metas), err) + } +} diff --git a/internal/dispatcher/agent.go b/internal/dispatcher/agent.go index 554c424..e2818e4 100644 --- a/internal/dispatcher/agent.go +++ b/internal/dispatcher/agent.go @@ -8,6 +8,7 @@ import ( "time" agentpb "git.ipao.vip/rogee/go-sip/gen/agent" + agentsession "git.ipao.vip/rogee/go-sip/internal/session" "git.ipao.vip/rogee/go-sip/internal/store" "git.ipao.vip/rogee/go-sip/internal/tenant" "github.com/google/uuid" @@ -206,9 +207,17 @@ func (c *AgentCoordinator) AuthorizeInboundMeta(meta *agentpb.RequestMeta) error c.mu.Unlock() if !active || !registered || client == nil || session.ExpiresAtUnixMs <= c.now().UnixMilli() || meta.CellId != session.CellID || meta.BootId != session.BootID || - meta.DispatcherEpoch != session.DispatcherEpoch || meta.SessionGeneration != session.SessionGeneration { + meta.DispatcherEpoch != session.DispatcherEpoch { return fmt.Errorf("Agent report does not match a current activated session: %w", store.ErrCommandConflict) } + if meta.SessionGeneration != session.SessionGeneration { + if meta.SessionGeneration > 0 && + ((meta.SessionGeneration < session.SessionGeneration && session.SessionGeneration-meta.SessionGeneration == 1) || + (meta.SessionGeneration > session.SessionGeneration && meta.SessionGeneration-session.SessionGeneration == 1)) { + return fmt.Errorf("Agent fact was fenced during session transition: %w: %w", store.ErrCommandConflict, agentsession.ErrGenerationTransition) + } + return fmt.Errorf("Agent report has an unapproved session generation: %w", store.ErrCommandConflict) + } return nil } diff --git a/internal/dispatcher/agent_inbound_session_test.go b/internal/dispatcher/agent_inbound_session_test.go index 6f0f9a3..7bc71c7 100644 --- a/internal/dispatcher/agent_inbound_session_test.go +++ b/internal/dispatcher/agent_inbound_session_test.go @@ -6,6 +6,7 @@ import ( "time" agentpb "git.ipao.vip/rogee/go-sip/gen/agent" + "git.ipao.vip/rogee/go-sip/internal/session" "git.ipao.vip/rogee/go-sip/internal/store" "google.golang.org/protobuf/proto" ) @@ -32,20 +33,26 @@ func TestInboundAgentFactRequiresCurrentUnexpiredActivatedSession(t *testing.T) t.Fatalf("active Agent was refused: %v", err) } for _, tc := range []struct { - name string - change func(*agentpb.RequestMeta) + name string + change func(*agentpb.RequestMeta) + transition bool }{ - {"old boot", func(m *agentpb.RequestMeta) { m.BootId = "boot-old" }}, - {"old generation", func(m *agentpb.RequestMeta) { m.SessionGeneration = 2 }}, - {"other cell", func(m *agentpb.RequestMeta) { m.CellId = "cell-other" }}, - {"other epoch", func(m *agentpb.RequestMeta) { m.DispatcherEpoch = "epoch-other" }}, - {"unregistered Agent", func(m *agentpb.RequestMeta) { m.AgentId = "agent-other" }}, + {"old boot", func(m *agentpb.RequestMeta) { m.BootId = "boot-old" }, false}, + {"old generation", func(m *agentpb.RequestMeta) { m.SessionGeneration = 2 }, true}, + {"next generation before activation response", func(m *agentpb.RequestMeta) { m.SessionGeneration = 4 }, true}, + {"distant generation", func(m *agentpb.RequestMeta) { m.SessionGeneration = 7 }, false}, + {"missing generation", func(m *agentpb.RequestMeta) { m.SessionGeneration = 0 }, false}, + {"other cell", func(m *agentpb.RequestMeta) { m.CellId = "cell-other" }, false}, + {"other epoch", func(m *agentpb.RequestMeta) { m.DispatcherEpoch = "epoch-other" }, false}, + {"unregistered Agent", func(m *agentpb.RequestMeta) { m.AgentId = "agent-other" }, false}, + {"old boot and generation", func(m *agentpb.RequestMeta) { m.BootId = "boot-old"; m.SessionGeneration = 2 }, false}, } { t.Run(tc.name, func(t *testing.T) { changed := proto.Clone(meta).(*agentpb.RequestMeta) tc.change(changed) - if err := coordinator.AuthorizeInboundMeta(changed); !errors.Is(err, store.ErrCommandConflict) { - t.Fatalf("unbound Agent report passed: %v", err) + err := coordinator.AuthorizeInboundMeta(changed) + if !errors.Is(err, store.ErrCommandConflict) || errors.Is(err, session.ErrGenerationTransition) != tc.transition { + t.Fatalf("Agent report refusal classification: got %v, transition=%t", err, tc.transition) } }) } @@ -53,4 +60,9 @@ func TestInboundAgentFactRequiresCurrentUnexpiredActivatedSession(t *testing.T) if err := coordinator.AuthorizeInboundMeta(meta); !errors.Is(err, store.ErrCommandConflict) { t.Fatalf("expired Agent report passed: %v", err) } + stale := proto.Clone(meta).(*agentpb.RequestMeta) + stale.SessionGeneration = 2 + if err := coordinator.AuthorizeInboundMeta(stale); errors.Is(err, session.ErrGenerationTransition) { + t.Fatalf("expired Agent session misclassified as a transient cutover: %v", err) + } } diff --git a/internal/rpc/recording_server.go b/internal/rpc/recording_server.go index bdc0a52..7b31b6f 100644 --- a/internal/rpc/recording_server.go +++ b/internal/rpc/recording_server.go @@ -12,6 +12,7 @@ import ( agentpb "git.ipao.vip/rogee/go-sip/gen/agent" "git.ipao.vip/rogee/go-sip/internal/oss" + "git.ipao.vip/rogee/go-sip/internal/session" "git.ipao.vip/rogee/go-sip/internal/store" "google.golang.org/grpc/codes" @@ -52,6 +53,10 @@ func (s *RecordingServer) authorize(ctx context.Context, meta *agentpb.RequestMe return err } if err := s.AuthorizeSession(meta); err != nil { + if errors.Is(err, session.ErrGenerationTransition) { + log.Printf("recording RPC session transition dispatcher=%s event=%s agent=%s generation=%d", dispatcherID, sourceEventID, meta.GetAgentId(), meta.GetSessionGeneration()) + return status.Error(codes.Unauthenticated, session.GenerationTransitionMessage) + } log.Printf("recording RPC denied dispatcher=%s event=%s reason=inactive_agent_session", dispatcherID, sourceEventID) return status.Error(codes.PermissionDenied, "Agent session is not active for this Dispatcher") } diff --git a/internal/rpc/recording_server_flow_test.go b/internal/rpc/recording_server_flow_test.go index b6bc5d3..a74467f 100644 --- a/internal/rpc/recording_server_flow_test.go +++ b/internal/rpc/recording_server_flow_test.go @@ -15,6 +15,7 @@ import ( agentpb "git.ipao.vip/rogee/go-sip/gen/agent" "git.ipao.vip/rogee/go-sip/internal/configread" "git.ipao.vip/rogee/go-sip/internal/oss" + "git.ipao.vip/rogee/go-sip/internal/session" "git.ipao.vip/rogee/go-sip/internal/store" "google.golang.org/grpc/codes" @@ -130,6 +131,47 @@ func recordingResultPayload(t *testing.T, task configread.Snapshot, grant *agent return raw } +func TestRecordingServerClassifiesOnlyVerifiedSessionCutover(t *testing.T) { + server, database, verified, request, _ := recordingRPCFixture(t) + server.AuthorizeSession = func(*agentpb.RequestMeta) error { return session.ErrGenerationTransition } + for _, tc := range []struct { + name string + call func(context.Context) error + }{ + {"grant", func(ctx context.Context) error { _, err := server.RequestRecordingUpload(ctx, request); return err }}, + {"call end", func(ctx context.Context) error { + _, err := server.ReportCallEnded(ctx, &agentpb.ReportCallEndedRequest{Meta: request.Meta, DispatcherId: request.DispatcherId, TenantId: request.TenantId, SourceEventId: request.SourceEventId}) + return err + }}, + {"final result", func(ctx context.Context) error { + _, err := server.ReportCallResult(ctx, &agentpb.ReportCallResultRequest{Meta: request.Meta, DispatcherId: request.DispatcherId, TenantId: request.TenantId, SourceEventId: request.SourceEventId}) + return err + }}, + } { + t.Run(tc.name, func(t *testing.T) { + err := tc.call(verified) + if status.Code(err) != codes.Unauthenticated || status.Convert(err).Message() != "verified Agent session generation changed" { + t.Fatalf("verified session transition must be a distinct definite refusal: %v", err) + } + if err := tc.call(context.Background()); status.Code(err) != codes.Unauthenticated || status.Convert(err).Message() == "verified Agent session generation changed" { + t.Fatalf("missing peer misclassified as verified session transition: %v", err) + } + }) + } + if _, err := database.LoadRecordingUpload(request.DispatcherId, request.SourceEventId); !errors.Is(err, store.ErrUploadNotFound) { + t.Fatalf("denied transition created an upload grant: %v", err) + } + server.TrustedFingerprints = map[string]struct{}{"different": {}} + if _, err := server.RequestRecordingUpload(verified, request); status.Code(err) != codes.PermissionDenied { + t.Fatalf("wrong mTLS fingerprint was treated as retriable: %v", err) + } + server.TrustedFingerprints = map[string]struct{}{CertificateFingerprint(&x509.Certificate{Raw: []byte("isolated-test-agent-peer")}): {}} + server.AuthorizeSession = func(*agentpb.RequestMeta) error { return errors.New("unapproved Agent identity") } + if _, err := server.RequestRecordingUpload(verified, request); status.Code(err) != codes.PermissionDenied { + t.Fatalf("unrelated authorization refusal was treated as retriable: %v", err) + } +} + func TestRecordingServerAuthenticatesOriginalGrantAndExplicitReissue(t *testing.T) { server, database, ctx, request, _ := recordingRPCFixture(t) if _, err := server.RequestRecordingUpload(context.Background(), request); status.Code(err) != codes.Unauthenticated { diff --git a/internal/rpc/recording_transport_test.go b/internal/rpc/recording_transport_test.go index 3ee5b8b..a66ee59 100644 --- a/internal/rpc/recording_transport_test.go +++ b/internal/rpc/recording_transport_test.go @@ -3,15 +3,19 @@ package rpc import ( "context" "net" + "sync/atomic" "testing" "time" agentpb "git.ipao.vip/rogee/go-sip/gen/agent" "git.ipao.vip/rogee/go-sip/internal/agent" + "git.ipao.vip/rogee/go-sip/internal/session" + "git.ipao.vip/rogee/go-sip/internal/store" "google.golang.org/grpc" "google.golang.org/grpc/credentials" "google.golang.org/grpc/test/bufconn" + "google.golang.org/protobuf/proto" ) func TestRecordingFactsCrossRealMutualTLSUnaryTransportWithoutExternalOSS(t *testing.T) { @@ -28,6 +32,22 @@ func TestRecordingFactsCrossRealMutualTLSUnaryTransportWithoutExternalOSS(t *tes t.Fatal(err) } server.TrustedFingerprints = map[string]struct{}{CertificateFingerprint(clientCert): {}} + var agentGeneration, activeGeneration atomic.Uint64 + var fencedReports atomic.Int32 + agentGeneration.Store(request.Meta.SessionGeneration) + activeGeneration.Store(request.Meta.SessionGeneration + 1) + server.AuthorizeSession = func(meta *agentpb.RequestMeta) error { + active := activeGeneration.Load() + if meta.SessionGeneration == active { + return nil + } + if meta.SessionGeneration > 0 && meta.SessionGeneration < active && active-meta.SessionGeneration == 1 { + fencedReports.Add(1) + agentGeneration.Store(active) + return session.ErrGenerationTransition + } + return store.ErrCommandConflict + } listener := bufconn.Listen(1 << 20) grpcServer := grpc.NewServer(grpc.Creds(credentials.NewTLS(serverTLS))) agentpb.RegisterAgentControlServiceServer(grpcServer, server) @@ -45,20 +65,29 @@ func TestRecordingFactsCrossRealMutualTLSUnaryTransportWithoutExternalOSS(t *tes defer cancel() client := agent.RecordingClient{ Client: agentpb.NewAgentControlServiceClient(conn), DispatcherID: request.DispatcherId, TenantID: request.TenantId, SourceEventID: request.SourceEventId, - Session: func(context.Context) (*agentpb.RequestMeta, error) { return request.Meta, nil }, + Session: func(context.Context) (*agentpb.RequestMeta, error) { + meta := proto.Clone(request.Meta).(*agentpb.RequestMeta) + meta.SessionGeneration = agentGeneration.Load() + return meta, nil + }, } grant, err := client.RequestUpload(ctx, request.Asset, request.UploadId) if err != nil || grant.GetBucket() != "mock-bucket" { t.Fatalf("mTLS Agent request did not bind Dispatcher-owned bucket: bucket=%q err=%v", grant.GetBucket(), err) } + activeGeneration.Add(1) if err := client.ReportEnded(ctx); err != nil { t.Fatalf("mTLS Agent end did not durably acknowledge original execution: %v", err) } observation := &agentpb.UploadObservation{UploadId: request.UploadId, RecordingId: request.Asset.AssetId, PutStatusCode: 200, SizeBytes: request.Asset.SizeBytes, ChecksumSha256: request.Asset.ChecksumSha256} + activeGeneration.Add(1) resultID, err := client.ReportFinal(ctx, recordingResultPayload(t, snapshot, grant), observation) if err != nil || resultID == "" { t.Fatalf("mTLS Agent final result not persisted: fact=%q err=%v", resultID, err) } + if fencedReports.Load() != 3 || agentGeneration.Load() != activeGeneration.Load() { + t.Fatalf("three mTLS recording facts did not cross the verified Agent session cutovers: fenced=%d current=%d", fencedReports.Load(), agentGeneration.Load()) + } if request.Meta.OperationId != "request-1" || request.Meta.IdempotencyKey != "request-1" { t.Fatal("concurrent Agent session metadata was mutated by the recording client") } diff --git a/internal/session/transition.go b/internal/session/transition.go new file mode 100644 index 0000000..e6cda75 --- /dev/null +++ b/internal/session/transition.go @@ -0,0 +1,12 @@ +// Package session shares the verified Agent cutover refusal across the +// Dispatcher, recording RPC server, and Agent without coupling their lifecycles. +package session + +import "errors" + +// GenerationTransitionMessage is the sole retryable recording RPC refusal. +const GenerationTransitionMessage = "verified Agent session generation changed" + +// ErrGenerationTransition marks an adjacent active-session mismatch; it does +// not authorize the rejected fact or bypass the latest-generation fence. +var ErrGenerationTransition = errors.New(GenerationTransitionMessage) diff --git a/proto/ERRORS.md b/proto/ERRORS.md index b934709..d1cf3f8 100644 --- a/proto/ERRORS.md +++ b/proto/ERRORS.md @@ -10,7 +10,7 @@ contract does not change SaaS/MQ commands, ownership, or application receipts. | gRPC status | Required caller behavior | | --- | --- | | `InvalidArgument` | Correct the request; do not replay it unchanged. | -| `Unauthenticated` | Re-establish the verified mTLS/active session before more work. | +| `Unauthenticated` | Re-establish the verified mTLS/active session before more work. Only the exact, verified session-generation cutover refusal described below permits a bounded re-report of the original fact. | | `PermissionDenied` | Stop; no self-reported identity or unapproved peer is accepted. | | `FailedPrecondition` | Keep admission closed until the missing state, approval, or loaded SIP revision is resolved. | | `Aborted` / `AlreadyExists` | Preserve the original execution identity; conflicting content must not originate another call. | @@ -49,6 +49,17 @@ require separate evidence. ## Recording and final result +- If a verified mTLS Agent peer reports `ReportCallEnded`, `ReportCallResult`, + or `RequestRecordingUpload` across an adjacent active-session generation + change for the same Agent, Cell, boot, and Dispatcher epoch, the Dispatcher + **rejects** the fact with `Unauthenticated: verified Agent session generation + changed`. The Agent may fetch its current session and re-report only that + original idempotent fact for at most six seconds (or the earlier caller + deadline), retaining the source event, operation/idempotency key, upload ID, + asset and result payload. A different identity, expired session, unrelated + `Unauthenticated`, `PermissionDenied`, or an unknown/timed-out RPC result does + not authorize a retry. This exception never re-originates, resends an OSS + PUT, or weakens the latest-generation fence. - `RequestRecordingUpload` issues a restricted, short-lived target for the original recording. The Agent makes an explicit request for a new grant after expiry; there is no automatic token renewal, SaaS upload session, or diff --git a/proto/manifest.json b/proto/manifest.json index 6acfa2c..36e4c8f 100644 --- a/proto/manifest.json +++ b/proto/manifest.json @@ -24,8 +24,8 @@ }, { "path": "proto/ERRORS.md", - "bytes": 4160, - "sha256": "04bcaa5a821fcdee25f5eeed61b1609541493b072d0a327c5c2b518fde15910f" + "bytes": 5108, + "sha256": "024f53a02c01151f7223ad49ad2d32c73d7701497dfd032bed0de820d72c7dc4" }, { "path": "proto/README.md",