From 6440b18afb69af776168c59e66b0faf8bd825437 Mon Sep 17 00:00:00 2001 From: James Salt Date: Fri, 10 Jul 2026 13:55:00 +0100 Subject: [PATCH 1/2] BCH-1341: Detect cross-SSP leverage drift, notify downstream, support re-attestation (Phase 5) Extends SyncExportOffering to flip stale leverage links to drifted (on a content-changing version bump, offering deprecation/revocation, or its leveraged authorization being deleted), raising a downstream-only inherited-revoked risk and a leverage_drifted/leverage_revoked notification without ever exposing the upstream's own risk register or evidence. A new re-attestation endpoint is the sole path that clears drift and remediates the risk. --- docs/docs.go | 79 +++++ docs/swagger.json | 79 +++++ docs/swagger.yaml | 54 ++++ internal/api/handler/oscal/api.go | 3 +- .../dashboard_suggestions_integration_test.go | 4 + .../oscal/ssp_export_offering_status_test.go | 117 +++++++ .../api/handler/oscal/ssp_export_offerings.go | 231 ++++++++++++-- .../oscal/ssp_export_offerings_unit_test.go | 141 ++++++++- internal/api/handler/oscal/ssp_leverage.go | 120 +++++++ .../api/handler/oscal/ssp_leverage_drift.go | 299 ++++++++++++++++++ .../oscal/ssp_leverage_drift_enqueue_test.go | 167 ++++++++++ .../ssp_leverage_drift_integration_test.go | 256 +++++++++++++++ .../handler/oscal/ssp_leverage_drift_test.go | 222 +++++++++++++ .../oscal/ssp_leverage_reattest_test.go | 108 +++++++ .../handler/oscal/ssp_leverage_unit_test.go | 9 + .../handler/oscal/system_security_plans.go | 44 ++- ...ecurity_plans_leveraged_auth_drift_test.go | 116 +++++++ internal/authz/manifest.yaml | 2 +- internal/service/relational/risks/models.go | 3 +- internal/service/worker/jobs.go | 3 + .../service/worker/leverage_notifications.go | 191 +++++++++++ .../worker/leverage_notifications_test.go | 76 +++++ internal/service/worker/risk_notifications.go | 12 + internal/service/worker/service.go | 47 +++ 24 files changed, 2344 insertions(+), 39 deletions(-) create mode 100644 internal/api/handler/oscal/ssp_export_offering_status_test.go create mode 100644 internal/api/handler/oscal/ssp_leverage_drift.go create mode 100644 internal/api/handler/oscal/ssp_leverage_drift_enqueue_test.go create mode 100644 internal/api/handler/oscal/ssp_leverage_drift_integration_test.go create mode 100644 internal/api/handler/oscal/ssp_leverage_drift_test.go create mode 100644 internal/api/handler/oscal/ssp_leverage_reattest_test.go create mode 100644 internal/api/handler/oscal/system_security_plans_leveraged_auth_drift_test.go create mode 100644 internal/service/worker/leverage_notifications.go create mode 100644 internal/service/worker/leverage_notifications_test.go diff --git a/docs/docs.go b/docs/docs.go index 2c652d63..5124ac13 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -20976,6 +20976,77 @@ const docTemplate = `{ ] } }, + "/oscal/system-security-plans/{id}/export-offerings/{offeringId}/status": { + "patch": { + "description": "Transitions a published export offering to deprecated or revoked,\nwhich — independent of any content change — drifts every active\nleverage link pointing at it (BCH-1341 Phase 5).", + "consumes": [ + "application/json" + ], + "produces": [ + "application/json" + ], + "tags": [ + "SSP Export Offerings" + ], + "summary": "Deprecate or revoke an export offering", + "parameters": [ + { + "type": "string", + "description": "SSP ID", + "name": "id", + "in": "path", + "required": true + }, + { + "type": "string", + "description": "Offering ID", + "name": "offeringId", + "in": "path", + "required": true + }, + { + "description": "New status", + "name": "body", + "in": "body", + "required": true, + "schema": { + "$ref": "#/definitions/oscal.updateOfferingStatusRequest" + } + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/handler.GenericDataResponse-relational_SSPExportOffering" + } + }, + "400": { + "description": "Bad Request", + "schema": { + "$ref": "#/definitions/api.Error" + } + }, + "404": { + "description": "Not Found", + "schema": { + "$ref": "#/definitions/api.Error" + } + }, + "500": { + "description": "Internal Server Error", + "schema": { + "$ref": "#/definitions/api.Error" + } + } + }, + "security": [ + { + "OAuth2Password": [] + } + ] + } + }, "/oscal/system-security-plans/{id}/import-profile": { "get": { "description": "Retrieves import-profile for a given SSP.", @@ -38873,6 +38944,14 @@ const docTemplate = `{ } } }, + "oscal.updateOfferingStatusRequest": { + "type": "object", + "properties": { + "status": { + "type": "string" + } + } + }, "oscal.upstreamResponsibility": { "type": "object", "properties": { diff --git a/docs/swagger.json b/docs/swagger.json index 849e0061..8d8c49f8 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -20970,6 +20970,77 @@ ] } }, + "/oscal/system-security-plans/{id}/export-offerings/{offeringId}/status": { + "patch": { + "description": "Transitions a published export offering to deprecated or revoked,\nwhich — independent of any content change — drifts every active\nleverage link pointing at it (BCH-1341 Phase 5).", + "consumes": [ + "application/json" + ], + "produces": [ + "application/json" + ], + "tags": [ + "SSP Export Offerings" + ], + "summary": "Deprecate or revoke an export offering", + "parameters": [ + { + "type": "string", + "description": "SSP ID", + "name": "id", + "in": "path", + "required": true + }, + { + "type": "string", + "description": "Offering ID", + "name": "offeringId", + "in": "path", + "required": true + }, + { + "description": "New status", + "name": "body", + "in": "body", + "required": true, + "schema": { + "$ref": "#/definitions/oscal.updateOfferingStatusRequest" + } + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/handler.GenericDataResponse-relational_SSPExportOffering" + } + }, + "400": { + "description": "Bad Request", + "schema": { + "$ref": "#/definitions/api.Error" + } + }, + "404": { + "description": "Not Found", + "schema": { + "$ref": "#/definitions/api.Error" + } + }, + "500": { + "description": "Internal Server Error", + "schema": { + "$ref": "#/definitions/api.Error" + } + } + }, + "security": [ + { + "OAuth2Password": [] + } + ] + } + }, "/oscal/system-security-plans/{id}/import-profile": { "get": { "description": "Retrieves import-profile for a given SSP.", @@ -38867,6 +38938,14 @@ } } }, + "oscal.updateOfferingStatusRequest": { + "type": "object", + "properties": { + "status": { + "type": "string" + } + } + }, "oscal.upstreamResponsibility": { "type": "object", "properties": { diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 8a204bb4..941559bd 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -4422,6 +4422,11 @@ definitions: name: type: string type: object + oscal.updateOfferingStatusRequest: + properties: + status: + type: string + type: object oscal.upstreamResponsibility: properties: description: @@ -25161,6 +25166,55 @@ paths: summary: Publish an export offering tags: - SSP Export Offerings + /oscal/system-security-plans/{id}/export-offerings/{offeringId}/status: + patch: + consumes: + - application/json + description: |- + Transitions a published export offering to deprecated or revoked, + which — independent of any content change — drifts every active + leverage link pointing at it (BCH-1341 Phase 5). + parameters: + - description: SSP ID + in: path + name: id + required: true + type: string + - description: Offering ID + in: path + name: offeringId + required: true + type: string + - description: New status + in: body + name: body + required: true + schema: + $ref: '#/definitions/oscal.updateOfferingStatusRequest' + produces: + - application/json + responses: + "200": + description: OK + schema: + $ref: '#/definitions/handler.GenericDataResponse-relational_SSPExportOffering' + "400": + description: Bad Request + schema: + $ref: '#/definitions/api.Error' + "404": + description: Not Found + schema: + $ref: '#/definitions/api.Error' + "500": + description: Internal Server Error + schema: + $ref: '#/definitions/api.Error' + security: + - OAuth2Password: [] + summary: Deprecate or revoke an export offering + tags: + - SSP Export Offerings /oscal/system-security-plans/{id}/import-profile: get: description: Retrieves import-profile for a given SSP. diff --git a/internal/api/handler/oscal/api.go b/internal/api/handler/oscal/api.go index d65bd31b..0198fe27 100644 --- a/internal/api/handler/oscal/api.go +++ b/internal/api/handler/oscal/api.go @@ -50,13 +50,14 @@ func RegisterHandlers(server *api.Server, logger *zap.SugaredLogger, db *gorm.DB offeringGroup := oscalGroup.Group("/ssp-export-offerings") offeringGuard := pep.For(authz.ResourceSSPExportOffering) - exportOfferingHandler := NewSSPExportOfferingHandler(logger, db) + exportOfferingHandler := NewSSPExportOfferingHandler(logger, db, jobEnqueuer) exportOfferingHandler.RegisterNested(sspGroup, sspGuard) exportOfferingHandler.Register(offeringGroup, offeringGuard) leverageHandler := NewSSPLeverageHandler(logger, db, pep.PDP(), pep.FailMode()) leverageHandler.RegisterSubscribe(offeringGroup, offeringGuard) leverageHandler.RegisterProjection(sspGroup, sspGuard) + leverageHandler.RegisterReAttest(sspGroup, sspGuard) if config.AI != nil && config.AI.Enabled { dashboardSuggestionHandler.Register(oscalGroup.Group("/system-security-plans"), jwtMiddleware, dashboardGuard) diff --git a/internal/api/handler/oscal/dashboard_suggestions_integration_test.go b/internal/api/handler/oscal/dashboard_suggestions_integration_test.go index f6d9fa3c..e74a366e 100644 --- a/internal/api/handler/oscal/dashboard_suggestions_integration_test.go +++ b/internal/api/handler/oscal/dashboard_suggestions_integration_test.go @@ -38,6 +38,10 @@ func (f *dashboardSuggestionFakeEnqueuer) EnqueueOrphanedRiskCleanup(context.Con return nil } +func (f *dashboardSuggestionFakeEnqueuer) EnqueueLeverageDriftNotification(context.Context, uuid.UUID, uuid.UUID, uuid.UUID, string) error { + return nil +} + func (f *dashboardSuggestionFakeEnqueuer) EnqueueDashboardSuggestionCells(_ context.Context, runID uuid.UUID, cellCount int) error { f.calls++ f.runID = runID diff --git a/internal/api/handler/oscal/ssp_export_offering_status_test.go b/internal/api/handler/oscal/ssp_export_offering_status_test.go new file mode 100644 index 00000000..42b66d4d --- /dev/null +++ b/internal/api/handler/oscal/ssp_export_offering_status_test.go @@ -0,0 +1,117 @@ +package oscal + +import ( + "bytes" + "net/http" + "net/http/httptest" + "testing" + + "github.com/compliance-framework/api/internal/service/relational" + "github.com/compliance-framework/api/internal/service/relational/risks" + "github.com/google/uuid" + "github.com/labstack/echo/v4" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + "gorm.io/gorm" +) + +func newOfferingStatusRequestContext(sspID, offeringID uuid.UUID, body string) (echo.Context, *httptest.ResponseRecorder) { + e := echo.New() + req := httptest.NewRequest(http.MethodPatch, "/", bytes.NewBufferString(body)) + req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON) + rec := httptest.NewRecorder() + ctx := e.NewContext(req, rec) + ctx.SetParamNames("id", "offeringId") + ctx.SetParamValues(sspID.String(), offeringID.String()) + return ctx, rec +} + +// seedPublishedOfferingWithActiveLink seeds a published offering (Version=1) plus one +// active SSPLeverageLink pointing at it, ready for a deprecate/revoke transition. +func seedPublishedOfferingWithActiveLink(t *testing.T, db *gorm.DB) (*relational.SSPExportOffering, *relational.SSPLeverageLink) { + t.Helper() + + upstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&upstreamSSP).Error) + downstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&downstreamSSP).Error) + + export := relational.Export{} + require.NoError(t, db.Create(&export).Error) + provided := relational.ProvidedControlImplementation{ExportId: *export.ID} + require.NoError(t, db.Create(&provided).Error) + + offering := relational.SSPExportOffering{ + SSPID: *upstreamSSP.ID, Title: "Offering", Status: relational.SSPExportOfferingStatusPublished, Version: 1, + } + require.NoError(t, db.Create(&offering).Error) + + link := relational.SSPLeverageLink{ + DownstreamSSPID: *downstreamSSP.ID, UpstreamSSPID: *upstreamSSP.ID, OfferingID: *offering.ID, OfferingVersion: 1, + ControlID: "ac-1", ProvidedUUID: *provided.ID, InheritedUUID: uuid.New(), LeveragedAuthUUID: uuid.New(), + Satisfaction: relational.SSPLeverageSatisfactionFull, Status: relational.SSPLeverageStatusActive, + } + require.NoError(t, db.Create(&link).Error) + + return &offering, &link +} + +func TestUpdateOfferingStatusDeprecateDriftsActiveLinks(t *testing.T) { + db := newSSPLeverageTestDB(t) + offering, link := seedPublishedOfferingWithActiveLink(t, db) + h := NewSSPExportOfferingHandler(zap.NewNop().Sugar(), db, nil) + + ctx, rec := newOfferingStatusRequestContext(offering.SSPID, *offering.ID, `{"status":"deprecated"}`) + require.NoError(t, h.UpdateOfferingStatus(ctx)) + require.Equal(t, http.StatusOK, rec.Code) + + var reloadedOffering relational.SSPExportOffering + require.NoError(t, db.First(&reloadedOffering, "id = ?", offering.ID).Error) + require.Equal(t, relational.SSPExportOfferingStatusDeprecated, reloadedOffering.Status) + require.Equal(t, 1, reloadedOffering.Version, "deprecating must not touch Version/ContentHash") + + var reloadedLink relational.SSPLeverageLink + require.NoError(t, db.First(&reloadedLink, "id = ?", link.ID).Error) + require.Equal(t, relational.SSPLeverageStatusDrifted, reloadedLink.Status) + + var riskCount int64 + require.NoError(t, db.Model(&risks.Risk{}).Where("source_type = ?", string(risks.RiskSourceTypeInheritedRevoked)).Count(&riskCount).Error) + require.Equal(t, int64(1), riskCount) +} + +func TestUpdateOfferingStatusRevokeDriftsActiveLinks(t *testing.T) { + db := newSSPLeverageTestDB(t) + offering, link := seedPublishedOfferingWithActiveLink(t, db) + h := NewSSPExportOfferingHandler(zap.NewNop().Sugar(), db, nil) + + ctx, rec := newOfferingStatusRequestContext(offering.SSPID, *offering.ID, `{"status":"revoked"}`) + require.NoError(t, h.UpdateOfferingStatus(ctx)) + require.Equal(t, http.StatusOK, rec.Code) + + var reloadedLink relational.SSPLeverageLink + require.NoError(t, db.First(&reloadedLink, "id = ?", link.ID).Error) + require.Equal(t, relational.SSPLeverageStatusDrifted, reloadedLink.Status) +} + +func TestUpdateOfferingStatusRejectsInvalidStatus(t *testing.T) { + db := newSSPLeverageTestDB(t) + offering, _ := seedPublishedOfferingWithActiveLink(t, db) + h := NewSSPExportOfferingHandler(zap.NewNop().Sugar(), db, nil) + + ctx, rec := newOfferingStatusRequestContext(offering.SSPID, *offering.ID, `{"status":"published"}`) + require.NoError(t, h.UpdateOfferingStatus(ctx)) + require.Equal(t, http.StatusBadRequest, rec.Code) +} + +func TestUpdateOfferingStatusRejectsNonPublishedOffering(t *testing.T) { + db := newSSPLeverageTestDB(t) + upstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&upstreamSSP).Error) + offering := relational.SSPExportOffering{SSPID: *upstreamSSP.ID, Title: "Draft", Status: relational.SSPExportOfferingStatusDraft} + require.NoError(t, db.Create(&offering).Error) + h := NewSSPExportOfferingHandler(zap.NewNop().Sugar(), db, nil) + + ctx, rec := newOfferingStatusRequestContext(offering.SSPID, *offering.ID, `{"status":"deprecated"}`) + require.NoError(t, h.UpdateOfferingStatus(ctx)) + require.Equal(t, http.StatusBadRequest, rec.Code) +} diff --git a/internal/api/handler/oscal/ssp_export_offerings.go b/internal/api/handler/oscal/ssp_export_offerings.go index 119bf344..088aec55 100644 --- a/internal/api/handler/oscal/ssp_export_offerings.go +++ b/internal/api/handler/oscal/ssp_export_offerings.go @@ -1,6 +1,7 @@ package oscal import ( + "context" "crypto/sha256" "encoding/hex" "encoding/json" @@ -29,17 +30,91 @@ import ( // SSPExportOfferingItem hashes into: only the fields that define what is being // offered, normalized so struct field order and JSON tag order can't perturb the hash. type canonicalOfferingItem struct { - ControlID string `json:"control_id"` - StatementID string `json:"statement_id"` - ComponentUUID string `json:"component_uuid"` - ProvidedUUID string `json:"provided_uuid"` + ControlID string `json:"control_id"` + StatementID string `json:"statement_id"` + ComponentUUID string `json:"component_uuid"` + ProvidedUUID string `json:"provided_uuid"` + ImplementationStatus string `json:"implementation_status"` +} + +// resolveItemImplementationStatuses resolves, for the distinct ProvidedUUIDs across +// items, the live ImplementationStatus of the ByComponent backing each provided +// capability (item.ProvidedUUID -> ProvidedControlImplementation.ExportId -> +// Export.ByComponentId -> ByComponent.ImplementationStatus). This is looked up fresh on +// every call rather than trusted from any cached copy, since it's exactly the signal a +// downgrade (implemented -> planned/partial) needs to be visible as drift (BCH-1341). A +// provided uuid whose chain doesn't resolve (dangling reference) contributes an empty +// status rather than failing the whole sync. +func resolveItemImplementationStatuses(db *gorm.DB, items []relational.SSPExportOfferingItem) (map[uuid.UUID]string, error) { + result := make(map[uuid.UUID]string, len(items)) + if len(items) == 0 { + return result, nil + } + + providedUUIDs := make([]uuid.UUID, 0, len(items)) + seen := make(map[uuid.UUID]struct{}, len(items)) + for _, item := range items { + if _, ok := seen[item.ProvidedUUID]; ok { + continue + } + seen[item.ProvidedUUID] = struct{}{} + providedUUIDs = append(providedUUIDs, item.ProvidedUUID) + } + + var provided []relational.ProvidedControlImplementation + if err := db.Where("id IN ?", providedUUIDs).Find(&provided).Error; err != nil { + return nil, fmt.Errorf("failed to load provided control implementations: %w", err) + } + if len(provided) == 0 { + return result, nil + } + + exportIDs := make([]uuid.UUID, 0, len(provided)) + exportIDByProvided := make(map[uuid.UUID]uuid.UUID, len(provided)) + for _, p := range provided { + exportIDByProvided[*p.ID] = p.ExportId + exportIDs = append(exportIDs, p.ExportId) + } + + var exports []relational.Export + if err := db.Where("id IN ?", exportIDs).Find(&exports).Error; err != nil { + return nil, fmt.Errorf("failed to load exports: %w", err) + } + byComponentIDByExport := make(map[uuid.UUID]uuid.UUID, len(exports)) + byComponentIDs := make([]uuid.UUID, 0, len(exports)) + for _, e := range exports { + byComponentIDByExport[*e.ID] = e.ByComponentId + byComponentIDs = append(byComponentIDs, e.ByComponentId) + } + + var byComponents []relational.ByComponent + if err := db.Where("id IN ?", byComponentIDs).Find(&byComponents).Error; err != nil { + return nil, fmt.Errorf("failed to load by-components: %w", err) + } + statusByComponent := make(map[uuid.UUID]string, len(byComponents)) + for _, bc := range byComponents { + statusByComponent[*bc.ID] = string(bc.ImplementationStatus.Data().State) + } + + for providedID, exportID := range exportIDByProvided { + byComponentID, ok := byComponentIDByExport[exportID] + if !ok { + continue + } + if status, ok := statusByComponent[byComponentID]; ok { + result[providedID] = status + } + } + + return result, nil } // computeOfferingContentHash returns a deterministic sha256 hex digest over an -// offering's curatorial content — title, description, and its items — so that two +// offering's curatorial content — title, description, and its items (including the +// live ImplementationStatus of the component backing each item, BCH-1341) — so that two // offerings (or the same offering re-read in a different item order) with identical // content always hash identically, and any real content change always changes it. -func computeOfferingContentHash(title, description string, items []relational.SSPExportOfferingItem) string { +func computeOfferingContentHash(title, description string, items []relational.SSPExportOfferingItem, statusByProvidedUUID map[uuid.UUID]string) string { canon := make([]canonicalOfferingItem, 0, len(items)) for _, item := range items { statementID := "" @@ -47,10 +122,11 @@ func computeOfferingContentHash(title, description string, items []relational.SS statementID = *item.StatementID } canon = append(canon, canonicalOfferingItem{ - ControlID: item.ControlID, - StatementID: statementID, - ComponentUUID: item.ComponentUUID.String(), - ProvidedUUID: item.ProvidedUUID.String(), + ControlID: item.ControlID, + StatementID: statementID, + ComponentUUID: item.ComponentUUID.String(), + ProvidedUUID: item.ProvidedUUID.String(), + ImplementationStatus: statusByProvidedUUID[item.ProvidedUUID], }) } sort.Slice(canon, func(i, j int) bool { @@ -85,16 +161,21 @@ func computeOfferingContentHash(title, description string, items []relational.SS // concurrency guard: it captures UpdatedAt before recomputing, then aborts inside the // write transaction if the row was modified in the meantime, rather than overwriting a // newer state with a stale computation. -func SyncExportOffering(db *gorm.DB, offeringID uuid.UUID) error { +func SyncExportOffering(db *gorm.DB, offeringID uuid.UUID) ([]driftedLinkInfo, error) { var offering relational.SSPExportOffering if err := db.Preload("Items").First(&offering, "id = ?", offeringID).Error; err != nil { - return err + return nil, err } originalUpdatedAt := offering.UpdatedAt - newHash := computeOfferingContentHash(offering.Title, offering.Description, offering.Items) + statusByProvidedUUID, err := resolveItemImplementationStatuses(db, offering.Items) + if err != nil { + return nil, err + } + newHash := computeOfferingContentHash(offering.Title, offering.Description, offering.Items, statusByProvidedUUID) - return db.Transaction(func(tx *gorm.DB) error { + var driftedLinks []driftedLinkInfo + txErr := db.Transaction(func(tx *gorm.DB) error { var current relational.SSPExportOffering if err := tx.First(¤t, "id = ?", offeringID).Error; err != nil { return err @@ -112,8 +193,21 @@ func SyncExportOffering(db *gorm.DB, offeringID uuid.UUID) error { current.ContentHash = newHash current.Version++ - return tx.Save(¤t).Error + if err := tx.Save(¤t).Error; err != nil { + return err + } + + links, err := evaluateLeverageDriftForOffering(tx, current) + if err != nil { + return err + } + driftedLinks = links + return nil }) + if txErr != nil { + return nil, txErr + } + return driftedLinks, nil } // touchOfferingUpdatedAt bumps an offering's own UpdatedAt without touching any other @@ -132,12 +226,13 @@ func touchOfferingUpdatedAt(tx *gorm.DB, offeringID uuid.UUID) error { // delete/publish, gated by ssp:export) and the top-level read-only ssp-export-offering // catalog (gated by ssp-export-offering:read). type SSPExportOfferingHandler struct { - sugar *zap.SugaredLogger - db *gorm.DB + sugar *zap.SugaredLogger + db *gorm.DB + jobEnqueuer SSPJobEnqueuer } -func NewSSPExportOfferingHandler(l *zap.SugaredLogger, db *gorm.DB) *SSPExportOfferingHandler { - return &SSPExportOfferingHandler{sugar: l, db: db} +func NewSSPExportOfferingHandler(l *zap.SugaredLogger, db *gorm.DB, jobEnqueuer SSPJobEnqueuer) *SSPExportOfferingHandler { + return &SSPExportOfferingHandler{sugar: l, db: db, jobEnqueuer: jobEnqueuer} } // RegisterNested mounts the SSP-scoped curation routes onto the SSP handler's own @@ -153,6 +248,7 @@ func (h *SSPExportOfferingHandler) RegisterNested(api *echo.Group, guard middlew api.PUT("/:id/export-offerings/:offeringId/items/:itemId", h.UpdateItem, guard.Do(authz.ActionExport)) api.DELETE("/:id/export-offerings/:offeringId/items/:itemId", h.DeleteItem, guard.Do(authz.ActionExport)) api.POST("/:id/export-offerings/:offeringId/publish", h.Publish, guard.Do(authz.ActionExport)) + api.PATCH("/:id/export-offerings/:offeringId/status", h.UpdateOfferingStatus, guard.Do(authz.ActionExport)) } // Register mounts the top-level, cross-SSP read-only catalog: list and get any @@ -626,10 +722,13 @@ func (h *SSPExportOfferingHandler) Publish(ctx echo.Context) error { } now := time.Now() + var driftedLinks []driftedLinkInfo if err := h.db.Transaction(func(tx *gorm.DB) error { - if err := SyncExportOffering(tx, *offering.ID); err != nil { + links, err := SyncExportOffering(tx, *offering.ID) + if err != nil { return err } + driftedLinks = links return tx.Model(&relational.SSPExportOffering{}). Where("id = ?", offering.ID). Updates(map[string]any{"status": relational.SSPExportOfferingStatusPublished, "published_at": &now}).Error @@ -637,6 +736,11 @@ func (h *SSPExportOfferingHandler) Publish(ctx echo.Context) error { h.sugar.Errorf("Failed to publish export offering: %v", err) return ctx.JSON(http.StatusInternalServerError, api.NewError(err)) } + if len(driftedLinks) > 0 { + enqueueCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx.Request().Context()), 10*time.Second) + defer cancel() + enqueueLeverageDriftNotifications(enqueueCtx, h.sugar, h.jobEnqueuer, driftedLinks) + } var published relational.SSPExportOffering if err := h.db.Preload("Items").First(&published, "id = ?", offering.ID).Error; err != nil { @@ -646,6 +750,93 @@ func (h *SSPExportOfferingHandler) Publish(ctx echo.Context) error { return ctx.JSON(http.StatusOK, handler.GenericDataResponse[relational.SSPExportOffering]{Data: published}) } +// updateOfferingStatusRequest is the body for UpdateOfferingStatus: only the two +// terminal, drift-triggering transitions are allowed here — draft/published stay owned +// by CreateOffering/Publish. +type updateOfferingStatusRequest struct { + Status string `json:"status"` +} + +func (r updateOfferingStatusRequest) validate() error { + switch relational.SSPExportOfferingStatus(r.Status) { + case relational.SSPExportOfferingStatusDeprecated, relational.SSPExportOfferingStatusRevoked: + return nil + default: + return fmt.Errorf("status must be %q or %q", relational.SSPExportOfferingStatusDeprecated, relational.SSPExportOfferingStatusRevoked) + } +} + +// UpdateOfferingStatus godoc +// +// @Summary Deprecate or revoke an export offering +// @Description Transitions a published export offering to deprecated or revoked, +// @Description which — independent of any content change — drifts every active +// @Description leverage link pointing at it (BCH-1341 Phase 5). +// @Tags SSP Export Offerings +// @Accept json +// @Produce json +// @Param id path string true "SSP ID" +// @Param offeringId path string true "Offering ID" +// @Param body body updateOfferingStatusRequest true "New status" +// @Success 200 {object} handler.GenericDataResponse[relational.SSPExportOffering] +// @Failure 400 {object} api.Error +// @Failure 404 {object} api.Error +// @Failure 500 {object} api.Error +// @Security OAuth2Password +// @Router /oscal/system-security-plans/{id}/export-offerings/{offeringId}/status [patch] +func (h *SSPExportOfferingHandler) UpdateOfferingStatus(ctx echo.Context) error { + offering, ok := h.resolveOfferingForSSP(ctx) + if !ok { + return nil + } + + var req updateOfferingStatusRequest + if err := ctx.Bind(&req); err != nil { + return ctx.JSON(http.StatusBadRequest, api.NewError(err)) + } + if err := req.validate(); err != nil { + return ctx.JSON(http.StatusBadRequest, api.NewError(err)) + } + + if offering.Status != relational.SSPExportOfferingStatusPublished { + return ctx.JSON(http.StatusBadRequest, api.NewError(fmt.Errorf("cannot transition an offering with status %q", offering.Status))) + } + + newStatus := relational.SSPExportOfferingStatus(req.Status) + var driftedLinks []driftedLinkInfo + if err := h.db.Transaction(func(tx *gorm.DB) error { + if err := tx.Model(&relational.SSPExportOffering{}). + Where("id = ?", offering.ID). + Update("status", newStatus).Error; err != nil { + return err + } + + updated := *offering + updated.Status = newStatus + links, err := evaluateLeverageDriftForOffering(tx, updated) + if err != nil { + return err + } + driftedLinks = links + return nil + }); err != nil { + h.sugar.Errorf("Failed to update export offering status: %v", err) + return ctx.JSON(http.StatusInternalServerError, api.NewError(err)) + } + if len(driftedLinks) > 0 { + enqueueCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx.Request().Context()), 10*time.Second) + defer cancel() + enqueueLeverageDriftNotifications(enqueueCtx, h.sugar, h.jobEnqueuer, driftedLinks) + } + + var updated relational.SSPExportOffering + if err := h.db.Preload("Items").First(&updated, "id = ?", offering.ID).Error; err != nil { + return ctx.JSON(http.StatusInternalServerError, api.NewError(err)) + } + + return ctx.JSON(http.StatusOK, handler.GenericDataResponse[relational.SSPExportOffering]{Data: updated}) +} + const ( defaultExportOfferingCatalogLimit = 100 maxExportOfferingCatalogLimit = 1000 diff --git a/internal/api/handler/oscal/ssp_export_offerings_unit_test.go b/internal/api/handler/oscal/ssp_export_offerings_unit_test.go index def7c7bf..093b356b 100644 --- a/internal/api/handler/oscal/ssp_export_offerings_unit_test.go +++ b/internal/api/handler/oscal/ssp_export_offerings_unit_test.go @@ -5,8 +5,10 @@ import ( "time" "github.com/compliance-framework/api/internal/service/relational" + "github.com/compliance-framework/api/internal/service/relational/risks" "github.com/google/uuid" "github.com/stretchr/testify/require" + "gorm.io/datatypes" "gorm.io/driver/sqlite" "gorm.io/gorm" ) @@ -20,8 +22,8 @@ func TestComputeOfferingContentHashIsOrderIndependent(t *testing.T) { a := relational.SSPExportOfferingItem{ControlID: "ac-1", ComponentUUID: uuid.New(), ProvidedUUID: uuid.New()} b := relational.SSPExportOfferingItem{ControlID: "ac-2", StatementID: statementID("ac-2_stmt.a"), ComponentUUID: uuid.New(), ProvidedUUID: uuid.New()} - hash1 := computeOfferingContentHash("Title", "Description", []relational.SSPExportOfferingItem{a, b}) - hash2 := computeOfferingContentHash("Title", "Description", []relational.SSPExportOfferingItem{b, a}) + hash1 := computeOfferingContentHash("Title", "Description", []relational.SSPExportOfferingItem{a, b}, nil) + hash2 := computeOfferingContentHash("Title", "Description", []relational.SSPExportOfferingItem{b, a}, nil) require.Equal(t, hash1, hash2) } @@ -32,24 +34,91 @@ func TestComputeOfferingContentHashChangesWithContent(t *testing.T) { items := []relational.SSPExportOfferingItem{ {ControlID: "ac-1", ComponentUUID: uuid.New(), ProvidedUUID: uuid.New()}, } - base := computeOfferingContentHash("Title", "Description", items) + base := computeOfferingContentHash("Title", "Description", items, nil) - require.NotEqual(t, base, computeOfferingContentHash("Different Title", "Description", items)) - require.NotEqual(t, base, computeOfferingContentHash("Title", "Different Description", items)) + require.NotEqual(t, base, computeOfferingContentHash("Different Title", "Description", items, nil)) + require.NotEqual(t, base, computeOfferingContentHash("Title", "Different Description", items, nil)) changedItems := []relational.SSPExportOfferingItem{ {ControlID: "ac-2", ComponentUUID: items[0].ComponentUUID, ProvidedUUID: items[0].ProvidedUUID}, } - require.NotEqual(t, base, computeOfferingContentHash("Title", "Description", changedItems)) + require.NotEqual(t, base, computeOfferingContentHash("Title", "Description", changedItems, nil)) } func newSyncExportOfferingTestDB(t *testing.T) *gorm.DB { db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) require.NoError(t, err) - require.NoError(t, db.AutoMigrate(&relational.SSPExportOffering{}, &relational.SSPExportOfferingItem{})) + require.NoError(t, db.AutoMigrate( + &relational.SSPExportOffering{}, + &relational.SSPExportOfferingItem{}, + &relational.ProvidedControlImplementation{}, + &relational.Export{}, + &relational.ByComponent{}, + &relational.SSPLeverageLink{}, + &relational.ControlImplementationResponsibility{}, + &relational.Profile{}, + &relational.Control{}, + &relational.SSPProfile{}, + &risks.Risk{}, + &risks.RiskEvent{}, + &risks.RiskScore{}, + &risks.RiskResponsibilityLink{}, + &risks.RiskControlLink{}, + )) return db } +// TestSyncExportOfferingBumpsVersionOnImplementationStatusDowngrade: a backing +// ByComponent's ImplementationStatus downgrading (implemented -> planned) is a content +// change even though none of an item's own fields (ControlID/StatementID/ComponentUUID/ +// ProvidedUUID) changed — this is the "downgrading a provided status" drift trigger +// (BCH-1341), resolved live via item.ProvidedUUID -> ProvidedControlImplementation. +// ExportId -> Export.ByComponentId -> ByComponent.ImplementationStatus. +func TestSyncExportOfferingBumpsVersionOnImplementationStatusDowngrade(t *testing.T) { + db := newSyncExportOfferingTestDB(t) + + byComponent := relational.ByComponent{ + ImplementationStatus: datatypes.NewJSONType(relational.ImplementationStatus{State: relational.ImplementationStatusImplemented}), + } + require.NoError(t, db.Create(&byComponent).Error) + + export := relational.Export{ByComponentId: *byComponent.ID} + require.NoError(t, db.Create(&export).Error) + + provided := relational.ProvidedControlImplementation{ExportId: *export.ID} + require.NoError(t, db.Create(&provided).Error) + + offering := relational.SSPExportOffering{SSPID: uuid.New(), Title: "Offering", Status: relational.SSPExportOfferingStatusDraft} + require.NoError(t, db.Create(&offering).Error) + require.NoError(t, db.Create(&relational.SSPExportOfferingItem{ + OfferingID: *offering.ID, ControlID: "ac-1", ComponentUUID: byComponent.ComponentUUID, ProvidedUUID: *provided.ID, + }).Error) + + _, err := SyncExportOffering(db, *offering.ID) + require.NoError(t, err) + var afterFirst relational.SSPExportOffering + require.NoError(t, db.First(&afterFirst, "id = ?", offering.ID).Error) + require.Equal(t, 1, afterFirst.Version) + + // Re-sync with no change: no-op. + _, err = SyncExportOffering(db, *offering.ID) + require.NoError(t, err) + var afterNoop relational.SSPExportOffering + require.NoError(t, db.First(&afterNoop, "id = ?", offering.ID).Error) + require.Equal(t, 1, afterNoop.Version) + + // Downgrade the backing component's status, then re-sync: must bump. + require.NoError(t, db.Model(&byComponent).Update( + "implementation_status", datatypes.NewJSONType(relational.ImplementationStatus{State: relational.ImplementationStatusPlanned}), + ).Error) + _, err = SyncExportOffering(db, *offering.ID) + require.NoError(t, err) + var afterDowngrade relational.SSPExportOffering + require.NoError(t, db.First(&afterDowngrade, "id = ?", offering.ID).Error) + require.Equal(t, 2, afterDowngrade.Version) + require.NotEqual(t, afterFirst.ContentHash, afterDowngrade.ContentHash) +} + // TestSyncExportOfferingBumpsVersionOnlyWhenContentChanges: calling SyncExportOffering // on an offering whose content hasn't changed since the last sync must be a no-op (no // version bump); it must only bump when an item was actually added. @@ -63,14 +132,16 @@ func TestSyncExportOfferingBumpsVersionOnlyWhenContentChanges(t *testing.T) { }).Error) // First sync: no hash stored yet, so this must bump the version and set the hash. - require.NoError(t, SyncExportOffering(db, *offering.ID)) + _, err := SyncExportOffering(db, *offering.ID) + require.NoError(t, err) var afterFirst relational.SSPExportOffering require.NoError(t, db.First(&afterFirst, "id = ?", offering.ID).Error) require.Equal(t, 1, afterFirst.Version) require.NotEmpty(t, afterFirst.ContentHash) // Republish with no content change: must be a no-op. - require.NoError(t, SyncExportOffering(db, *offering.ID)) + _, err = SyncExportOffering(db, *offering.ID) + require.NoError(t, err) var afterSecond relational.SSPExportOffering require.NoError(t, db.First(&afterSecond, "id = ?", offering.ID).Error) require.Equal(t, 1, afterSecond.Version) @@ -80,13 +151,61 @@ func TestSyncExportOfferingBumpsVersionOnlyWhenContentChanges(t *testing.T) { require.NoError(t, db.Create(&relational.SSPExportOfferingItem{ OfferingID: *offering.ID, ControlID: "ac-2", ComponentUUID: uuid.New(), ProvidedUUID: uuid.New(), }).Error) - require.NoError(t, SyncExportOffering(db, *offering.ID)) + _, err = SyncExportOffering(db, *offering.ID) + require.NoError(t, err) var afterChange relational.SSPExportOffering require.NoError(t, db.First(&afterChange, "id = ?", offering.ID).Error) require.Equal(t, 2, afterChange.Version) require.NotEqual(t, afterFirst.ContentHash, afterChange.ContentHash) } +// TestSyncExportOfferingDriftsLeverageLinksOnVersionBump: re-syncing an offering whose +// content actually changed (BCH-1341) must drift every active leverage link pointing at +// it — the version-bump trigger, exercised through the real public entrypoint rather +// than evaluateLeverageDriftForOffering directly. +func TestSyncExportOfferingDriftsLeverageLinksOnVersionBump(t *testing.T) { + db := newSyncExportOfferingTestDB(t) + + export := relational.Export{} + require.NoError(t, db.Create(&export).Error) + provided := relational.ProvidedControlImplementation{ExportId: *export.ID} + require.NoError(t, db.Create(&provided).Error) + + offering := relational.SSPExportOffering{SSPID: uuid.New(), Title: "Offering", Status: relational.SSPExportOfferingStatusDraft} + require.NoError(t, db.Create(&offering).Error) + require.NoError(t, db.Create(&relational.SSPExportOfferingItem{ + OfferingID: *offering.ID, ControlID: "ac-1", ComponentUUID: uuid.New(), ProvidedUUID: *provided.ID, + }).Error) + + // First sync sets Version=1; subscribe snapshots that as the link's OfferingVersion. + _, err := SyncExportOffering(db, *offering.ID) + require.NoError(t, err) + + link := relational.SSPLeverageLink{ + DownstreamSSPID: uuid.New(), UpstreamSSPID: offering.SSPID, OfferingID: *offering.ID, OfferingVersion: 1, + ControlID: "ac-1", ProvidedUUID: *provided.ID, InheritedUUID: uuid.New(), LeveragedAuthUUID: uuid.New(), + Satisfaction: relational.SSPLeverageSatisfactionFull, Status: relational.SSPLeverageStatusActive, + } + require.NoError(t, db.Create(&link).Error) + + // Change content, then re-sync: Version bumps to 2, link is now behind. + require.NoError(t, db.Create(&relational.SSPExportOfferingItem{ + OfferingID: *offering.ID, ControlID: "ac-2", ComponentUUID: uuid.New(), ProvidedUUID: uuid.New(), + }).Error) + driftedLinks, err := SyncExportOffering(db, *offering.ID) + require.NoError(t, err) + require.Len(t, driftedLinks, 1) + require.Equal(t, *link.ID, driftedLinks[0].LinkID) + + var reloadedLink relational.SSPLeverageLink + require.NoError(t, db.First(&reloadedLink, "id = ?", link.ID).Error) + require.Equal(t, relational.SSPLeverageStatusDrifted, reloadedLink.Status) + + var riskCount int64 + require.NoError(t, db.Model(&risks.Risk{}).Where("source_type = ?", string(risks.RiskSourceTypeInheritedRevoked)).Count(&riskCount).Error) + require.Equal(t, int64(1), riskCount) +} + // TestSyncExportOfferingAbortsOnConcurrentModification: SyncExportOffering reads the // offering once to snapshot UpdatedAt before recomputing the hash, then again inside its // write transaction. If another write lands in between those two reads, it must abort @@ -112,7 +231,7 @@ func TestSyncExportOfferingAbortsOnConcurrentModification(t *testing.T) { })) t.Cleanup(func() { _ = db.Callback().Query().Remove("test:simulate-concurrent-write") }) - err := SyncExportOffering(db, *offering.ID) + _, err := SyncExportOffering(db, *offering.ID) require.Error(t, err) var reloaded relational.SSPExportOffering diff --git a/internal/api/handler/oscal/ssp_leverage.go b/internal/api/handler/oscal/ssp_leverage.go index 2fe042ca..28396843 100644 --- a/internal/api/handler/oscal/ssp_leverage.go +++ b/internal/api/handler/oscal/ssp_leverage.go @@ -17,6 +17,7 @@ import ( "github.com/compliance-framework/api/internal/api/middleware" "github.com/compliance-framework/api/internal/authz" "github.com/compliance-framework/api/internal/service/relational" + "github.com/compliance-framework/api/internal/service/relational/risks" ) // thisSystemComponentType is the OSCAL convention for a placeholder component @@ -165,6 +166,14 @@ func (h *SSPLeverageHandler) RegisterSubscribe(g *echo.Group, guard middleware.R g.POST("/:id/subscribe", h.Subscribe, guard.Do(authz.ActionSubscribe)) } +// RegisterReAttest mounts the re-attestation route onto the SSP handler's own group, +// gated by the same ssp:update guard as any other write against the downstream SSP's own +// data — the SSP id is in the URL here (unlike Subscribe), so no bespoke +// authorizeDownstreamUpdate-style check is needed. +func (h *SSPLeverageHandler) RegisterReAttest(g *echo.Group, guard middleware.ResourceGuard) { + g.POST("/:id/leveraged-controls/:linkId/attest", h.ReAttest, guard.Do(authz.ActionUpdate)) +} + // RegisterProjection mounts the leveraged-controls projection onto the SSP handler's own // route group, gated by the standard ssp:read. func (h *SSPLeverageHandler) RegisterProjection(g *echo.Group, guard middleware.ResourceGuard) { @@ -764,3 +773,114 @@ func (h *SSPLeverageHandler) LeveragedControls(ctx echo.Context) error { return ctx.JSON(http.StatusOK, handler.GenericDataListResponse[leveragedControlResponse]{Data: result}) } + +// ReAttest clears drift on a leverage link (BCH-1341): bumps OfferingVersion to the +// offering's current Version, refreshes Satisfaction against the current upstream +// responsibility set and downstream satisfied set, flips the link back to active, and +// marks the associated drift risk remediated — all in one transaction. Human-in-the-loop +// only: this is the sole path that clears drift, there is no automatic re-activation. +func (h *SSPLeverageHandler) ReAttest(ctx echo.Context) error { + sspIdParam := ctx.Param("id") + sspID, err := uuid.Parse(sspIdParam) + if err != nil { + h.sugar.Warnw("Invalid SSP id", "id", sspIdParam, "error", err) + return ctx.JSON(http.StatusBadRequest, api.NewError(err)) + } + + linkIdParam := ctx.Param("linkId") + linkID, err := uuid.Parse(linkIdParam) + if err != nil { + h.sugar.Warnw("Invalid leverage link id", "linkId", linkIdParam, "error", err) + return ctx.JSON(http.StatusBadRequest, api.NewError(err)) + } + + var link relational.SSPLeverageLink + if err := h.db.First(&link, "id = ?", linkID).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return ctx.JSON(http.StatusNotFound, api.NewError(fmt.Errorf("leverage link not found"))) + } + h.sugar.Errorf("Failed to load leverage link: %v", err) + return ctx.JSON(http.StatusInternalServerError, api.NewError(err)) + } + if link.DownstreamSSPID != sspID { + return ctx.JSON(http.StatusNotFound, api.NewError(fmt.Errorf("leverage link not found"))) + } + if link.Status != relational.SSPLeverageStatusDrifted { + return ctx.JSON(http.StatusBadRequest, api.NewError(fmt.Errorf("leverage link is not drifted"))) + } + + attestedBy := actorUserID(ctx) + now := time.Now() + + if err := h.db.Transaction(func(tx *gorm.DB) error { + var offering relational.SSPExportOffering + if err := tx.First(&offering, "id = ?", link.OfferingID).Error; err != nil { + return fmt.Errorf("failed to load offering: %w", err) + } + + var inherited relational.InheritedControlImplementation + if err := tx.First(&inherited, "id = ?", link.InheritedUUID).Error; err != nil { + return fmt.Errorf("failed to load inherited control implementation: %w", err) + } + + var satisfiedRows []relational.SatisfiedControlImplementationResponsibility + if err := tx.Where("by_component_id = ?", inherited.ByComponentId).Find(&satisfiedRows).Error; err != nil { + return fmt.Errorf("failed to load satisfied responsibilities: %w", err) + } + satisfiedUUIDs := make(map[uuid.UUID]bool, len(satisfiedRows)) + for _, s := range satisfiedRows { + satisfiedUUIDs[s.ResponsibilityUuid] = true + } + + fullSet, err := resolveUpstreamResponsibilities(tx, link.ProvidedUUID) + if err != nil { + return fmt.Errorf("failed to resolve upstream responsibilities: %w", err) + } + satisfaction, _ := deriveSatisfaction(fullSet, satisfiedUUIDs) + + if err := tx.Model(&relational.SSPLeverageLink{}).Where("id = ?", link.ID).Updates(map[string]any{ + "offering_version": offering.Version, + "status": relational.SSPLeverageStatusActive, + "satisfaction": satisfaction, + "attested_at": &now, + "attested_by_id": attestedBy, + }).Error; err != nil { + return fmt.Errorf("failed to update leverage link: %w", err) + } + + dedupeKey := computeDedupeKeyForLeverageDrift(*link.ID) + var driftRisk risks.Risk + err = tx.Where("dedupe_key = ? AND status != ?", dedupeKey, risks.RiskStatusClosed).First(&driftRisk).Error + if err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil + } + return fmt.Errorf("failed to load drift risk: %w", err) + } + + oldStatus := driftRisk.Status + driftRisk.Status = string(risks.RiskStatusRemediated) + if err := tx.Save(&driftRisk).Error; err != nil { + return fmt.Errorf("failed to remediate drift risk: %w", err) + } + if err := emitLeverageDriftRiskEvent(tx, *driftRisk.ID, string(risks.RiskEventTypeStatusChange), map[string]interface{}{ + "from": oldStatus, + "to": string(risks.RiskStatusRemediated), + "reason": "re-attested", + "leverage_link_id": link.ID, + }, now); err != nil { + return err + } + return risks.NewRiskService(tx).RecordRiskScoreSnapshot(tx, *driftRisk.ID, risks.RiskEventTypeStatusChange, attestedBy, now) + }); err != nil { + h.sugar.Errorf("Failed to re-attest leverage link: %v", err) + return ctx.JSON(http.StatusInternalServerError, api.NewError(err)) + } + + var updated relational.SSPLeverageLink + if err := h.db.First(&updated, "id = ?", link.ID).Error; err != nil { + return ctx.JSON(http.StatusInternalServerError, api.NewError(err)) + } + + return ctx.JSON(http.StatusOK, handler.GenericDataResponse[relational.SSPLeverageLink]{Data: updated}) +} diff --git a/internal/api/handler/oscal/ssp_leverage_drift.go b/internal/api/handler/oscal/ssp_leverage_drift.go new file mode 100644 index 00000000..11e7fc64 --- /dev/null +++ b/internal/api/handler/oscal/ssp_leverage_drift.go @@ -0,0 +1,299 @@ +package oscal + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/google/uuid" + "go.uber.org/zap" + "gorm.io/datatypes" + "gorm.io/gorm" + "gorm.io/gorm/clause" + + "github.com/compliance-framework/api/internal/service/relational" + "github.com/compliance-framework/api/internal/service/relational/risks" +) + +// driftedLinkInfo describes one SSPLeverageLink that was just flipped to drifted, +// along with the risk that now represents it downstream — returned so callers can +// enqueue a leverage_drifted/leverage_revoked notification after their transaction +// commits (BCH-1341). +type driftedLinkInfo struct { + LinkID uuid.UUID + DownstreamSSPID uuid.UUID + RiskID uuid.UUID + Reason string +} + +// enqueueLeverageDriftNotifications calls jobEnqueuer once per drifted link — using only +// plain uuid.UUID/string parameters (mirroring EnqueueOrphanedRiskCleanup's shape) +// rather than a shared struct type, since a struct type defined in this package couldn't +// satisfy an interface method implemented by *worker.Service (a different package) +// without introducing a circular import between the oscal and worker packages. A failed +// enqueue is logged and swallowed, not returned — the drift itself already committed +// successfully, and losing a notification shouldn't fail the request that caused it +// (same non-atomicity tradeoff as EnqueueOrphanedRiskCleanup's call sites). +func enqueueLeverageDriftNotifications(ctx context.Context, sugar *zap.SugaredLogger, jobEnqueuer SSPJobEnqueuer, links []driftedLinkInfo) { + if jobEnqueuer == nil { + return + } + for _, info := range links { + if err := jobEnqueuer.EnqueueLeverageDriftNotification(ctx, info.RiskID, info.LinkID, info.DownstreamSSPID, info.Reason); err != nil { + sugar.Warnw("Failed to enqueue leverage drift notification", + "risk_id", info.RiskID, "link_id", info.LinkID, "error", err) + } + } +} + +// computeDedupeKeyForLeverageDrift returns the dedupe key for the drift risk +// associated with a single SSPLeverageLink. One risk per leverage link: the link +// itself is the natural, deterministic scope (no risk template is involved, unlike +// evidence-driven risks), and the key is directly parseable back to the link that +// produced it without needing a separate link table. +func computeDedupeKeyForLeverageDrift(linkID uuid.UUID) string { + return fmt.Sprintf("leverage-drift:%s", linkID) +} + +// applyDriftToLink flips an active SSPLeverageLink to drifted and creates (or reopens) +// its inherited-revoked risk. It is the single unit every drift trigger (version bump, +// offering deprecate/revoke, leveraged-authorization delete) calls into. Idempotent: a +// link that isn't currently active is left untouched (zero value, ok=false returned) so +// re-running a trigger never re-drifts or duplicates work. +func applyDriftToLink(tx *gorm.DB, link *relational.SSPLeverageLink, reason string) (info driftedLinkInfo, ok bool, err error) { + if link.Status != relational.SSPLeverageStatusActive { + return driftedLinkInfo{}, false, nil + } + + if err := tx.Model(&relational.SSPLeverageLink{}). + Where("id = ?", link.ID). + Update("status", relational.SSPLeverageStatusDrifted).Error; err != nil { + return driftedLinkInfo{}, false, fmt.Errorf("failed to flip leverage link to drifted: %w", err) + } + link.Status = relational.SSPLeverageStatusDrifted + + dedupeKey := computeDedupeKeyForLeverageDrift(*link.ID) + + var existingRisk risks.Risk + err = tx.Where("dedupe_key = ? AND status != ?", dedupeKey, risks.RiskStatusClosed).First(&existingRisk).Error + switch { + case err == nil: + if updateErr := reopenLeverageDriftRisk(tx, &existingRisk, reason); updateErr != nil { + return driftedLinkInfo{}, false, updateErr + } + case errors.Is(err, gorm.ErrRecordNotFound): + if existingRisk, err = createLeverageDriftRisk(tx, *link, dedupeKey, reason); err != nil { + return driftedLinkInfo{}, false, err + } + default: + return driftedLinkInfo{}, false, fmt.Errorf("failed to check for existing leverage drift risk: %w", err) + } + + if err := createLeverageDriftRiskLinks(tx, *existingRisk.ID, *link); err != nil { + return driftedLinkInfo{}, false, err + } + + return driftedLinkInfo{ + LinkID: *link.ID, + DownstreamSSPID: link.DownstreamSSPID, + RiskID: *existingRisk.ID, + Reason: reason, + }, true, nil +} + +// createLeverageDriftRisk creates a new inherited-revoked risk for a just-drifted +// leverage link. +func createLeverageDriftRisk(tx *gorm.DB, link relational.SSPLeverageLink, dedupeKey, reason string) (risks.Risk, error) { + now := time.Now().UTC() + + newRisk := risks.Risk{ + Title: fmt.Sprintf("Leveraged control %s drifted", link.ControlID), + Description: fmt.Sprintf("The upstream capability leveraged for control %s changed: %s.", link.ControlID, reason), + Status: string(risks.RiskStatusOpen), + SSPID: link.DownstreamSSPID, + SourceType: string(risks.RiskSourceTypeInheritedRevoked), + DedupeKey: dedupeKey, + FirstSeenAt: now, + LastSeenAt: now, + } + + if err := tx.Create(&newRisk).Error; err != nil { + return risks.Risk{}, fmt.Errorf("failed to create leverage drift risk: %w", err) + } + + if err := emitLeverageDriftRiskEvent(tx, *newRisk.ID, string(risks.RiskEventTypeCreated), map[string]interface{}{ + "leverage_link_id": link.ID, + "dedupe_key": dedupeKey, + "reason": reason, + }, now); err != nil { + return risks.Risk{}, err + } + if err := risks.NewRiskService(tx).RecordRiskScoreSnapshot(tx, *newRisk.ID, risks.RiskEventTypeCreated, nil, now); err != nil { + return risks.Risk{}, fmt.Errorf("failed to record created risk score snapshot: %w", err) + } + + return newRisk, nil +} + +// reopenLeverageDriftRisk reopens a previously-remediated drift risk (re-attested, then +// drifted again) and bumps LastSeenAt on any other non-closed status. +func reopenLeverageDriftRisk(tx *gorm.DB, existingRisk *risks.Risk, reason string) error { + now := time.Now().UTC() + previousLastSeen := existingRisk.LastSeenAt + + reopened := false + oldStatus := existingRisk.Status + if existingRisk.Status == string(risks.RiskStatusRemediated) { + existingRisk.Status = string(risks.RiskStatusOpen) + reopened = true + } + existingRisk.LastSeenAt = now + + if err := tx.Save(existingRisk).Error; err != nil { + return fmt.Errorf("failed to update existing leverage drift risk: %w", err) + } + + if reopened { + if err := emitLeverageDriftRiskEvent(tx, *existingRisk.ID, string(risks.RiskEventTypeStatusChange), map[string]interface{}{ + "from": oldStatus, + "to": string(risks.RiskStatusOpen), + "reason": reason, + }, now); err != nil { + return err + } + if err := risks.NewRiskService(tx).RecordRiskScoreSnapshot(tx, *existingRisk.ID, risks.RiskEventTypeStatusChange, nil, now); err != nil { + return fmt.Errorf("failed to record reopened risk score snapshot: %w", err) + } + } + + return emitLeverageDriftRiskEvent(tx, *existingRisk.ID, string(risks.RiskEventTypeLastSeen), map[string]interface{}{ + "previous_last_seen": previousLastSeen, + "new_last_seen": now, + "reason": reason, + }, now) +} + +// createLeverageDriftRiskLinks links a drift risk to every downstream responsibility +// consumer of the leveraged capability (RiskResponsibilityLink, unfiltered — unlike the +// evidence-driven BCH-1339/1340 arm, drift affects the whole leveraged capability, not a +// filter-matched subset) and, best-effort, to the downstream SSP's own catalog control +// entry (RiskControlLink) if one resolves. Both idempotent via OnConflict{DoNothing}. +func createLeverageDriftRiskLinks(tx *gorm.DB, riskID uuid.UUID, link relational.SSPLeverageLink) error { + now := time.Now().UTC() + + var responsibilities []relational.ControlImplementationResponsibility + if err := tx.Where("provided_uuid = ?", link.ProvidedUUID).Find(&responsibilities).Error; err != nil { + return fmt.Errorf("failed to load responsibilities for leverage drift link: %w", err) + } + for _, resp := range responsibilities { + respLink := &risks.RiskResponsibilityLink{ + RiskID: riskID, + ResponsibilityUUID: *resp.ID, + CreatedAt: now, + } + if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(respLink).Error; err != nil { + return fmt.Errorf("failed to create responsibility link for leverage drift risk: %w", err) + } + } + + catalogID, ok, err := resolveCatalogIDForControl(tx, link.DownstreamSSPID, link.ControlID) + if err != nil { + return fmt.Errorf("failed to resolve catalog id for leverage drift link: %w", err) + } + if ok { + controlLink := &risks.RiskControlLink{ + RiskID: riskID, + CatalogID: catalogID, + ControlID: link.ControlID, + CreatedAt: now, + } + if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(controlLink).Error; err != nil { + return fmt.Errorf("failed to create control link for leverage drift risk: %w", err) + } + } + + return nil +} + +// resolveCatalogIDForControl finds the catalog id a downstream SSP's own profile +// associates with controlID, mirroring the join risk_evidence_worker.go's +// resolveSSPsViaFilters uses for its filter_controls arm. Best-effort: a downstream SSP +// need not have the leveraged control in its own profile at all (it may rely on the +// leveraged capability alone), so a miss is not an error. +func resolveCatalogIDForControl(tx *gorm.DB, downstreamSSPID uuid.UUID, controlID string) (uuid.UUID, bool, error) { + type catalogRow struct { + ControlCatalogID uuid.UUID `gorm:"column:control_catalog_id"` + } + var rows []catalogRow + if err := tx.Table("ssp_profiles sp"). + Select("DISTINCT pc.control_catalog_id"). + Joins("JOIN profile_controls pc ON CAST(sp.profile_id AS uuid) = CAST(pc.profile_id AS uuid) AND UPPER(pc.control_id) = UPPER(?)", controlID). + Where("CAST(sp.system_security_plan_id AS uuid) = CAST(? AS uuid)", downstreamSSPID). + Scan(&rows).Error; err != nil { + return uuid.Nil, false, err + } + if len(rows) == 0 { + return uuid.Nil, false, nil + } + return rows[0].ControlCatalogID, true, nil +} + +// emitLeverageDriftRiskEvent mirrors risk_evidence_worker.go's emitRiskEventAt: builds +// and creates a RiskEvent row directly, since the risks package's own event-logging +// helpers (logRiskEvent/logRiskEventWithSnapshot) are unexported. +func emitLeverageDriftRiskEvent(tx *gorm.DB, riskID uuid.UUID, eventType string, payload map[string]interface{}, occurredAt time.Time) error { + payloadMap := datatypes.JSONMap(payload) + details := risks.BuildRiskEventDetails(eventType, payloadMap, occurredAt) + + event := &risks.RiskEvent{ + RiskID: riskID, + EventType: eventType, + OccurredAt: occurredAt, + Details: &details, + Payload: payloadMap, + } + + return tx.Create(event).Error +} + +// evaluateLeverageDriftForOffering re-evaluates every SSPLeverageLink pointing at +// offering and drifts the ones whose snapshot is now behind: either the offering's +// Version has moved past what the link recorded at subscribe/re-attest time (covers +// both a real content change and a backing component's ImplementationStatus downgrade, +// since both bump Version via SyncExportOffering), or the offering itself has been +// deprecated/revoked (independent of Version). Must be called within the same +// transaction that persisted the offering's current Version/Status. +func evaluateLeverageDriftForOffering(tx *gorm.DB, offering relational.SSPExportOffering) ([]driftedLinkInfo, error) { + var links []relational.SSPLeverageLink + if err := tx.Where("offering_id = ?", offering.ID).Find(&links).Error; err != nil { + return nil, fmt.Errorf("failed to load leverage links for offering: %w", err) + } + + var results []driftedLinkInfo + for i := range links { + link := links[i] + + var reason string + switch { + case offering.Status == relational.SSPExportOfferingStatusDeprecated: + reason = "upstream offering deprecated" + case offering.Status == relational.SSPExportOfferingStatusRevoked: + reason = "upstream offering revoked" + case offering.Version > link.OfferingVersion: + reason = "upstream offering content changed" + default: + continue + } + + info, ok, err := applyDriftToLink(tx, &link, reason) + if err != nil { + return nil, err + } + if ok { + results = append(results, info) + } + } + + return results, nil +} diff --git a/internal/api/handler/oscal/ssp_leverage_drift_enqueue_test.go b/internal/api/handler/oscal/ssp_leverage_drift_enqueue_test.go new file mode 100644 index 00000000..ca46fab2 --- /dev/null +++ b/internal/api/handler/oscal/ssp_leverage_drift_enqueue_test.go @@ -0,0 +1,167 @@ +package oscal + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/compliance-framework/api/internal/service/relational" + "github.com/google/uuid" + "github.com/labstack/echo/v4" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +func newPublishRequestContext(sspID, offeringID uuid.UUID) (echo.Context, *httptest.ResponseRecorder) { + e := echo.New() + req := httptest.NewRequest(http.MethodPost, "/", nil) + rec := httptest.NewRecorder() + ctx := e.NewContext(req, rec) + ctx.SetParamNames("id", "offeringId") + ctx.SetParamValues(sspID.String(), offeringID.String()) + return ctx, rec +} + +// spyJobEnqueuer records every EnqueueLeverageDriftNotification call so tests can assert +// exactly which (risk, link) pairs were enqueued, without needing a real River client. +type spyJobEnqueuer struct { + calls []spyEnqueueCall +} + +type spyEnqueueCall struct { + RiskID uuid.UUID + LinkID uuid.UUID + DownstreamSSPID uuid.UUID + Reason string +} + +func (s *spyJobEnqueuer) EnqueueOrphanedRiskCleanup(context.Context, uuid.UUID, *uuid.UUID, *uuid.UUID) error { + return nil +} + +func (s *spyJobEnqueuer) EnqueueDashboardSuggestionCells(context.Context, uuid.UUID, int) error { + return nil +} + +func (s *spyJobEnqueuer) EnqueueLeverageDriftNotification(_ context.Context, riskID, linkID, downstreamSSPID uuid.UUID, reason string) error { + s.calls = append(s.calls, spyEnqueueCall{RiskID: riskID, LinkID: linkID, DownstreamSSPID: downstreamSSPID, Reason: reason}) + return nil +} + +func TestPublishEnqueuesLeverageDriftNotificationOnVersionBump(t *testing.T) { + db := newSyncExportOfferingTestDB(t) + + export := relational.Export{} + require.NoError(t, db.Create(&export).Error) + provided := relational.ProvidedControlImplementation{ExportId: *export.ID} + require.NoError(t, db.Create(&provided).Error) + + upstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&upstreamSSP).Error) + offering := relational.SSPExportOffering{SSPID: *upstreamSSP.ID, Title: "Offering", Status: relational.SSPExportOfferingStatusDraft} + require.NoError(t, db.Create(&offering).Error) + require.NoError(t, db.Create(&relational.SSPExportOfferingItem{ + OfferingID: *offering.ID, ControlID: "ac-1", ComponentUUID: uuid.New(), ProvidedUUID: *provided.ID, + }).Error) + + spy := &spyJobEnqueuer{} + h := NewSSPExportOfferingHandler(zap.NewNop().Sugar(), db, spy) + + // First publish: sets Version=1, no leverage links exist yet. + ctx, rec := newPublishRequestContext(offering.SSPID, *offering.ID) + require.NoError(t, h.Publish(ctx)) + require.Equal(t, http.StatusOK, rec.Code) + require.Empty(t, spy.calls) + + downstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&downstreamSSP).Error) + link := relational.SSPLeverageLink{ + DownstreamSSPID: *downstreamSSP.ID, UpstreamSSPID: *upstreamSSP.ID, OfferingID: *offering.ID, OfferingVersion: 1, + ControlID: "ac-1", ProvidedUUID: *provided.ID, InheritedUUID: uuid.New(), LeveragedAuthUUID: uuid.New(), + Satisfaction: relational.SSPLeverageSatisfactionFull, Status: relational.SSPLeverageStatusActive, + } + require.NoError(t, db.Create(&link).Error) + + // Change content, re-publish: Version bumps to 2, link drifts, notification enqueued. + require.NoError(t, db.Create(&relational.SSPExportOfferingItem{ + OfferingID: *offering.ID, ControlID: "ac-2", ComponentUUID: uuid.New(), ProvidedUUID: uuid.New(), + }).Error) + ctx2, rec2 := newPublishRequestContext(offering.SSPID, *offering.ID) + require.NoError(t, h.Publish(ctx2)) + require.Equal(t, http.StatusOK, rec2.Code) + + require.Len(t, spy.calls, 1) + require.Equal(t, *link.ID, spy.calls[0].LinkID) + require.Equal(t, link.DownstreamSSPID, spy.calls[0].DownstreamSSPID) +} + +func TestUpdateOfferingStatusEnqueuesLeverageDriftNotification(t *testing.T) { + db := newSSPLeverageTestDB(t) + offering, link := seedPublishedOfferingWithActiveLink(t, db) + spy := &spyJobEnqueuer{} + h := NewSSPExportOfferingHandler(zap.NewNop().Sugar(), db, spy) + + ctx, rec := newOfferingStatusRequestContext(offering.SSPID, *offering.ID, `{"status":"deprecated"}`) + require.NoError(t, h.UpdateOfferingStatus(ctx)) + require.Equal(t, http.StatusOK, rec.Code) + + require.Len(t, spy.calls, 1) + require.Equal(t, *link.ID, spy.calls[0].LinkID) + require.Equal(t, "upstream offering deprecated", spy.calls[0].Reason) +} + +func TestDeleteLeveragedAuthorizationEnqueuesLeverageDriftNotification(t *testing.T) { + db := newSSPLeverageTestDB(t) + require.NoError(t, db.AutoMigrate(&relational.SystemImplementation{}, &relational.LeveragedAuthorization{})) + + ssp := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&ssp).Error) + sysImpl := relational.SystemImplementation{SystemSecurityPlanId: *ssp.ID} + require.NoError(t, db.Create(&sysImpl).Error) + auth := relational.LeveragedAuthorization{Title: "Trust", PartyUUID: uuid.New(), SystemImplementationId: *sysImpl.ID} + require.NoError(t, db.Create(&auth).Error) + + upstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&upstreamSSP).Error) + downstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&downstreamSSP).Error) + export := relational.Export{} + require.NoError(t, db.Create(&export).Error) + provided := relational.ProvidedControlImplementation{ExportId: *export.ID} + require.NoError(t, db.Create(&provided).Error) + offering := relational.SSPExportOffering{SSPID: *upstreamSSP.ID, Title: "Offering", Status: relational.SSPExportOfferingStatusPublished, Version: 1} + require.NoError(t, db.Create(&offering).Error) + link := relational.SSPLeverageLink{ + DownstreamSSPID: *downstreamSSP.ID, UpstreamSSPID: *upstreamSSP.ID, OfferingID: *offering.ID, OfferingVersion: 1, + ControlID: "ac-1", ProvidedUUID: *provided.ID, InheritedUUID: uuid.New(), LeveragedAuthUUID: *auth.ID, + Satisfaction: relational.SSPLeverageSatisfactionFull, Status: relational.SSPLeverageStatusActive, + } + require.NoError(t, db.Create(&link).Error) + + spy := &spyJobEnqueuer{} + h := NewSystemSecurityPlanHandler(zap.NewNop().Sugar(), db, nil, spy) + ctx, rec := newDeleteLeveragedAuthRequestContext(*ssp.ID, *auth.ID) + require.NoError(t, h.DeleteSystemImplementationLeveragedAuthorization(ctx)) + require.Equal(t, http.StatusNoContent, rec.Code) + + require.Len(t, spy.calls, 1) + require.Equal(t, *link.ID, spy.calls[0].LinkID) + require.Equal(t, "leveraged authorization revoked", spy.calls[0].Reason) +} + +func TestUpdateOfferingStatusNoDriftEnqueuesNothing(t *testing.T) { + db := newSSPLeverageTestDB(t) + upstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&upstreamSSP).Error) + offering := relational.SSPExportOffering{SSPID: *upstreamSSP.ID, Title: "Offering", Status: relational.SSPExportOfferingStatusPublished, Version: 1} + require.NoError(t, db.Create(&offering).Error) + + spy := &spyJobEnqueuer{} + h := NewSSPExportOfferingHandler(zap.NewNop().Sugar(), db, spy) + + ctx, rec := newOfferingStatusRequestContext(offering.SSPID, *offering.ID, `{"status":"deprecated"}`) + require.NoError(t, h.UpdateOfferingStatus(ctx)) + require.Equal(t, http.StatusOK, rec.Code) + require.Empty(t, spy.calls) +} diff --git a/internal/api/handler/oscal/ssp_leverage_drift_integration_test.go b/internal/api/handler/oscal/ssp_leverage_drift_integration_test.go new file mode 100644 index 00000000..0209c795 --- /dev/null +++ b/internal/api/handler/oscal/ssp_leverage_drift_integration_test.go @@ -0,0 +1,256 @@ +//go:build integration + +package oscal + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/compliance-framework/api/internal/authz" + "github.com/compliance-framework/api/internal/service/relational" + "github.com/compliance-framework/api/internal/service/relational/risks" + "github.com/compliance-framework/api/internal/tests" + "github.com/google/uuid" + "github.com/labstack/echo/v4" + "github.com/stretchr/testify/suite" + "go.uber.org/zap" + "gorm.io/datatypes" +) + +// LeverageDriftIntegrationSuite exercises BCH-1341's full drift pipeline against real +// Postgres: the ticket's own Tests bullet (publish v1 -> subscribe -> downgrade a +// provided status -> re-sync -> link drifted + inherited-revoked risk + notification; +// re-attest -> cleared), plus the deprecate/revoke and leveraged-auth-lapse triggers, +// idempotency, and the AC #3 isolation requirement (no upstream risk/evidence exposed +// downstream). +type LeverageDriftIntegrationSuite struct { + tests.IntegrationTestSuite +} + +func TestLeverageDriftIntegrationSuite(t *testing.T) { + suite.Run(t, new(LeverageDriftIntegrationSuite)) +} + +func (suite *LeverageDriftIntegrationSuite) SetupTest() { + suite.Require().NoError(suite.Migrator.Refresh()) +} + +// driftFixture is the minimal real-schema graph every test in this suite needs: an +// upstream SSP with one published offering item backed by a real ByComponent (so the +// ImplementationStatus-downgrade trigger has something live to read), and a downstream +// SSP with the scaffolding Subscribe requires (ControlImplementation/SystemImplementation). +type driftFixture struct { + upstreamSSPID uuid.UUID + downstreamSSPID uuid.UUID + offeringID uuid.UUID + itemID uuid.UUID + byComponentID uuid.UUID + providedID uuid.UUID + respID uuid.UUID +} + +func seedDriftFixture(suite *LeverageDriftIntegrationSuite) driftFixture { + db := suite.DB + + upstreamSSP := relational.SystemSecurityPlan{} + suite.Require().NoError(db.Create(&upstreamSSP).Error) + + byComponent := relational.ByComponent{ + ImplementationStatus: datatypes.NewJSONType(relational.ImplementationStatus{State: relational.ImplementationStatusImplemented}), + } + suite.Require().NoError(db.Create(&byComponent).Error) + + export := relational.Export{ByComponentId: *byComponent.ID} + suite.Require().NoError(db.Create(&export).Error) + + provided := relational.ProvidedControlImplementation{ExportId: *export.ID, Description: "provided"} + suite.Require().NoError(db.Create(&provided).Error) + + resp := relational.ControlImplementationResponsibility{ExportId: *export.ID, ProvidedUuid: *provided.ID, Description: "responsibility"} + suite.Require().NoError(db.Create(&resp).Error) + + offering := relational.SSPExportOffering{SSPID: *upstreamSSP.ID, Title: "Offering", Status: relational.SSPExportOfferingStatusDraft} + suite.Require().NoError(db.Create(&offering).Error) + + item := relational.SSPExportOfferingItem{ + OfferingID: *offering.ID, ControlID: "ac-1", ComponentUUID: byComponent.ComponentUUID, ProvidedUUID: *provided.ID, + } + suite.Require().NoError(db.Create(&item).Error) + + downstreamSSP := relational.SystemSecurityPlan{} + suite.Require().NoError(db.Create(&downstreamSSP).Error) + suite.Require().NoError(db.Create(&relational.ControlImplementation{SystemSecurityPlanId: *downstreamSSP.ID}).Error) + suite.Require().NoError(db.Create(&relational.SystemImplementation{SystemSecurityPlanId: *downstreamSSP.ID}).Error) + + fx := driftFixture{ + upstreamSSPID: *upstreamSSP.ID, downstreamSSPID: *downstreamSSP.ID, + offeringID: *offering.ID, itemID: *item.ID, byComponentID: *byComponent.ID, + providedID: *provided.ID, respID: *resp.ID, + } + + // Publish v1 for real (rather than seeding Status=published/Version=1 directly) so + // ContentHash is genuinely established — the ticket's own flow starts with "publish + // v1", and skipping this step would make the *next* Publish call look like a content + // change (empty ContentHash != any computed hash) even when nothing actually changed. + publishResp := suite.publish(fx.upstreamSSPID, fx.offeringID, nil) + suite.Require().Equal(http.StatusOK, publishResp.Code) + + return fx +} + +func (suite *LeverageDriftIntegrationSuite) subscribe(fx driftFixture) relational.SSPLeverageLink { + pdp := &stubPDP{allow: true} + h := NewSSPLeverageHandler(zap.NewNop().Sugar(), suite.DB, pdp, authz.FailClosed) + body := subscribeBody(fx.downstreamSSPID, fx.itemID, fx.respID) + ctx, _, rec := newSubscribeRequestContext(fx.offeringID, body) + suite.Require().NoError(h.Subscribe(ctx)) + suite.Require().Equal(http.StatusCreated, rec.Code) + + var link relational.SSPLeverageLink + suite.Require().NoError(suite.DB.Where("downstream_ssp_id = ? AND offering_id = ?", fx.downstreamSSPID, fx.offeringID).First(&link).Error) + return link +} + +func (suite *LeverageDriftIntegrationSuite) publish(sspID, offeringID uuid.UUID, jobEnqueuer SSPJobEnqueuer) *httptest.ResponseRecorder { + h := NewSSPExportOfferingHandler(zap.NewNop().Sugar(), suite.DB, jobEnqueuer) + ctx, rec := newPublishRequestContext(sspID, offeringID) + suite.Require().NoError(h.Publish(ctx)) + return rec +} + +// TestFullDriftAndReAttestFlow is the ticket's own Tests bullet, end to end: publish v1 +// -> subscribe -> downgrade a provided status -> re-sync -> link drifted + +// inherited-revoked risk + notification; re-attest -> cleared. +func (suite *LeverageDriftIntegrationSuite) TestFullDriftAndReAttestFlow() { + fx := seedDriftFixture(suite) + link := suite.subscribe(fx) + suite.Equal(relational.SSPLeverageStatusActive, link.Status) + suite.Equal(1, link.OfferingVersion) + + spy := &spyJobEnqueuer{} + resp := suite.publish(fx.upstreamSSPID, fx.offeringID, spy) + suite.Equal(http.StatusOK, resp.Code) + suite.Empty(spy.calls, "re-publishing with no content change must not drift anything") + + // Downgrade the backing component's status, then re-sync (re-publish). + suite.Require().NoError(suite.DB.Model(&relational.ByComponent{}).Where("id = ?", fx.byComponentID).Update( + "implementation_status", datatypes.NewJSONType(relational.ImplementationStatus{State: relational.ImplementationStatusPlanned}), + ).Error) + + resp = suite.publish(fx.upstreamSSPID, fx.offeringID, spy) + suite.Equal(http.StatusOK, resp.Code) + + var reloadedLink relational.SSPLeverageLink + suite.Require().NoError(suite.DB.First(&reloadedLink, "id = ?", link.ID).Error) + suite.Equal(relational.SSPLeverageStatusDrifted, reloadedLink.Status) + + var driftRisk risks.Risk + suite.Require().NoError(suite.DB.Where("ssp_id = ? AND source_type = ?", fx.downstreamSSPID, string(risks.RiskSourceTypeInheritedRevoked)).First(&driftRisk).Error) + suite.Equal(string(risks.RiskStatusOpen), driftRisk.Status) + + var responsibilityLink risks.RiskResponsibilityLink + suite.Require().NoError(suite.DB.Where("risk_id = ? AND responsibility_uuid = ?", driftRisk.ID, fx.respID).First(&responsibilityLink).Error) + + suite.Require().Len(spy.calls, 1) + suite.Equal(*driftRisk.ID, spy.calls[0].RiskID) + suite.Equal(*reloadedLink.ID, spy.calls[0].LinkID) + + // Re-run the same re-publish (idempotency): no duplicate risk, no re-flip, no second + // notification, since the link is already drifted and its Version hasn't moved again. + spy.calls = nil + resp = suite.publish(fx.upstreamSSPID, fx.offeringID, spy) + suite.Equal(http.StatusOK, resp.Code) + suite.Empty(spy.calls) + var riskCount int64 + suite.Require().NoError(suite.DB.Model(&risks.Risk{}).Where("source_type = ?", string(risks.RiskSourceTypeInheritedRevoked)).Count(&riskCount).Error) + suite.Equal(int64(1), riskCount) + + // Re-attest: clears drift and remediates the risk. + leverageHandler := NewSSPLeverageHandler(zap.NewNop().Sugar(), suite.DB, &stubPDP{allow: true}, authz.FailClosed) + attestCtx, attestRec := newReAttestRequestContext(fx.downstreamSSPID, *reloadedLink.ID) + suite.Require().NoError(leverageHandler.ReAttest(attestCtx)) + suite.Equal(http.StatusOK, attestRec.Code) + + var attestedLink relational.SSPLeverageLink + suite.Require().NoError(suite.DB.First(&attestedLink, "id = ?", link.ID).Error) + suite.Equal(relational.SSPLeverageStatusActive, attestedLink.Status) + suite.Equal(2, attestedLink.OfferingVersion) + + var remediatedRisk risks.Risk + suite.Require().NoError(suite.DB.First(&remediatedRisk, "id = ?", driftRisk.ID).Error) + suite.Equal(string(risks.RiskStatusRemediated), remediatedRisk.Status) + + // Isolation (AC #3): no risk was ever created for the upstream SSP, and the + // downstream-facing projection response has no risk/evidence fields at all. + var upstreamRiskCount int64 + suite.Require().NoError(suite.DB.Model(&risks.Risk{}).Where("ssp_id = ?", fx.upstreamSSPID).Count(&upstreamRiskCount).Error) + suite.Zero(upstreamRiskCount) + + suite.assertProjectionExposesNoUpstreamRiskOrEvidence(fx.downstreamSSPID) +} + +// TestDeprecateOfferingDriftsAndNotifies covers the offering-deprecated trigger, +// independent of any Version bump. +func (suite *LeverageDriftIntegrationSuite) TestDeprecateOfferingDriftsAndNotifies() { + fx := seedDriftFixture(suite) + link := suite.subscribe(fx) + + spy := &spyJobEnqueuer{} + h := NewSSPExportOfferingHandler(zap.NewNop().Sugar(), suite.DB, spy) + ctx, rec := newOfferingStatusRequestContext(fx.upstreamSSPID, fx.offeringID, `{"status":"deprecated"}`) + suite.Require().NoError(h.UpdateOfferingStatus(ctx)) + suite.Equal(http.StatusOK, rec.Code) + + var reloadedLink relational.SSPLeverageLink + suite.Require().NoError(suite.DB.First(&reloadedLink, "id = ?", link.ID).Error) + suite.Equal(relational.SSPLeverageStatusDrifted, reloadedLink.Status) + suite.Equal(1, reloadedLink.OfferingVersion, "deprecating must not touch OfferingVersion") + + suite.Require().Len(spy.calls, 1) + suite.Equal("upstream offering deprecated", spy.calls[0].Reason) +} + +// TestLeveragedAuthorizationDeleteDriftsAndNotifies covers the "leveraged authorization +// lapsed" trigger. +func (suite *LeverageDriftIntegrationSuite) TestLeveragedAuthorizationDeleteDriftsAndNotifies() { + fx := seedDriftFixture(suite) + link := suite.subscribe(fx) + + spy := &spyJobEnqueuer{} + sspHandler := NewSystemSecurityPlanHandler(zap.NewNop().Sugar(), suite.DB, nil, spy) + ctx, rec := newDeleteLeveragedAuthRequestContext(fx.downstreamSSPID, link.LeveragedAuthUUID) + suite.Require().NoError(sspHandler.DeleteSystemImplementationLeveragedAuthorization(ctx)) + suite.Equal(http.StatusNoContent, rec.Code) + + var reloadedLink relational.SSPLeverageLink + suite.Require().NoError(suite.DB.First(&reloadedLink, "id = ?", link.ID).Error) + suite.Equal(relational.SSPLeverageStatusDrifted, reloadedLink.Status) + + suite.Require().Len(spy.calls, 1) + suite.Equal("leveraged authorization revoked", spy.calls[0].Reason) +} + +// assertProjectionExposesNoUpstreamRiskOrEvidence calls the existing (unchanged) +// LeveragedControls projection endpoint and confirms its JSON shape has no risk- or +// evidence-related keys at any level — AC #3's "never exposes the upstream's risk +// register" checked at the actual response shape, not just DB state. +func (suite *LeverageDriftIntegrationSuite) assertProjectionExposesNoUpstreamRiskOrEvidence(downstreamSSPID uuid.UUID) { + h := NewSSPLeverageHandler(zap.NewNop().Sugar(), suite.DB, &stubPDP{allow: true}, authz.FailClosed) + e := echo.New() + req := httptest.NewRequest(http.MethodGet, "/", nil) + rec := httptest.NewRecorder() + ctx := e.NewContext(req, rec) + ctx.SetParamNames("id") + ctx.SetParamValues(downstreamSSPID.String()) + suite.Require().NoError(h.LeveragedControls(ctx)) + suite.Equal(http.StatusOK, rec.Code) + + raw := rec.Body.String() + suite.NotContains(raw, `"risk`) + suite.NotContains(raw, `"evidence`) + + var parsed map[string]any + suite.Require().NoError(json.Unmarshal([]byte(raw), &parsed)) +} diff --git a/internal/api/handler/oscal/ssp_leverage_drift_test.go b/internal/api/handler/oscal/ssp_leverage_drift_test.go new file mode 100644 index 00000000..bfc45f4e --- /dev/null +++ b/internal/api/handler/oscal/ssp_leverage_drift_test.go @@ -0,0 +1,222 @@ +package oscal + +import ( + "testing" + + "github.com/compliance-framework/api/internal/service/relational" + "github.com/compliance-framework/api/internal/service/relational/risks" + "github.com/google/uuid" + "github.com/stretchr/testify/require" + "gorm.io/gorm" +) + +// seedLeverageLinkForDrift creates the minimal graph applyDriftToLink/ +// evaluateLeverageDriftForOffering need: an Export with one +// ProvidedControlImplementation and one ControlImplementationResponsibility under it, +// an upstream+downstream SSP, an SSPExportOffering, and an active SSPLeverageLink tying +// them together at offeringVersion. +func seedLeverageLinkForDrift(t *testing.T, db *gorm.DB, offeringVersion int) (*relational.SSPExportOffering, *relational.SSPLeverageLink, uuid.UUID) { + t.Helper() + + export := relational.Export{} + require.NoError(t, db.Create(&export).Error) + + provided := relational.ProvidedControlImplementation{ExportId: *export.ID} + require.NoError(t, db.Create(&provided).Error) + + responsibility := relational.ControlImplementationResponsibility{ExportId: *export.ID, ProvidedUuid: *provided.ID} + require.NoError(t, db.Create(&responsibility).Error) + + upstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&upstreamSSP).Error) + downstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&downstreamSSP).Error) + + offering := relational.SSPExportOffering{ + SSPID: *upstreamSSP.ID, Title: "Offering", Status: relational.SSPExportOfferingStatusPublished, Version: offeringVersion, + } + require.NoError(t, db.Create(&offering).Error) + + link := relational.SSPLeverageLink{ + DownstreamSSPID: *downstreamSSP.ID, + UpstreamSSPID: *upstreamSSP.ID, + OfferingID: *offering.ID, + OfferingVersion: 1, + ControlID: "ac-1", + ProvidedUUID: *provided.ID, + InheritedUUID: uuid.New(), + LeveragedAuthUUID: uuid.New(), + Satisfaction: relational.SSPLeverageSatisfactionFull, + Status: relational.SSPLeverageStatusActive, + } + require.NoError(t, db.Create(&link).Error) + + return &offering, &link, *responsibility.ID +} + +func TestApplyDriftToLinkCreatesRiskAndFlipsStatus(t *testing.T) { + db := newSSPLeverageTestDB(t) + _, link, responsibilityUUID := seedLeverageLinkForDrift(t, db, 2) + + info, ok, err := applyDriftToLink(db, link, "upstream offering content changed") + require.NoError(t, err) + require.True(t, ok) + require.Equal(t, *link.ID, info.LinkID) + require.Equal(t, link.DownstreamSSPID, info.DownstreamSSPID) + + var reloadedLink relational.SSPLeverageLink + require.NoError(t, db.First(&reloadedLink, "id = ?", link.ID).Error) + require.Equal(t, relational.SSPLeverageStatusDrifted, reloadedLink.Status) + + var risk risks.Risk + require.NoError(t, db.First(&risk, "id = ?", info.RiskID).Error) + require.Equal(t, string(risks.RiskSourceTypeInheritedRevoked), risk.SourceType) + require.Equal(t, string(risks.RiskStatusOpen), risk.Status) + require.Equal(t, link.DownstreamSSPID, risk.SSPID) + + var responsibilityLink risks.RiskResponsibilityLink + require.NoError(t, db.Where("risk_id = ? AND responsibility_uuid = ?", risk.ID, responsibilityUUID).First(&responsibilityLink).Error) + + var riskCount int64 + require.NoError(t, db.Model(&risks.Risk{}).Count(&riskCount).Error) + require.Equal(t, int64(1), riskCount) +} + +func TestApplyDriftToLinkIsNoOpWhenAlreadyDrifted(t *testing.T) { + db := newSSPLeverageTestDB(t) + _, link, _ := seedLeverageLinkForDrift(t, db, 2) + + _, ok, err := applyDriftToLink(db, link, "first drift") + require.NoError(t, err) + require.True(t, ok) + + // link is now Drifted (in-memory struct updated by applyDriftToLink); calling again + // on the same struct must be a no-op — no second risk, no re-flip. + info, ok, err := applyDriftToLink(db, link, "second drift") + require.NoError(t, err) + require.False(t, ok) + require.Zero(t, info) + + var riskCount int64 + require.NoError(t, db.Model(&risks.Risk{}).Count(&riskCount).Error) + require.Equal(t, int64(1), riskCount) +} + +func TestApplyDriftToLinkReopensRemediatedRisk(t *testing.T) { + db := newSSPLeverageTestDB(t) + _, link, _ := seedLeverageLinkForDrift(t, db, 2) + + info, _, err := applyDriftToLink(db, link, "first drift") + require.NoError(t, err) + + // Simulate re-attestation clearing the risk and reactivating the link. + require.NoError(t, db.Model(&risks.Risk{}).Where("id = ?", info.RiskID).Update("status", string(risks.RiskStatusRemediated)).Error) + require.NoError(t, db.Model(&relational.SSPLeverageLink{}).Where("id = ?", link.ID).Update("status", relational.SSPLeverageStatusActive).Error) + link.Status = relational.SSPLeverageStatusActive + + reInfo, ok, err := applyDriftToLink(db, link, "drifted again") + require.NoError(t, err) + require.True(t, ok) + require.Equal(t, info.RiskID, reInfo.RiskID) + + var risk risks.Risk + require.NoError(t, db.First(&risk, "id = ?", info.RiskID).Error) + require.Equal(t, string(risks.RiskStatusOpen), risk.Status) + + var riskCount int64 + require.NoError(t, db.Model(&risks.Risk{}).Count(&riskCount).Error) + require.Equal(t, int64(1), riskCount) +} + +func TestEvaluateLeverageDriftForOfferingOnlyDriftsBehindLinks(t *testing.T) { + db := newSSPLeverageTestDB(t) + offering, link, _ := seedLeverageLinkForDrift(t, db, 1) + // link.OfferingVersion == 1, offering.Version == 1: up to date, must not drift. + + results, err := evaluateLeverageDriftForOffering(db, *offering) + require.NoError(t, err) + require.Empty(t, results) + + var reloadedLink relational.SSPLeverageLink + require.NoError(t, db.First(&reloadedLink, "id = ?", link.ID).Error) + require.Equal(t, relational.SSPLeverageStatusActive, reloadedLink.Status) +} + +func TestEvaluateLeverageDriftForOfferingDriftsOnVersionBump(t *testing.T) { + db := newSSPLeverageTestDB(t) + offering, link, _ := seedLeverageLinkForDrift(t, db, 2) + // link.OfferingVersion == 1, offering.Version == 2: behind, must drift. + + results, err := evaluateLeverageDriftForOffering(db, *offering) + require.NoError(t, err) + require.Len(t, results, 1) + require.Equal(t, *link.ID, results[0].LinkID) +} + +func TestEvaluateLeverageDriftForOfferingDriftsOnDeprecated(t *testing.T) { + db := newSSPLeverageTestDB(t) + offering, link, _ := seedLeverageLinkForDrift(t, db, 1) + offering.Status = relational.SSPExportOfferingStatusDeprecated + + results, err := evaluateLeverageDriftForOffering(db, *offering) + require.NoError(t, err) + require.Len(t, results, 1) + require.Equal(t, *link.ID, results[0].LinkID) +} + +func TestResolveCatalogIDForControlResolvesFromDownstreamProfile(t *testing.T) { + db := newSSPLeverageTestDB(t) + downstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&downstreamSSP).Error) + + catalogID := uuid.New() + control := relational.Control{CatalogID: catalogID, ID: "ac-1", Title: "Access Control"} + require.NoError(t, db.Create(&control).Error) + + profile := relational.Profile{Controls: []relational.Control{control}} + require.NoError(t, db.Create(&profile).Error) + + require.NoError(t, db.Create(&relational.SSPProfile{ + SystemSecurityPlanID: *downstreamSSP.ID, ProfileID: *profile.ID, + }).Error) + + resolved, ok, err := resolveCatalogIDForControl(db, *downstreamSSP.ID, "AC-1") + require.NoError(t, err) + require.True(t, ok) + require.Equal(t, catalogID, resolved) +} + +func TestResolveCatalogIDForControlMissReturnsNotOK(t *testing.T) { + db := newSSPLeverageTestDB(t) + downstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&downstreamSSP).Error) + + _, ok, err := resolveCatalogIDForControl(db, *downstreamSSP.ID, "ac-1") + require.NoError(t, err) + require.False(t, ok) +} + +// TestSameLinkDriftedByTwoTriggersInQuickSuccessionDoesNotDuplicateRisk: a link's dedupe +// key is per-link, not per-trigger — if a version-bump drift and a deprecate/revoke +// drift both land on the same link (e.g. a version bump processed just before an +// operator deprecates the offering), the second call must be a no-op rather than +// creating a second risk, since applyDriftToLink already skips non-active links. +func TestSameLinkDriftedByTwoTriggersInQuickSuccessionDoesNotDuplicateRisk(t *testing.T) { + db := newSSPLeverageTestDB(t) + _, link, _ := seedLeverageLinkForDrift(t, db, 2) + + _, ok, err := applyDriftToLink(db, link, "upstream offering content changed") + require.NoError(t, err) + require.True(t, ok) + + // A second, differently-reasoned trigger arriving right after (e.g. deprecate) finds + // the link already Drifted and must be a no-op. + info, ok, err := applyDriftToLink(db, link, "upstream offering deprecated") + require.NoError(t, err) + require.False(t, ok) + require.Zero(t, info) + + var riskCount int64 + require.NoError(t, db.Model(&risks.Risk{}).Where("source_type = ?", string(risks.RiskSourceTypeInheritedRevoked)).Count(&riskCount).Error) + require.Equal(t, int64(1), riskCount) +} diff --git a/internal/api/handler/oscal/ssp_leverage_reattest_test.go b/internal/api/handler/oscal/ssp_leverage_reattest_test.go new file mode 100644 index 00000000..1531dd8b --- /dev/null +++ b/internal/api/handler/oscal/ssp_leverage_reattest_test.go @@ -0,0 +1,108 @@ +package oscal + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/compliance-framework/api/internal/authz" + "github.com/compliance-framework/api/internal/service/relational" + "github.com/compliance-framework/api/internal/service/relational/risks" + "github.com/google/uuid" + "github.com/labstack/echo/v4" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + "gorm.io/gorm" +) + +func newReAttestRequestContext(sspID, linkID uuid.UUID) (echo.Context, *httptest.ResponseRecorder) { + e := echo.New() + req := httptest.NewRequest(http.MethodPost, "/", nil) + rec := httptest.NewRecorder() + ctx := e.NewContext(req, rec) + ctx.SetParamNames("id", "linkId") + ctx.SetParamValues(sspID.String(), linkID.String()) + return ctx, rec +} + +// subscribeAndDrift subscribes fx's downstream SSP to fx's offering item (satisfying +// respAID), then simulates an upstream content change bumping the offering's Version and +// drifts the resulting link — returning the now-drifted link and its drift risk id. +func subscribeAndDrift(t *testing.T, db *gorm.DB, fx leverageFixture) (relational.SSPLeverageLink, uuid.UUID) { + t.Helper() + + pdp := &stubPDP{allow: true} + subscribeHandler := NewSSPLeverageHandler(zap.NewNop().Sugar(), db, pdp, authz.FailClosed) + body := subscribeBody(fx.downstreamSSPID, fx.itemID, fx.respAID) + ctx, _, rec := newSubscribeRequestContext(fx.offeringID, body) + require.NoError(t, subscribeHandler.Subscribe(ctx)) + require.Equal(t, http.StatusCreated, rec.Code) + + var link relational.SSPLeverageLink + require.NoError(t, db.Where("downstream_ssp_id = ?", fx.downstreamSSPID).First(&link).Error) + + require.NoError(t, db.Model(&relational.SSPExportOffering{}).Where("id = ?", fx.offeringID).Update("version", 2).Error) + var offering relational.SSPExportOffering + require.NoError(t, db.First(&offering, "id = ?", fx.offeringID).Error) + + info, ok, err := applyDriftToLink(db, &link, "upstream offering content changed") + require.NoError(t, err) + require.True(t, ok) + + require.NoError(t, db.First(&link, "id = ?", link.ID).Error) + return link, info.RiskID +} + +func TestReAttestClearsDriftAndRemediatesRisk(t *testing.T) { + db := newSSPLeverageTestDB(t) + fx := newLeverageFixture(t, db) + link, riskID := subscribeAndDrift(t, db, fx) + require.Equal(t, relational.SSPLeverageStatusDrifted, link.Status) + + h := NewSSPLeverageHandler(zap.NewNop().Sugar(), db, &stubPDP{allow: true}, authz.FailClosed) + ctx, rec := newReAttestRequestContext(fx.downstreamSSPID, *link.ID) + require.NoError(t, h.ReAttest(ctx)) + require.Equal(t, http.StatusOK, rec.Code) + + var reloadedLink relational.SSPLeverageLink + require.NoError(t, db.First(&reloadedLink, "id = ?", link.ID).Error) + require.Equal(t, relational.SSPLeverageStatusActive, reloadedLink.Status) + require.Equal(t, 2, reloadedLink.OfferingVersion) + require.NotNil(t, reloadedLink.AttestedAt) + + var risk risks.Risk + require.NoError(t, db.First(&risk, "id = ?", riskID).Error) + require.Equal(t, string(risks.RiskStatusRemediated), risk.Status) +} + +func TestReAttestRejectsNonDriftedLink(t *testing.T) { + db := newSSPLeverageTestDB(t) + fx := newLeverageFixture(t, db) + + pdp := &stubPDP{allow: true} + subscribeHandler := NewSSPLeverageHandler(zap.NewNop().Sugar(), db, pdp, authz.FailClosed) + body := subscribeBody(fx.downstreamSSPID, fx.itemID, fx.respAID) + ctx, _, rec := newSubscribeRequestContext(fx.offeringID, body) + require.NoError(t, subscribeHandler.Subscribe(ctx)) + require.Equal(t, http.StatusCreated, rec.Code) + + var link relational.SSPLeverageLink + require.NoError(t, db.Where("downstream_ssp_id = ?", fx.downstreamSSPID).First(&link).Error) + require.Equal(t, relational.SSPLeverageStatusActive, link.Status) + + h := NewSSPLeverageHandler(zap.NewNop().Sugar(), db, &stubPDP{allow: true}, authz.FailClosed) + attestCtx, attestRec := newReAttestRequestContext(fx.downstreamSSPID, *link.ID) + require.NoError(t, h.ReAttest(attestCtx)) + require.Equal(t, http.StatusBadRequest, attestRec.Code) +} + +func TestReAttestRejectsWrongDownstreamSSP(t *testing.T) { + db := newSSPLeverageTestDB(t) + fx := newLeverageFixture(t, db) + link, _ := subscribeAndDrift(t, db, fx) + + h := NewSSPLeverageHandler(zap.NewNop().Sugar(), db, &stubPDP{allow: true}, authz.FailClosed) + ctx, rec := newReAttestRequestContext(uuid.New(), *link.ID) + require.NoError(t, h.ReAttest(ctx)) + require.Equal(t, http.StatusNotFound, rec.Code) +} diff --git a/internal/api/handler/oscal/ssp_leverage_unit_test.go b/internal/api/handler/oscal/ssp_leverage_unit_test.go index e9489322..74121a38 100644 --- a/internal/api/handler/oscal/ssp_leverage_unit_test.go +++ b/internal/api/handler/oscal/ssp_leverage_unit_test.go @@ -13,6 +13,7 @@ import ( "github.com/compliance-framework/api/internal/authn" "github.com/compliance-framework/api/internal/authz" "github.com/compliance-framework/api/internal/service/relational" + "github.com/compliance-framework/api/internal/service/relational/risks" "github.com/google/uuid" "github.com/labstack/echo/v4" "github.com/stretchr/testify/require" @@ -43,6 +44,14 @@ func newSSPLeverageTestDB(t *testing.T) *gorm.DB { &relational.LeveragedAuthorization{}, &relational.Filter{}, &relational.FilterResponsibility{}, + &relational.Profile{}, + &relational.Control{}, + &relational.SSPProfile{}, + &risks.Risk{}, + &risks.RiskEvent{}, + &risks.RiskScore{}, + &risks.RiskResponsibilityLink{}, + &risks.RiskControlLink{}, )) return db } diff --git a/internal/api/handler/oscal/system_security_plans.go b/internal/api/handler/oscal/system_security_plans.go index cc553532..5beb6d3a 100644 --- a/internal/api/handler/oscal/system_security_plans.go +++ b/internal/api/handler/oscal/system_security_plans.go @@ -34,6 +34,7 @@ import ( type SSPJobEnqueuer interface { EnqueueOrphanedRiskCleanup(ctx context.Context, sspID uuid.UUID, oldProfileID, newProfileID *uuid.UUID) error EnqueueDashboardSuggestionCells(ctx context.Context, runID uuid.UUID, cellCount int) error + EnqueueLeverageDriftNotification(ctx context.Context, riskID, linkID, downstreamSSPID uuid.UUID, reason string) error } // profileSummary is a lightweight DTO returned by the multi-profile list endpoint. @@ -3475,13 +3476,46 @@ func (h *SystemSecurityPlanHandler) DeleteSystemImplementationLeveragedAuthoriza return ctx.JSON(http.StatusNotFound, api.NewError(err)) } - result := h.db.Where("id = ? AND system_implementation_id = ?", authID, *systemImpl.ID).Delete(&relational.LeveragedAuthorization{}) - if result.Error != nil { - h.sugar.Errorf("Failed to delete leveraged authorization: %v", result.Error) - return ctx.JSON(http.StatusInternalServerError, api.NewError(result.Error)) + var driftedLinks []driftedLinkInfo + var rowsAffected int64 + if err := h.db.Transaction(func(tx *gorm.DB) error { + result := tx.Where("id = ? AND system_implementation_id = ?", authID, *systemImpl.ID).Delete(&relational.LeveragedAuthorization{}) + if result.Error != nil { + return result.Error + } + rowsAffected = result.RowsAffected + if rowsAffected == 0 { + return nil + } + + // The leveraged authorization backing these links is gone — treat as "lapsed" + // (BCH-1341): drift every active leverage link that referenced it, independent of + // any offering version/status. + var links []relational.SSPLeverageLink + if err := tx.Where("leveraged_auth_uuid = ? AND status = ?", authID, relational.SSPLeverageStatusActive).Find(&links).Error; err != nil { + return fmt.Errorf("failed to load leverage links for lapsed authorization: %w", err) + } + for i := range links { + info, ok, err := applyDriftToLink(tx, &links[i], "leveraged authorization revoked") + if err != nil { + return err + } + if ok { + driftedLinks = append(driftedLinks, info) + } + } + return nil + }); err != nil { + h.sugar.Errorf("Failed to delete leveraged authorization: %v", err) + return ctx.JSON(http.StatusInternalServerError, api.NewError(err)) + } + if len(driftedLinks) > 0 { + enqueueCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx.Request().Context()), 10*time.Second) + defer cancel() + enqueueLeverageDriftNotifications(enqueueCtx, h.sugar, h.jobEnqueuer, driftedLinks) } - if result.RowsAffected == 0 { + if rowsAffected == 0 { return ctx.JSON(http.StatusNotFound, api.NewError(fmt.Errorf("leveraged authorization not found"))) } diff --git a/internal/api/handler/oscal/system_security_plans_leveraged_auth_drift_test.go b/internal/api/handler/oscal/system_security_plans_leveraged_auth_drift_test.go new file mode 100644 index 00000000..72a3d63f --- /dev/null +++ b/internal/api/handler/oscal/system_security_plans_leveraged_auth_drift_test.go @@ -0,0 +1,116 @@ +package oscal + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/compliance-framework/api/internal/service/relational" + "github.com/compliance-framework/api/internal/service/relational/risks" + "github.com/google/uuid" + "github.com/labstack/echo/v4" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +func newDeleteLeveragedAuthRequestContext(sspID, authID uuid.UUID) (echo.Context, *httptest.ResponseRecorder) { + e := echo.New() + req := httptest.NewRequest(http.MethodDelete, "/", nil) + rec := httptest.NewRecorder() + ctx := e.NewContext(req, rec) + ctx.SetParamNames("id", "authId") + ctx.SetParamValues(sspID.String(), authID.String()) + return ctx, rec +} + +// TestDeleteLeveragedAuthorizationDriftsReferencingActiveLinks: deleting a +// LeveragedAuthorization referenced by an active SSPLeverageLink is treated as the +// authorization having "lapsed" (BCH-1341) — the link drifts and gets its +// inherited-revoked risk, in the same transaction as the delete. +func TestDeleteLeveragedAuthorizationDriftsReferencingActiveLinks(t *testing.T) { + db := newSSPLeverageTestDB(t) + require.NoError(t, db.AutoMigrate(&relational.SystemImplementation{}, &relational.LeveragedAuthorization{})) + + ssp := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&ssp).Error) + sysImpl := relational.SystemImplementation{SystemSecurityPlanId: *ssp.ID} + require.NoError(t, db.Create(&sysImpl).Error) + auth := relational.LeveragedAuthorization{Title: "Trust", PartyUUID: uuid.New(), SystemImplementationId: *sysImpl.ID} + require.NoError(t, db.Create(&auth).Error) + + upstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&upstreamSSP).Error) + downstreamSSP := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&downstreamSSP).Error) + export := relational.Export{} + require.NoError(t, db.Create(&export).Error) + provided := relational.ProvidedControlImplementation{ExportId: *export.ID} + require.NoError(t, db.Create(&provided).Error) + offering := relational.SSPExportOffering{SSPID: *upstreamSSP.ID, Title: "Offering", Status: relational.SSPExportOfferingStatusPublished, Version: 1} + require.NoError(t, db.Create(&offering).Error) + + link := relational.SSPLeverageLink{ + DownstreamSSPID: *downstreamSSP.ID, UpstreamSSPID: *upstreamSSP.ID, OfferingID: *offering.ID, OfferingVersion: 1, + ControlID: "ac-1", ProvidedUUID: *provided.ID, InheritedUUID: uuid.New(), LeveragedAuthUUID: *auth.ID, + Satisfaction: relational.SSPLeverageSatisfactionFull, Status: relational.SSPLeverageStatusActive, + } + require.NoError(t, db.Create(&link).Error) + + h := NewSystemSecurityPlanHandler(zap.NewNop().Sugar(), db, nil, nil) + ctx, rec := newDeleteLeveragedAuthRequestContext(*ssp.ID, *auth.ID) + require.NoError(t, h.DeleteSystemImplementationLeveragedAuthorization(ctx)) + require.Equal(t, http.StatusNoContent, rec.Code) + + var authCount int64 + require.NoError(t, db.Model(&relational.LeveragedAuthorization{}).Count(&authCount).Error) + require.Zero(t, authCount) + + var reloadedLink relational.SSPLeverageLink + require.NoError(t, db.First(&reloadedLink, "id = ?", link.ID).Error) + require.Equal(t, relational.SSPLeverageStatusDrifted, reloadedLink.Status) + + var riskCount int64 + require.NoError(t, db.Model(&risks.Risk{}).Where("source_type = ?", string(risks.RiskSourceTypeInheritedRevoked)).Count(&riskCount).Error) + require.Equal(t, int64(1), riskCount) +} + +// TestDeleteLeveragedAuthorizationNotReferencedIsPureNoOp: deleting an authorization no +// leverage link references behaves exactly as before — no drift, no error. +func TestDeleteLeveragedAuthorizationNotReferencedIsPureNoOp(t *testing.T) { + db := newSSPLeverageTestDB(t) + require.NoError(t, db.AutoMigrate(&relational.SystemImplementation{}, &relational.LeveragedAuthorization{})) + + ssp := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&ssp).Error) + sysImpl := relational.SystemImplementation{SystemSecurityPlanId: *ssp.ID} + require.NoError(t, db.Create(&sysImpl).Error) + auth := relational.LeveragedAuthorization{Title: "Trust", PartyUUID: uuid.New(), SystemImplementationId: *sysImpl.ID} + require.NoError(t, db.Create(&auth).Error) + + h := NewSystemSecurityPlanHandler(zap.NewNop().Sugar(), db, nil, nil) + ctx, rec := newDeleteLeveragedAuthRequestContext(*ssp.ID, *auth.ID) + require.NoError(t, h.DeleteSystemImplementationLeveragedAuthorization(ctx)) + require.Equal(t, http.StatusNoContent, rec.Code) + + var riskCount int64 + require.NoError(t, db.Model(&risks.Risk{}).Count(&riskCount).Error) + require.Zero(t, riskCount) +} + +// TestDeleteLeveragedAuthorizationNotFoundReturns404: regression check — deleting a +// nonexistent authorization still 404s (unaffected by wrapping the delete in a +// transaction). +func TestDeleteLeveragedAuthorizationNotFoundReturns404(t *testing.T) { + db := newSSPLeverageTestDB(t) + require.NoError(t, db.AutoMigrate(&relational.SystemImplementation{}, &relational.LeveragedAuthorization{})) + + ssp := relational.SystemSecurityPlan{} + require.NoError(t, db.Create(&ssp).Error) + sysImpl := relational.SystemImplementation{SystemSecurityPlanId: *ssp.ID} + require.NoError(t, db.Create(&sysImpl).Error) + + h := NewSystemSecurityPlanHandler(zap.NewNop().Sugar(), db, nil, nil) + ctx, rec := newDeleteLeveragedAuthRequestContext(*ssp.ID, uuid.New()) + require.NoError(t, h.DeleteSystemImplementationLeveragedAuthorization(ctx)) + require.Equal(t, http.StatusNotFound, rec.Code) +} diff --git a/internal/authz/manifest.yaml b/internal/authz/manifest.yaml index 812d283d..7e2a201a 100644 --- a/internal/authz/manifest.yaml +++ b/internal/authz/manifest.yaml @@ -143,7 +143,7 @@ resources: ssp_id: uuid # C0 on /system-security-plans/:sspId/risks, else C1 owned_by: set # PrimaryOwnerUserID — matches subject.user_uuid status: string - source_type: string # manual | evidence-auto | oscal-import | inherited-responsibility + source_type: string # manual | evidence-auto | oscal-import | inherited-responsibility | inherited-revoked poam_item: actions: [read, create, update, delete] attributes: diff --git a/internal/service/relational/risks/models.go b/internal/service/relational/risks/models.go index 1926219b..727fa379 100644 --- a/internal/service/relational/risks/models.go +++ b/internal/service/relational/risks/models.go @@ -93,11 +93,12 @@ const ( RiskSourceTypeEvidenceAuto RiskSourceType = "evidence-auto" RiskSourceTypeOscalImport RiskSourceType = "oscal-import" RiskSourceTypeInheritedResponsibility RiskSourceType = "inherited-responsibility" + RiskSourceTypeInheritedRevoked RiskSourceType = "inherited-revoked" ) func (s RiskSourceType) IsValid() bool { switch s { - case RiskSourceTypeManual, RiskSourceTypeEvidenceAuto, RiskSourceTypeOscalImport, RiskSourceTypeInheritedResponsibility: + case RiskSourceTypeManual, RiskSourceTypeEvidenceAuto, RiskSourceTypeOscalImport, RiskSourceTypeInheritedResponsibility, RiskSourceTypeInheritedRevoked: return true default: return false diff --git a/internal/service/worker/jobs.go b/internal/service/worker/jobs.go index 740c6dd3..180220b7 100644 --- a/internal/service/worker/jobs.go +++ b/internal/service/worker/jobs.go @@ -701,6 +701,9 @@ func Workers( riskOpenDigestWorker := NewRiskOpenDigestWorker(db, userRepo, webBaseURL, riskNotificationServiceFactory, logger) river.AddWorker(workers, river.WorkFunc(riskOpenDigestWorker.Work)) + leverageDriftNotificationWorker := NewLeverageDriftNotificationWorker(db, userRepo, webBaseURL, riskNotificationServiceFactory, logger) + river.AddWorker(workers, river.WorkFunc(leverageDriftNotificationWorker.Work)) + // Register POAM notification workers (BCH-1186 Phase 3) poamDeadlineReminderWorker := NewPoamDeadlineReminderWorker( userRepo, diff --git a/internal/service/worker/leverage_notifications.go b/internal/service/worker/leverage_notifications.go new file mode 100644 index 00000000..a1d2e82c --- /dev/null +++ b/internal/service/worker/leverage_notifications.go @@ -0,0 +1,191 @@ +package worker + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/compliance-framework/api/internal/service/notification" + emailprovider "github.com/compliance-framework/api/internal/service/notification/providers/email" + "github.com/google/uuid" + "github.com/riverqueue/river" + "go.uber.org/zap" + "gorm.io/gorm" +) + +// JobTypeLeverageDriftNotification is the single River job kind for both +// leverage_drifted and leverage_revoked notifications (BCH-1341) — which of the two +// notification.Kind values actually gets dispatched is decided at Work() time by +// notificationKindForReason, since a drift risk (risks.Risk with SourceType +// inherited-revoked) is the same underlying entity regardless of which trigger produced +// it. +const JobTypeLeverageDriftNotification = "leverage_drift_notification" + +const ( + leverageDriftedNotificationKind = notification.Kind("leverage_drifted") + leverageRevokedNotificationKind = notification.Kind("leverage_revoked") +) + +// LeverageDriftNotificationArgs is the River job payload for one (risk, owner) pair, +// enqueued right after applyDriftToLink creates or reopens the drift risk. +type LeverageDriftNotificationArgs struct { + Channel string `json:"channel,omitempty"` + RiskID uuid.UUID `json:"risk_id"` + LinkID uuid.UUID `json:"link_id"` + OwnerUserID uuid.UUID `json:"owner_user_id"` + Reason string `json:"reason,omitempty"` +} + +func (LeverageDriftNotificationArgs) Kind() string { return JobTypeLeverageDriftNotification } + +func (LeverageDriftNotificationArgs) Timeout() time.Duration { return 30 * time.Second } + +// JobInsertOptionsForLeverageDriftNotification returns insert options for leverage drift +// notification jobs — mirrors JobInsertOptionsForRiskNotification's shape (email queue, +// deduped by args over a short window so a rapid re-drift/reopen doesn't spam). +func JobInsertOptionsForLeverageDriftNotification() *river.InsertOpts { + return &river.InsertOpts{ + Queue: "email", + MaxAttempts: 3, + UniqueOpts: river.UniqueOpts{ + ByArgs: true, + ByPeriod: 5 * time.Minute, + }, + } +} + +// notificationKindForReason maps applyDriftToLink's free-text reason to one of the two +// notification kinds the ticket asks for: an explicit revocation (the upstream offering +// was revoked, or its leveraged authorization was deleted) is leverage_revoked; every +// other drift trigger (a content-changing version bump, or deprecation) is the softer +// leverage_drifted. +func notificationKindForReason(reason string) notification.Kind { + if strings.Contains(reason, "revoked") { + return leverageRevokedNotificationKind + } + return leverageDriftedNotificationKind +} + +// leverageDriftNotificationModel is the template model for both leverage_drifted and +// leverage_revoked — deliberately small, reusing riskNotificationData (the drift risk is +// a plain risks.Risk row) rather than introducing a parallel data-loading path. +type leverageDriftNotificationModel struct { + OwnerName string + RiskTitle string + SSPName string + Reason string + RiskURL string +} + +func (m leverageDriftNotificationModel) templateData() map[string]interface{} { + return map[string]interface{}{ + "OwnerName": m.OwnerName, + "RiskTitle": m.RiskTitle, + "SSPName": m.SSPName, + "Reason": m.Reason, + "RiskURL": m.RiskURL, + } +} + +func newLeverageDriftNotificationModel(userName string, data riskNotificationData, reason string) leverageDriftNotificationModel { + ownerName := strings.TrimSpace(userName) + if ownerName == "" { + ownerName = strings.TrimSpace(data.OwnerName) + } + return leverageDriftNotificationModel{ + OwnerName: ownerName, + RiskTitle: strings.TrimSpace(data.RiskTitle), + SSPName: strings.TrimSpace(data.SSPName), + Reason: strings.TrimSpace(reason), + RiskURL: strings.TrimSpace(data.RiskURL), + } +} + +func renderLeverageDriftedEmail(model leverageDriftNotificationModel) (emailprovider.TemplateContent, error) { + return leverageDriftEmailContent(model, "leverage-drifted", "Leveraged control drifted"), nil +} + +func renderLeverageRevokedEmail(model leverageDriftNotificationModel) (emailprovider.TemplateContent, error) { + return leverageDriftEmailContent(model, "leverage-revoked", "Leveraged control revoked"), nil +} + +func leverageDriftEmailContent(model leverageDriftNotificationModel, templateName, subjectPrefix string) emailprovider.TemplateContent { + textBody := model.RiskTitle + if model.Reason != "" { + textBody = fmt.Sprintf("%s (%s)", textBody, model.Reason) + } + return emailprovider.TemplateContent{ + TemplateName: templateName, + TemplateData: model.templateData(), + Subject: fmt.Sprintf("%s: %s", subjectPrefix, model.RiskTitle), + TextBody: textBody, + } +} + +func buildLeverageDriftNotificationRequest( + args LeverageDriftNotificationArgs, + userName string, + data riskNotificationData, +) notification.Request { + kind := notificationKindForReason(args.Reason) + return newUserNotificationRequest( + kind, + args.OwnerUserID.String(), + newLeverageDriftNotificationModel(userName, data, args.Reason), + newJobDispatchOptions(string(kind), args.Channel, args.RiskID.String(), args.LinkID.String(), args.OwnerUserID.String()), + ) +} + +// LeverageDriftNotificationWorker dispatches leverage_drifted/leverage_revoked +// notifications for a drift risk, mirroring RiskReviewDueReminderWorker's shape exactly +// — the drift risk is a plain risks.Risk row, so dispatchRiskReminderNotification's +// existing load-risk/load-owner/dispatch pipeline needs no changes to serve it. +type LeverageDriftNotificationWorker struct { + db *gorm.DB + userRepo UserRepository + notificationServiceFactory *RiskNotificationServiceFactory + webBaseURL string + logger *zap.SugaredLogger +} + +func NewLeverageDriftNotificationWorker( + db *gorm.DB, + userRepo UserRepository, + webBaseURL string, + notificationServiceFactory *RiskNotificationServiceFactory, + logger *zap.SugaredLogger, +) *LeverageDriftNotificationWorker { + return &LeverageDriftNotificationWorker{ + db: db, + userRepo: userRepo, + notificationServiceFactory: notificationServiceFactory, + webBaseURL: webBaseURL, + logger: logger, + } +} + +func (w *LeverageDriftNotificationWorker) Work(ctx context.Context, job *river.Job[LeverageDriftNotificationArgs]) error { + args := job.Args + if _, ok := normalizeRequestedDeliveryChannel(args.Channel); !ok { + w.logger.Warnw("LeverageDriftNotificationWorker: invalid delivery channel, skipping", + "risk_id", args.RiskID, + "owner_user_id", args.OwnerUserID, + "channel", args.Channel, + ) + return nil + } + + return dispatchRiskReminderNotification( + ctx, + w.db, + w.userRepo, + w.notificationServiceFactory, + w.webBaseURL, + w.logger, + args.RiskID, args.OwnerUserID, + func(userName string, data riskNotificationData) notification.Request { + return requestWithSourceJobID(buildLeverageDriftNotificationRequest(args, userName, data), riverJobID(job)) + }, + ) +} diff --git a/internal/service/worker/leverage_notifications_test.go b/internal/service/worker/leverage_notifications_test.go new file mode 100644 index 00000000..1f2ae000 --- /dev/null +++ b/internal/service/worker/leverage_notifications_test.go @@ -0,0 +1,76 @@ +package worker + +import ( + "context" + "testing" + "time" + + "github.com/compliance-framework/api/internal/service/email/types" + "github.com/compliance-framework/api/internal/service/notification" + riskrel "github.com/compliance-framework/api/internal/service/relational/risks" + "github.com/google/uuid" + "github.com/riverqueue/river" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +func TestNotificationKindForReasonMapsRevocationVsDrift(t *testing.T) { + require.Equal(t, leverageRevokedNotificationKind, notificationKindForReason("upstream offering revoked")) + require.Equal(t, leverageRevokedNotificationKind, notificationKindForReason("leveraged authorization revoked")) + require.Equal(t, leverageDriftedNotificationKind, notificationKindForReason("upstream offering content changed")) + require.Equal(t, leverageDriftedNotificationKind, notificationKindForReason("upstream offering deprecated")) +} + +func TestLeverageDriftNotificationWorker_DispatchesEmailOnRevocation(t *testing.T) { + db := newRiskWorkersTestDB(t) + logger := zap.NewNop().Sugar() + risk, ownerID := createTestRiskWithOwner(t, db, riskrel.RiskStatusOpen, nil, time.Now().UTC()) + require.NoError(t, db.Model(&riskrel.Risk{}).Where("id = ?", risk.ID).Update("source_type", string(riskrel.RiskSourceTypeInheritedRevoked)).Error) + createTestUser(t, db, ownerID, true) + + mockEmail := &MockEmailService{} + mockEmail.On("UseTemplate", "leverage-revoked", mock.Anything).Return("ok", "ok", nil) + mockEmail.On("GetDefaultFromAddress").Return("noreply@example.com") + mockEmail.On("Send", mock.Anything, mock.MatchedBy(func(msg *types.Message) bool { + return len(msg.To) == 1 && msg.To[0] == ownerID.String()+"@example.com" + })).Return(&types.SendResult{Success: true, MessageID: "leverage-msg"}, nil) + + userRepo := NewGORMUserRepository(db) + worker := NewLeverageDriftNotificationWorker(db, userRepo, "https://app.example.com", newTestRiskNotificationServiceFactory(mockEmail, nil), logger) + + err := worker.Work(context.Background(), &river.Job[LeverageDriftNotificationArgs]{ + Args: LeverageDriftNotificationArgs{ + RiskID: *risk.ID, + LinkID: uuid.New(), + OwnerUserID: ownerID, + Channel: notification.DeliveryChannelEmail, + Reason: "upstream offering revoked", + }, + }) + require.NoError(t, err) + mockEmail.AssertExpectations(t) +} + +func TestLeverageDriftNotificationWorker_RespectsSubscription(t *testing.T) { + db := newRiskWorkersTestDB(t) + logger := zap.NewNop().Sugar() + risk, ownerID := createTestRiskWithOwner(t, db, riskrel.RiskStatusOpen, nil, time.Now().UTC()) + createTestUser(t, db, ownerID, false) + + mockEmail := &MockEmailService{} + userRepo := NewGORMUserRepository(db) + worker := NewLeverageDriftNotificationWorker(db, userRepo, "https://app.example.com", newTestRiskNotificationServiceFactory(mockEmail, nil), logger) + + err := worker.Work(context.Background(), &river.Job[LeverageDriftNotificationArgs]{ + Args: LeverageDriftNotificationArgs{ + RiskID: *risk.ID, + LinkID: uuid.New(), + OwnerUserID: ownerID, + Channel: notification.DeliveryChannelEmail, + Reason: "upstream offering content changed", + }, + }) + require.NoError(t, err) + mockEmail.AssertNotCalled(t, "Send") +} diff --git a/internal/service/worker/risk_notifications.go b/internal/service/worker/risk_notifications.go index 241b37f6..f38f176a 100644 --- a/internal/service/worker/risk_notifications.go +++ b/internal/service/worker/risk_notifications.go @@ -84,6 +84,18 @@ func NewRiskNotificationServiceFactory( renderRiskOpenDigestEmail, renderRiskOpenDigestSlack, ), + newTypedEmailOnlyNotificationDefinition( + leverageDriftedNotificationKind, + notification.SubscriptionGateRiskNotifications, + newNotificationModelDecoder[leverageDriftNotificationModel]("leverage drift notification model"), + renderLeverageDriftedEmail, + ), + newTypedEmailOnlyNotificationDefinition( + leverageRevokedNotificationKind, + notification.SubscriptionGateRiskNotifications, + newNotificationModelDecoder[leverageDriftNotificationModel]("leverage drift notification model"), + renderLeverageRevokedEmail, + ), }, } } diff --git a/internal/service/worker/service.go b/internal/service/worker/service.go index 79396bd9..222b237a 100644 --- a/internal/service/worker/service.go +++ b/internal/service/worker/service.go @@ -1106,6 +1106,53 @@ func (s *Service) EnqueueOrphanedRiskCleanup(ctx context.Context, sspID uuid.UUI return nil } +// EnqueueLeverageDriftNotification enqueues one leverage-drift notification job per +// resolved owner of riskID (BCH-1341). Resolution reuses the same risk-owner mechanism +// every other risk notification is targeted at (resolveRiskOwnerUserIDsBatch) — a drift +// risk is a plain risks.Risk row, so no new "who owns this SSP" concept is needed. A risk +// with no assigned owner yet enqueues no job (nobody to notify), consistent with how +// every other owner-directed risk notification behaves. Plain uuid.UUID/string +// parameters (not a shared struct) so *worker.Service can satisfy +// oscal.SSPJobEnqueuer's method signature without a circular import. +func (s *Service) EnqueueLeverageDriftNotification(ctx context.Context, riskID, linkID, downstreamSSPID uuid.UUID, reason string) error { + if !s.config.Enabled || s.client == nil { + return nil + } + + var risk riskrel.Risk + if err := s.db.WithContext(ctx).First(&risk, "id = ?", riskID).Error; err != nil { + return fmt.Errorf("failed to load drift risk for notification enqueue: %w", err) + } + + ownersByRiskID, err := resolveRiskOwnerUserIDsBatch(ctx, s.db, []riskrel.Risk{risk}) + if err != nil { + return fmt.Errorf("failed to resolve drift risk owners: %w", err) + } + + owners := ownersByRiskID[riskID] + if len(owners) == 0 { + return nil + } + + params := make([]river.InsertManyParams, 0, len(owners)) + for _, ownerID := range owners { + params = append(params, river.InsertManyParams{ + Args: LeverageDriftNotificationArgs{ + RiskID: riskID, + LinkID: linkID, + OwnerUserID: ownerID, + Reason: reason, + }, + InsertOpts: JobInsertOptionsForLeverageDriftNotification(), + }) + } + + if _, err := s.client.InsertMany(ctx, params); err != nil { + return fmt.Errorf("failed to enqueue leverage drift notification jobs for risk %s: %w", riskID, err) + } + return nil +} + func (s *Service) EnqueueDashboardSuggestionCells(ctx context.Context, runID uuid.UUID, cellCount int) error { if s == nil || s.config == nil || !s.config.Enabled || s.client == nil || !s.aiEnabled { return ErrDashboardSuggestionWorkerDisabled From df3940135d36016a4f8fd201f7eef942cb648ec9 Mon Sep 17 00:00:00 2001 From: James Salt Date: Fri, 10 Jul 2026 15:44:33 +0100 Subject: [PATCH 2/2] BCH-1341: Address PR review comments on doc/enum gaps, duplicated enqueue logic, a re-attestation race, and a fragile reason-string match Documents the two valid updateOfferingStatusRequest.Status values in swagger, extracts the repeated drift-notification enqueue block into one helper, closes a TOCTOU window where a stale re-attestation could silently overwrite a link that had already been cleared or re-drifted, replaces a substring match with an explicit reason-to-kind mapping, and drops an unused parameter from the notification enqueuer interface. --- docs/docs.go | 9 +++- docs/swagger.json | 9 +++- docs/swagger.yaml | 5 +++ .../dashboard_suggestions_integration_test.go | 2 +- .../api/handler/oscal/ssp_export_offerings.go | 15 ++----- internal/api/handler/oscal/ssp_leverage.go | 35 +++++++++++---- .../api/handler/oscal/ssp_leverage_drift.go | 18 +++++++- .../oscal/ssp_leverage_drift_enqueue_test.go | 12 +++--- .../oscal/ssp_leverage_reattest_test.go | 43 +++++++++++++++++++ .../handler/oscal/system_security_plans.go | 8 +--- .../service/worker/leverage_notifications.go | 18 +++++--- internal/service/worker/service.go | 2 +- 12 files changed, 133 insertions(+), 43 deletions(-) diff --git a/docs/docs.go b/docs/docs.go index 5124ac13..a03be220 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -38946,9 +38946,16 @@ const docTemplate = `{ }, "oscal.updateOfferingStatusRequest": { "type": "object", + "required": [ + "status" + ], "properties": { "status": { - "type": "string" + "type": "string", + "enum": [ + "deprecated", + "revoked" + ] } } }, diff --git a/docs/swagger.json b/docs/swagger.json index 8d8c49f8..3ab0818f 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -38940,9 +38940,16 @@ }, "oscal.updateOfferingStatusRequest": { "type": "object", + "required": [ + "status" + ], "properties": { "status": { - "type": "string" + "type": "string", + "enum": [ + "deprecated", + "revoked" + ] } } }, diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 941559bd..4170b0a7 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -4425,7 +4425,12 @@ definitions: oscal.updateOfferingStatusRequest: properties: status: + enum: + - deprecated + - revoked type: string + required: + - status type: object oscal.upstreamResponsibility: properties: diff --git a/internal/api/handler/oscal/dashboard_suggestions_integration_test.go b/internal/api/handler/oscal/dashboard_suggestions_integration_test.go index e74a366e..fdf078cd 100644 --- a/internal/api/handler/oscal/dashboard_suggestions_integration_test.go +++ b/internal/api/handler/oscal/dashboard_suggestions_integration_test.go @@ -38,7 +38,7 @@ func (f *dashboardSuggestionFakeEnqueuer) EnqueueOrphanedRiskCleanup(context.Con return nil } -func (f *dashboardSuggestionFakeEnqueuer) EnqueueLeverageDriftNotification(context.Context, uuid.UUID, uuid.UUID, uuid.UUID, string) error { +func (f *dashboardSuggestionFakeEnqueuer) EnqueueLeverageDriftNotification(context.Context, uuid.UUID, uuid.UUID, string) error { return nil } diff --git a/internal/api/handler/oscal/ssp_export_offerings.go b/internal/api/handler/oscal/ssp_export_offerings.go index 088aec55..1833c8e9 100644 --- a/internal/api/handler/oscal/ssp_export_offerings.go +++ b/internal/api/handler/oscal/ssp_export_offerings.go @@ -1,7 +1,6 @@ package oscal import ( - "context" "crypto/sha256" "encoding/hex" "encoding/json" @@ -736,11 +735,7 @@ func (h *SSPExportOfferingHandler) Publish(ctx echo.Context) error { h.sugar.Errorf("Failed to publish export offering: %v", err) return ctx.JSON(http.StatusInternalServerError, api.NewError(err)) } - if len(driftedLinks) > 0 { - enqueueCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx.Request().Context()), 10*time.Second) - defer cancel() - enqueueLeverageDriftNotifications(enqueueCtx, h.sugar, h.jobEnqueuer, driftedLinks) - } + enqueueLeverageDriftNotificationsAsync(ctx, h.sugar, h.jobEnqueuer, driftedLinks) var published relational.SSPExportOffering if err := h.db.Preload("Items").First(&published, "id = ?", offering.ID).Error; err != nil { @@ -754,7 +749,7 @@ func (h *SSPExportOfferingHandler) Publish(ctx echo.Context) error { // terminal, drift-triggering transitions are allowed here — draft/published stay owned // by CreateOffering/Publish. type updateOfferingStatusRequest struct { - Status string `json:"status"` + Status string `json:"status" binding:"required" enums:"deprecated,revoked"` } func (r updateOfferingStatusRequest) validate() error { @@ -823,11 +818,7 @@ func (h *SSPExportOfferingHandler) UpdateOfferingStatus(ctx echo.Context) error h.sugar.Errorf("Failed to update export offering status: %v", err) return ctx.JSON(http.StatusInternalServerError, api.NewError(err)) } - if len(driftedLinks) > 0 { - enqueueCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx.Request().Context()), 10*time.Second) - defer cancel() - enqueueLeverageDriftNotifications(enqueueCtx, h.sugar, h.jobEnqueuer, driftedLinks) - } + enqueueLeverageDriftNotificationsAsync(ctx, h.sugar, h.jobEnqueuer, driftedLinks) var updated relational.SSPExportOffering if err := h.db.Preload("Items").First(&updated, "id = ?", offering.ID).Error; err != nil { diff --git a/internal/api/handler/oscal/ssp_leverage.go b/internal/api/handler/oscal/ssp_leverage.go index 28396843..d365e5fe 100644 --- a/internal/api/handler/oscal/ssp_leverage.go +++ b/internal/api/handler/oscal/ssp_leverage.go @@ -32,6 +32,13 @@ const thisSystemComponentType = "this-system" // (non-racy) duplicate detection. var errDuplicateLeverageLink = errors.New("already subscribed to this provided-uuid") +// errLeverageLinkNoLongerDrifted signals that ReAttest's pre-check (link.Status == +// Drifted, read before the transaction opened) is now stale by the time the +// transaction's own update runs — a concurrent re-attest already cleared it, or a new +// drift trigger landed in between. Mapped to 409: retrying with a fresh read is the +// correct client action, not treated as a 500. +var errLeverageLinkNoLongerDrifted = errors.New("leverage link is no longer drifted") + // isUniqueViolation reports whether err is a Postgres unique-constraint violation. GORM's // ErrDuplicatedKey translation is unreliable here (same issue noted in // internal/api/handler/groups.go's isUniqueViolation), so the driver code is inspected @@ -838,14 +845,23 @@ func (h *SSPLeverageHandler) ReAttest(ctx echo.Context) error { } satisfaction, _ := deriveSatisfaction(fullSet, satisfiedUUIDs) - if err := tx.Model(&relational.SSPLeverageLink{}).Where("id = ?", link.ID).Updates(map[string]any{ - "offering_version": offering.Version, - "status": relational.SSPLeverageStatusActive, - "satisfaction": satisfaction, - "attested_at": &now, - "attested_by_id": attestedBy, - }).Error; err != nil { - return fmt.Errorf("failed to update leverage link: %w", err) + // Require the link to still be Drifted at update time, not just at the pre-check + // read above — otherwise a concurrent re-attest, or a fresh drift trigger landing + // in between, would be silently overwritten by this stale request. + result := tx.Model(&relational.SSPLeverageLink{}). + Where("id = ? AND status = ?", link.ID, relational.SSPLeverageStatusDrifted). + Updates(map[string]any{ + "offering_version": offering.Version, + "status": relational.SSPLeverageStatusActive, + "satisfaction": satisfaction, + "attested_at": &now, + "attested_by_id": attestedBy, + }) + if result.Error != nil { + return fmt.Errorf("failed to update leverage link: %w", result.Error) + } + if result.RowsAffected == 0 { + return errLeverageLinkNoLongerDrifted } dedupeKey := computeDedupeKeyForLeverageDrift(*link.ID) @@ -873,6 +889,9 @@ func (h *SSPLeverageHandler) ReAttest(ctx echo.Context) error { } return risks.NewRiskService(tx).RecordRiskScoreSnapshot(tx, *driftRisk.ID, risks.RiskEventTypeStatusChange, attestedBy, now) }); err != nil { + if errors.Is(err, errLeverageLinkNoLongerDrifted) { + return ctx.JSON(http.StatusConflict, api.NewError(err)) + } h.sugar.Errorf("Failed to re-attest leverage link: %v", err) return ctx.JSON(http.StatusInternalServerError, api.NewError(err)) } diff --git a/internal/api/handler/oscal/ssp_leverage_drift.go b/internal/api/handler/oscal/ssp_leverage_drift.go index 11e7fc64..cab808a2 100644 --- a/internal/api/handler/oscal/ssp_leverage_drift.go +++ b/internal/api/handler/oscal/ssp_leverage_drift.go @@ -7,6 +7,7 @@ import ( "time" "github.com/google/uuid" + "github.com/labstack/echo/v4" "go.uber.org/zap" "gorm.io/datatypes" "gorm.io/gorm" @@ -40,13 +41,28 @@ func enqueueLeverageDriftNotifications(ctx context.Context, sugar *zap.SugaredLo return } for _, info := range links { - if err := jobEnqueuer.EnqueueLeverageDriftNotification(ctx, info.RiskID, info.LinkID, info.DownstreamSSPID, info.Reason); err != nil { + if err := jobEnqueuer.EnqueueLeverageDriftNotification(ctx, info.RiskID, info.LinkID, info.Reason); err != nil { sugar.Warnw("Failed to enqueue leverage drift notification", "risk_id", info.RiskID, "link_id", info.LinkID, "error", err) } } } +// enqueueLeverageDriftNotificationsAsync is the shared post-commit call every drift +// trigger handler (Publish, UpdateOfferingStatus, +// DeleteSystemImplementationLeveragedAuthorization) makes: a no-op when nothing +// drifted, otherwise a detached, timeout-bounded context (mirroring the +// EnqueueOrphanedRiskCleanup call sites) so a slow/failed enqueue can't hang or be +// cancelled by the already-completed request. +func enqueueLeverageDriftNotificationsAsync(ctx echo.Context, sugar *zap.SugaredLogger, jobEnqueuer SSPJobEnqueuer, links []driftedLinkInfo) { + if len(links) == 0 { + return + } + enqueueCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx.Request().Context()), 10*time.Second) + defer cancel() + enqueueLeverageDriftNotifications(enqueueCtx, sugar, jobEnqueuer, links) +} + // computeDedupeKeyForLeverageDrift returns the dedupe key for the drift risk // associated with a single SSPLeverageLink. One risk per leverage link: the link // itself is the natural, deterministic scope (no risk template is involved, unlike diff --git a/internal/api/handler/oscal/ssp_leverage_drift_enqueue_test.go b/internal/api/handler/oscal/ssp_leverage_drift_enqueue_test.go index ca46fab2..1a6c2686 100644 --- a/internal/api/handler/oscal/ssp_leverage_drift_enqueue_test.go +++ b/internal/api/handler/oscal/ssp_leverage_drift_enqueue_test.go @@ -30,10 +30,9 @@ type spyJobEnqueuer struct { } type spyEnqueueCall struct { - RiskID uuid.UUID - LinkID uuid.UUID - DownstreamSSPID uuid.UUID - Reason string + RiskID uuid.UUID + LinkID uuid.UUID + Reason string } func (s *spyJobEnqueuer) EnqueueOrphanedRiskCleanup(context.Context, uuid.UUID, *uuid.UUID, *uuid.UUID) error { @@ -44,8 +43,8 @@ func (s *spyJobEnqueuer) EnqueueDashboardSuggestionCells(context.Context, uuid.U return nil } -func (s *spyJobEnqueuer) EnqueueLeverageDriftNotification(_ context.Context, riskID, linkID, downstreamSSPID uuid.UUID, reason string) error { - s.calls = append(s.calls, spyEnqueueCall{RiskID: riskID, LinkID: linkID, DownstreamSSPID: downstreamSSPID, Reason: reason}) +func (s *spyJobEnqueuer) EnqueueLeverageDriftNotification(_ context.Context, riskID, linkID uuid.UUID, reason string) error { + s.calls = append(s.calls, spyEnqueueCall{RiskID: riskID, LinkID: linkID, Reason: reason}) return nil } @@ -93,7 +92,6 @@ func TestPublishEnqueuesLeverageDriftNotificationOnVersionBump(t *testing.T) { require.Len(t, spy.calls, 1) require.Equal(t, *link.ID, spy.calls[0].LinkID) - require.Equal(t, link.DownstreamSSPID, spy.calls[0].DownstreamSSPID) } func TestUpdateOfferingStatusEnqueuesLeverageDriftNotification(t *testing.T) { diff --git a/internal/api/handler/oscal/ssp_leverage_reattest_test.go b/internal/api/handler/oscal/ssp_leverage_reattest_test.go index 1531dd8b..2bf4d7e7 100644 --- a/internal/api/handler/oscal/ssp_leverage_reattest_test.go +++ b/internal/api/handler/oscal/ssp_leverage_reattest_test.go @@ -106,3 +106,46 @@ func TestReAttestRejectsWrongDownstreamSSP(t *testing.T) { require.NoError(t, h.ReAttest(ctx)) require.Equal(t, http.StatusNotFound, rec.Code) } + +// TestReAttestRejectsWhenLinkNoLongerDriftedAtUpdateTime: ReAttest's pre-check reads the +// link once (must be Drifted) before opening its transaction. If the link stops being +// Drifted in between that read and the transaction's own update — a concurrent +// re-attest, or a fresh drift landing right after — the update must not silently +// overwrite that newer state. A query callback deterministically injects the concurrent +// change right after the pre-check read completes and before the transactional update +// runs, mirroring TestSyncExportOfferingAbortsOnConcurrentModification's technique. +func TestReAttestRejectsWhenLinkNoLongerDriftedAtUpdateTime(t *testing.T) { + db := newSSPLeverageTestDB(t) + fx := newLeverageFixture(t, db) + link, riskID := subscribeAndDrift(t, db, fx) + require.Equal(t, relational.SSPLeverageStatusDrifted, link.Status) + + var queryCount int + require.NoError(t, db.Callback().Query().After("gorm:query"). + Register("test:simulate-concurrent-reattest", func(*gorm.DB) { + queryCount++ + if queryCount == 1 { + // Simulate a concurrent request winning the race and clearing drift first. + require.NoError(t, db.Exec( + "UPDATE ssp_leverage_links SET status = ? WHERE id = ?", + string(relational.SSPLeverageStatusActive), link.ID.String(), + ).Error) + } + })) + t.Cleanup(func() { _ = db.Callback().Query().Remove("test:simulate-concurrent-reattest") }) + + h := NewSSPLeverageHandler(zap.NewNop().Sugar(), db, &stubPDP{allow: true}, authz.FailClosed) + ctx, rec := newReAttestRequestContext(fx.downstreamSSPID, *link.ID) + require.NoError(t, h.ReAttest(ctx)) + require.Equal(t, http.StatusConflict, rec.Code) + + // The concurrent write's state must survive untouched — no bogus double-attest. + var reloadedLink relational.SSPLeverageLink + require.NoError(t, db.First(&reloadedLink, "id = ?", link.ID).Error) + require.Equal(t, relational.SSPLeverageStatusActive, reloadedLink.Status) + require.Equal(t, 1, reloadedLink.OfferingVersion, "the losing request must not have bumped OfferingVersion") + + var risk risks.Risk + require.NoError(t, db.First(&risk, "id = ?", riskID).Error) + require.Equal(t, string(risks.RiskStatusOpen), risk.Status, "the losing request must not have remediated the risk") +} diff --git a/internal/api/handler/oscal/system_security_plans.go b/internal/api/handler/oscal/system_security_plans.go index 5beb6d3a..009061a8 100644 --- a/internal/api/handler/oscal/system_security_plans.go +++ b/internal/api/handler/oscal/system_security_plans.go @@ -34,7 +34,7 @@ import ( type SSPJobEnqueuer interface { EnqueueOrphanedRiskCleanup(ctx context.Context, sspID uuid.UUID, oldProfileID, newProfileID *uuid.UUID) error EnqueueDashboardSuggestionCells(ctx context.Context, runID uuid.UUID, cellCount int) error - EnqueueLeverageDriftNotification(ctx context.Context, riskID, linkID, downstreamSSPID uuid.UUID, reason string) error + EnqueueLeverageDriftNotification(ctx context.Context, riskID, linkID uuid.UUID, reason string) error } // profileSummary is a lightweight DTO returned by the multi-profile list endpoint. @@ -3509,11 +3509,7 @@ func (h *SystemSecurityPlanHandler) DeleteSystemImplementationLeveragedAuthoriza h.sugar.Errorf("Failed to delete leveraged authorization: %v", err) return ctx.JSON(http.StatusInternalServerError, api.NewError(err)) } - if len(driftedLinks) > 0 { - enqueueCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx.Request().Context()), 10*time.Second) - defer cancel() - enqueueLeverageDriftNotifications(enqueueCtx, h.sugar, h.jobEnqueuer, driftedLinks) - } + enqueueLeverageDriftNotificationsAsync(ctx, h.sugar, h.jobEnqueuer, driftedLinks) if rowsAffected == 0 { return ctx.JSON(http.StatusNotFound, api.NewError(fmt.Errorf("leveraged authorization not found"))) diff --git a/internal/service/worker/leverage_notifications.go b/internal/service/worker/leverage_notifications.go index a1d2e82c..831aff28 100644 --- a/internal/service/worker/leverage_notifications.go +++ b/internal/service/worker/leverage_notifications.go @@ -55,13 +55,21 @@ func JobInsertOptionsForLeverageDriftNotification() *river.InsertOpts { } } +// leverageRevocationReasons is the exact set of applyDriftToLink's free-text reasons +// (ssp_leverage_drift.go, system_security_plans.go) that count as an explicit +// revocation — the upstream offering was revoked, or its leveraged authorization was +// deleted. Every other drift trigger (a content-changing version bump, or deprecation) +// is the softer leverage_drifted. +var leverageRevocationReasons = map[string]bool{ + "upstream offering revoked": true, + "leveraged authorization revoked": true, +} + // notificationKindForReason maps applyDriftToLink's free-text reason to one of the two -// notification kinds the ticket asks for: an explicit revocation (the upstream offering -// was revoked, or its leveraged authorization was deleted) is leverage_revoked; every -// other drift trigger (a content-changing version bump, or deprecation) is the softer -// leverage_drifted. +// notification kinds the ticket asks for. An unrecognized reason defaults to +// leverage_drifted, the softer of the two. func notificationKindForReason(reason string) notification.Kind { - if strings.Contains(reason, "revoked") { + if leverageRevocationReasons[reason] { return leverageRevokedNotificationKind } return leverageDriftedNotificationKind diff --git a/internal/service/worker/service.go b/internal/service/worker/service.go index 222b237a..cefed17a 100644 --- a/internal/service/worker/service.go +++ b/internal/service/worker/service.go @@ -1114,7 +1114,7 @@ func (s *Service) EnqueueOrphanedRiskCleanup(ctx context.Context, sspID uuid.UUI // every other owner-directed risk notification behaves. Plain uuid.UUID/string // parameters (not a shared struct) so *worker.Service can satisfy // oscal.SSPJobEnqueuer's method signature without a circular import. -func (s *Service) EnqueueLeverageDriftNotification(ctx context.Context, riskID, linkID, downstreamSSPID uuid.UUID, reason string) error { +func (s *Service) EnqueueLeverageDriftNotification(ctx context.Context, riskID, linkID uuid.UUID, reason string) error { if !s.config.Enabled || s.client == nil { return nil }