From 730202038404db73ea66f9e2f67f45ce344e549e Mon Sep 17 00:00:00 2001 From: captainpacket Date: Sat, 25 Jul 2026 19:42:27 -0500 Subject: [PATCH 1/2] feat: detect post-apply divergence, guard manifest quality, lock webhook state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three residual risks from v3.0.0. None is a regression; each is a gap where a partial mitigation was possible but unbuilt. Post-PATCH verification. The gateway now re-reads each setup after a successful PATCH and compares the observed state against the approved target. A mismatch stops the apply, records planned -> pending -> applied -> conflicted in the journal, prints a critical alert, and points at the rollback artifact. It deliberately does not attempt remediation: a corrective PATCH would repeat the same overwrite, and deciding whose change wins is a human call. It ignores region test timestamps, server-managed errorMsg, account ordering, empty-versus-nil containers, and unmodeled response fields. Flagging ordinary server behavior as a clobber would train operators to ignore the alert, which is worse than not having it. This narrows rather than closes the no-CAS gap, and the boundary is worth stating. It detects unexplained divergence after a write. It cannot detect the case where a concurrent edit landed between the pre-PATCH re-read and the PATCH itself: our full-list write erases it and the result then equals our intent exactly, leaving nothing observable. Combined with the existing pre-PATCH comparison, what remains uncovered is an edit arriving inside the PATCH execution window — narrow, silent, and not addressable without a version token Forward does not expose. Manifest safeguards. sync-accounts is the only removal path and is exactly as safe as its manifest. The docs warn against deriving one from NQE output, since that reproduces by hand the unsound inference v3.0.0 retired, but nothing enforced it. A manifest closely matching observed NQE inventory while diverging from configured membership now emits manifest_matches_nqe_observation. It warns rather than blocks — a legitimate manifest can coincidentally match, and blocking on a heuristic invites workarounds. Removal previews now state the fraction outright ("removes 968 of 978 accounts (98.98%)") regardless of what the ceilings permit. Webhook state locking. Two daemons sharing a state file would corrupt each other's dedupe records and watermarks; this was documented as unsupported and is now enforced with a flock on a stable .lock sidecar. The sidecar matters because the state file is atomically renamed on write, so locking that inode would silently lose protection. Existing tests were made stateful so the verification GET observes the PATCH it accepted. No assertion was removed. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Arw4zhKVDNry9zV1Ej6jRt --- cmd/awssync/main.go | 85 ++++++++- cmd/awssync/main_test.go | 140 +++++++++++++-- internal/app/account_manifest.go | 124 ++++++++++++- internal/app/account_manifest_test.go | 112 +++++++++++- internal/app/apply_gateway.go | 234 +++++++++++++++++++++++-- internal/app/apply_gateway_test.go | 156 +++++++++++++++++ internal/app/apply_plan.go | 44 ++--- internal/app/apply_plan_test.go | 27 ++- internal/app/external_id.go | 50 +++--- internal/app/external_id_test.go | 9 +- internal/app/run.go | 129 +++++++++----- internal/app/run_test.go | 18 +- internal/webhook/durable_queue_test.go | 11 ++ internal/webhook/server.go | 23 +++ internal/webhook/server_test.go | 32 ++++ internal/webhook/state.go | 48 +++++ 16 files changed, 1114 insertions(+), 128 deletions(-) diff --git a/cmd/awssync/main.go b/cmd/awssync/main.go index 98ec88a..acbe076 100644 --- a/cmd/awssync/main.go +++ b/cmd/awssync/main.go @@ -157,7 +157,7 @@ func newRootCommand() *cobra.Command { } summary, err := app.Run(cmd.Context(), cfg) if err != nil { - return err + return emitVerificationFailureResult(cmd, v, summary, summaryVerificationFailures(summary), err) } return emitResult(cmd, v, summary) }, @@ -268,7 +268,7 @@ func newSafeSyncCommand(v *viper.Viper) *cobra.Command { } result, err := app.Run(cmd.Context(), base) if err != nil { - return err + return emitVerificationFailureResult(cmd, v, result, summaryVerificationFailures(result), err) } emitSafeSyncComplete(result) return nil @@ -337,7 +337,7 @@ func newExternalIDCommand(v *viper.Viper) *cobra.Command { }, }) if err != nil { - return err + return emitVerificationFailureResult(cmd, v, summary, externalIDVerificationFailures(summary), err) } return emitResult(cmd, v, summary) }, @@ -567,7 +567,7 @@ func newApplyPlanCommand(v *viper.Viper) *cobra.Command { AuthorizationActor: "CLI apply-plan --yes", }) if err != nil { - return err + return emitVerificationFailureResult(cmd, v, summary, applyPlanVerificationFailures(summary), err) } return emitResult(cmd, v, summary) }, @@ -897,7 +897,7 @@ func newSyncAccountsCommand(v *viper.Viper) *cobra.Command { } summary, err := app.SyncAWSAccountManifest(cmd.Context(), cfg, accounts) if err != nil { - return err + return emitVerificationFailureResult(cmd, v, summary, summaryVerificationFailures(summary), err) } return emitResult(cmd, v, summary) }, @@ -1441,6 +1441,7 @@ func emitWaitHuman(result *monitor.WaitResult) error { func emitApplyPlanHuman(summary *app.ApplyPlanSummary) error { fmt.Fprintln(os.Stdout, "Apply-plan report") + emitApplyVerificationFailures(summary.ApplyVerificationFailures) fmt.Fprintf(os.Stdout, " host: %s\n", summary.Host) fmt.Fprintf(os.Stdout, " network: %s\n", summary.NetworkID) fmt.Fprintf(os.Stdout, " plan: %s\n", summary.PlanPath) @@ -1457,6 +1458,7 @@ func emitApplyPlanHuman(summary *app.ApplyPlanSummary) error { func emitExternalIDHuman(summary *app.ExternalIDSummary) error { fmt.Fprintln(os.Stdout, "External ID migration report") + emitApplyVerificationFailures(summary.ApplyVerificationFailures) fmt.Fprintf(os.Stdout, " host: %s\n", summary.Host) fmt.Fprintf(os.Stdout, " network: %s\n", summary.NetworkID) fmt.Fprintf(os.Stdout, " setup: %s\n", summary.SetupID) @@ -1533,6 +1535,9 @@ func emitSummaryHuman(summary *app.Summary) error { return emitAWSOrganizationsHuman(summary) } fmt.Fprintf(os.Stdout, "Sync report\n") + emitApplyVerificationFailures(summary.ApplyVerificationFailures) + emitSafetyWarnings(os.Stdout, summary.SafetyWarnings) + emitRemovalImpacts(os.Stdout, summary.RemovalImpacts) fmt.Fprintf(os.Stdout, " host: %s\n", summary.Host) fmt.Fprintf(os.Stdout, " network: %s\n", summary.NetworkID) if summary.Source != "" { @@ -1618,6 +1623,74 @@ func emitSummaryHuman(summary *app.Summary) error { return nil } +func emitApplyVerificationFailures(failures []app.ApplyVerificationFailure) { + if len(failures) == 0 { + return + } + fmt.Fprintln(os.Stdout, "\n!!! CRITICAL POST-APPLY SAFETY ALERT !!!") + for _, failure := range failures { + fmt.Fprintf(os.Stdout, " setup %s: %s\n", failure.SetupID, failure.Message) + } + fmt.Fprintln(os.Stdout, " No automatic remediation was attempted. Review the journal and rollback artifact before deciding whose change wins.") +} + +func emitSafetyWarnings(output io.Writer, warnings []app.SafetyWarning) { + if len(warnings) == 0 { + return + } + fmt.Fprintln(output, "\n!!! SAFETY WARNING !!!") + for _, warning := range warnings { + fmt.Fprintf(output, " [%s] %s\n", warning.Code, warning.Message) + } +} + +func emitRemovalImpacts(output io.Writer, impacts []app.RemovalImpact) { + if len(impacts) == 0 { + return + } + fmt.Fprintln(output, "\n!!! DESTRUCTIVE REMOVAL PREVIEW !!!") + for _, impact := range impacts { + fmt.Fprintf(output, " %s\n", impact.Message) + } +} + +func emitVerificationFailureResult( + cmd *cobra.Command, + v *viper.Viper, + value any, + failures []app.ApplyVerificationFailure, + runErr error, +) error { + if value == nil || len(failures) == 0 { + return runErr + } + if err := emitResult(cmd, v, value); err != nil { + return fmt.Errorf("%w; additionally failed to emit post-apply safety result: %v", runErr, err) + } + return runErr +} + +func summaryVerificationFailures(summary *app.Summary) []app.ApplyVerificationFailure { + if summary == nil { + return nil + } + return summary.ApplyVerificationFailures +} + +func externalIDVerificationFailures(summary *app.ExternalIDSummary) []app.ApplyVerificationFailure { + if summary == nil { + return nil + } + return summary.ApplyVerificationFailures +} + +func applyPlanVerificationFailures(summary *app.ApplyPlanSummary) []app.ApplyVerificationFailure { + if summary == nil { + return nil + } + return summary.ApplyVerificationFailures +} + func safeSyncPreflightError(summary *app.PreflightSummary) error { failures := make([]string, 0) for _, check := range summary.Checks { @@ -1779,6 +1852,8 @@ func confirmPost(post, yes bool, setupID string, stdin *os.File, stderr io.Write } func confirmApplyFromSummary(summary *app.Summary, stdin *os.File, stderr io.Writer) error { + emitSafetyWarnings(stderr, summary.SafetyWarnings) + emitRemovalImpacts(stderr, summary.RemovalImpacts) addedTotal, disabledTotal, removedTotal := 0, 0, 0 for _, setup := range summary.PlannedSetups { addedTotal += len(setup.AddedAccounts) diff --git a/cmd/awssync/main_test.go b/cmd/awssync/main_test.go index 47f2cbb..2c21400 100644 --- a/cmd/awssync/main_test.go +++ b/cmd/awssync/main_test.go @@ -145,6 +145,75 @@ func TestEmitSummaryHumanReportsSkippedNQERows(t *testing.T) { } } +func TestEmitSummaryHumanAndJSONExposePostPatchVerificationFailure(t *testing.T) { + failure := app.ApplyVerificationFailure{ + SetupID: "setup-a", + Status: "mismatch", + Message: "CRITICAL: post-PATCH verification detected unexplained Forward state; use rollback artifact /tmp/rollback.json", + RollbackOutput: "/tmp/rollback.json", + } + summary := &app.Summary{ + Output: "/tmp/plan.json", + ApplyVerificationFailures: []app.ApplyVerificationFailure{failure}, + } + human := captureStdout(t, func() { + if err := emitSummaryHuman(summary); err != nil { + t.Fatalf("emitSummaryHuman() error = %v", err) + } + }) + for _, want := range []string{ + "!!! CRITICAL POST-APPLY SAFETY ALERT !!!", + "setup setup-a", + "rollback artifact /tmp/rollback.json", + "No automatic remediation was attempted", + } { + if !strings.Contains(human, want) { + t.Fatalf("human output missing %q:\n%s", want, human) + } + } + data, err := json.Marshal(summary) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(data), `"apply_verification_failures"`) || + !strings.Contains(string(data), `"status":"mismatch"`) || + !strings.Contains(string(data), `"rollback_output":"/tmp/rollback.json"`) { + t.Fatalf("JSON output did not expose post-PATCH failure: %s", data) + } +} + +func TestEmitSummaryHumanHighlightsManifestWarningAndRemovalFraction(t *testing.T) { + summary := &app.Summary{ + Output: "/tmp/plan.json", + SafetyWarnings: []app.SafetyWarning{{ + Code: "manifest_matches_nqe_observation", + Message: "WARNING: manifest resembles NQE-observed inventory", + }}, + RemovalImpacts: []app.RemovalImpact{{ + SetupID: "setup-a", + RemovedCount: 968, + ConfiguredCount: 978, + RemovalPercent: 98.9775, + Message: "This removes 968 of 978 accounts (98.98%) from setup setup-a.", + }}, + } + output := captureStdout(t, func() { + if err := emitSummaryHuman(summary); err != nil { + t.Fatalf("emitSummaryHuman() error = %v", err) + } + }) + for _, want := range []string{ + "!!! SAFETY WARNING !!!", + "manifest_matches_nqe_observation", + "!!! DESTRUCTIVE REMOVAL PREVIEW !!!", + "This removes 968 of 978 accounts (98.98%)", + } { + if !strings.Contains(output, want) { + t.Fatalf("human output missing %q:\n%s", want, output) + } + } +} + func TestSyncAccountsDryRunReportsUnattendedDestructiveGate(t *testing.T) { patchCount := 0 server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -237,12 +306,18 @@ func TestSyncAccountsDryRunReportsUnattendedDestructiveGate(t *testing.T) { func TestApplyPlanCommandHonorsLocalYesFlag(t *testing.T) { patched := false + current := api.CloudAccount{Type: "AWS", Name: "setup-a", AssumeRoleInfos: []api.AssumeRoleInfo{}} server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { case r.Method == http.MethodGet && r.URL.Path == "/api/networks/network-1/cloudAccounts": - _, _ = w.Write([]byte(`[{"type":"AWS","name":"setup-a","assumeRoleInfos":[]}]`)) + _ = json.NewEncoder(w).Encode([]api.CloudAccount{current}) case r.Method == http.MethodPatch && r.URL.Path == "/api/networks/network-1/cloudAccounts/setup-a": patched = true + var payload api.PatchPayload + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Fatalf("decode PATCH: %v", err) + } + current = testCloudAccountFromPatchPayload(payload) _, _ = w.Write([]byte(`{}`)) default: http.NotFound(w, r) @@ -334,6 +409,15 @@ func TestHumanRecoveryOutputIncludesArtifactPaths(t *testing.T) { func TestSafeSyncRunsPreflightPreviewAndAdditiveApply(t *testing.T) { enabled := false patched := false + current := api.CloudAccount{ + Type: "AWS", + Name: "setup-a", + Regions: map[string]api.RegionMeta{"us-east-1": {TestInstant: 123}}, + AssumeRoleInfos: []api.AssumeRoleInfo{{ + RoleArn: "arn:aws:iam::111111111111:role/ForwardRole", + Enabled: false, + }}, + } processedAt := time.Now().UTC().Add(-time.Hour).Format(time.RFC3339) server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { @@ -347,11 +431,7 @@ func TestSafeSyncRunsPreflightPreviewAndAdditiveApply(t *testing.T) { _, _ = w.Write([]byte(`{"items":[{"Cloud Setup ID":"setup-a","Cloud Account ID":"111111111111","Cloud Account Name":"acct-a","Collected?":false}]}`)) case r.Method == http.MethodGet && r.URL.Path == "/api/networks/network-1/cloudAccounts": w.Header().Set("Content-Type", "application/json") - _, _ = fmt.Fprintf( - w, - `[{"type":"AWS","name":"setup-a","regions":{"us-east-1":{"testInstant":123}},"assumeRoleInfos":[{"roleArn":"arn:aws:iam::111111111111:role/ForwardRole","enabled":%t}]}]`, - enabled, - ) + _ = json.NewEncoder(w).Encode([]api.CloudAccount{current}) case r.Method == http.MethodPatch && r.URL.Path == "/api/networks/network-1/cloudAccounts/setup-a": var payload api.PatchPayload if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { @@ -361,6 +441,7 @@ func TestSafeSyncRunsPreflightPreviewAndAdditiveApply(t *testing.T) { t.Fatalf("safe-sync payload did not re-enable account: %#v", payload) } enabled = true + current = testCloudAccountFromPatchPayload(payload) patched = true _, _ = w.Write([]byte(`{}`)) default: @@ -408,6 +489,24 @@ func TestSafeSyncRunsPreflightPreviewAndAdditiveApply(t *testing.T) { func TestSafeSyncHandlesMultipleSetups(t *testing.T) { patched := map[string]bool{} + current := map[string]api.CloudAccount{ + "setup-a": { + Type: "AWS", + Name: "setup-a", + AssumeRoleInfos: []api.AssumeRoleInfo{{ + RoleArn: "arn:aws:iam::111111111111:role/ForwardRole", + Enabled: false, + }}, + }, + "setup-b": { + Type: "AWS", + Name: "setup-b", + AssumeRoleInfos: []api.AssumeRoleInfo{{ + RoleArn: "arn:aws:iam::222222222222:role/ForwardRole", + Enabled: false, + }}, + }, + } processedAt := time.Now().UTC().Add(-time.Hour).Format(time.RFC3339) server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { @@ -421,12 +520,14 @@ func TestSafeSyncHandlesMultipleSetups(t *testing.T) { {"Cloud Setup ID":"setup-b","Cloud Account ID":"222222222222","Collected?":false} ]}`)) case r.Method == http.MethodGet && r.URL.Path == "/api/networks/network-1/cloudAccounts": - _, _ = w.Write([]byte(`[ - {"type":"AWS","name":"setup-a","assumeRoleInfos":[{"roleArn":"arn:aws:iam::111111111111:role/ForwardRole","enabled":false}]}, - {"type":"AWS","name":"setup-b","assumeRoleInfos":[{"roleArn":"arn:aws:iam::222222222222:role/ForwardRole","enabled":false}]} - ]`)) + _ = json.NewEncoder(w).Encode([]api.CloudAccount{current["setup-a"], current["setup-b"]}) case r.Method == http.MethodPatch && strings.HasPrefix(r.URL.Path, "/api/networks/network-1/cloudAccounts/"): setupID := strings.TrimPrefix(r.URL.Path, "/api/networks/network-1/cloudAccounts/") + var payload api.PatchPayload + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Fatalf("decode PATCH: %v", err) + } + current[setupID] = testCloudAccountFromPatchPayload(payload) patched[setupID] = true _, _ = w.Write([]byte(`{}`)) default: @@ -1029,6 +1130,25 @@ func TestResolveSetupIDsForCLINonInteractiveAutoSelectsSingle(t *testing.T) { } } +func testCloudAccountFromPatchPayload(payload api.PatchPayload) api.CloudAccount { + regions := make(map[string]api.RegionMeta, len(payload.Regions)) + for region, instant := range payload.Regions { + regions[region] = api.RegionMeta{TestInstant: instant} + } + regionToProxy := make(map[string]string, len(payload.RegionToProxyServerID)) + for region, proxyID := range payload.RegionToProxyServerID { + regionToProxy[region] = proxyID + } + return api.CloudAccount{ + Type: payload.Type, + Name: payload.Name, + ProxyServerID: payload.ProxyServerID, + RegionToProxyServerID: regionToProxy, + Regions: regions, + AssumeRoleInfos: append([]api.AssumeRoleInfo(nil), payload.AssumeRoleInfos...), + } +} + func captureStdout(t *testing.T, fn func()) string { t.Helper() original := os.Stdout diff --git a/internal/app/account_manifest.go b/internal/app/account_manifest.go index 0c33f75..c69f175 100644 --- a/internal/app/account_manifest.go +++ b/internal/app/account_manifest.go @@ -11,6 +11,11 @@ import ( "github.com/forwardnetworks/aws-sync/internal/api" ) +const ( + manifestNQESimilarityThreshold = 0.95 + manifestConfiguredDifferenceThreshold = 0.20 +) + type AWSAccountManifestEntry struct { ID string `json:"id"` Name string `json:"name,omitempty"` @@ -95,6 +100,7 @@ func SyncAWSAccountManifest(ctx context.Context, cfg Config, accounts []AWSOrgan if err != nil { return nil, err } + warnings := inspectManifestInventoryShape(ctx, client, cfg, setupID, accounts, cloudAccounts) discovered, err := adaptManifestAccountsToSetupRows(accounts, setupID) if err != nil { return nil, err @@ -111,5 +117,121 @@ func SyncAWSAccountManifest(ctx context.Context, cfg Config, accounts []AWSOrgan } else { snapshot.SelectedSetupIDs = []SetupID{SetupID(setupID)} } - return runPlannedSyncFromSnapshot(ctx, cfg, client, snapshot, cloudAccounts) + summary, runErr := runPlannedSyncFromSnapshot(ctx, cfg, client, snapshot, cloudAccounts) + if summary != nil { + summary.SafetyWarnings = append(summary.SafetyWarnings, warnings...) + } + return summary, runErr +} + +func inspectManifestInventoryShape( + ctx context.Context, + client *api.Client, + cfg Config, + setupID string, + manifest []AWSOrganizationAccount, + cloudAccounts []api.CloudAccount, +) []SafetyWarning { + query, queryID, parameters := queryInputs(cfg) + queryResult, err := client.QueryAWSAccountsWithMetadata( + ctx, + cfg.NetworkID, + cfg.SnapshotID, + query, + queryID, + parameters, + []string{setupID}, + ) + if err != nil { + return []SafetyWarning{{ + Code: "manifest_nqe_shape_check_unavailable", + Message: fmt.Sprintf( + "WARNING: could not compare the reviewed manifest with current NQE-observed inventory: %v. This does not block sync-accounts; independently confirm the manifest is authoritative and was not generated from NQE output.", + err, + ), + }} + } + nqeSnapshot, err := parseNQESnapshotFromMapsWithOptions( + queryResult.Items, + nqeParseOptionsFromQueryResult(queryResult, cfg.AllowMalformedRows), + ) + if err != nil { + return []SafetyWarning{{ + Code: "manifest_nqe_shape_check_unavailable", + Message: fmt.Sprintf( + "WARNING: could not interpret current NQE-observed inventory for the manifest safeguard: %v. This does not block sync-accounts; independently confirm the manifest is authoritative and was not generated from NQE output.", + err, + ), + }} + } + + manifestIDs := makeStringSet(len(manifest)) + for _, account := range manifest { + manifestIDs[strings.TrimSpace(account.ID)] = struct{}{} + } + nqeIDs := makeStringSet(len(nqeSnapshot.DiscoveredAccounts)) + for _, account := range nqeSnapshot.DiscoveredAccounts { + if !account.SetupID.IsZero() && account.SetupID.String() != setupID { + continue + } + nqeIDs[account.AccountID.String()] = struct{}{} + } + configuredIDs := makeStringSet(0) + for _, cloudAccount := range cloudAccounts { + if strings.TrimSpace(cloudAccount.Name) != setupID { + continue + } + configuredIDs = makeStringSet(len(cloudAccount.AssumeRoleInfos)) + for _, account := range cloudAccount.AssumeRoleInfos { + if accountID := assumeRoleAccountID(account); accountID != "" { + configuredIDs[accountID] = struct{}{} + } + } + break + } + + similarity := setJaccardSimilarity(manifestIDs, nqeIDs) + configuredDifference := setDifferenceFraction(manifestIDs, configuredIDs) + if len(manifestIDs) == 0 || len(nqeIDs) == 0 || len(configuredIDs) == 0 || + similarity < manifestNQESimilarityThreshold || + configuredDifference < manifestConfiguredDifferenceThreshold { + return nil + } + return []SafetyWarning{{ + Code: "manifest_matches_nqe_observation", + Message: fmt.Sprintf( + "WARNING: the reviewed manifest account set matches current NQE-observed inventory by %.2f%% while differing from configured membership by %.2f%%. This is the signature of an NQE-derived manifest; NQE is observed and potentially partial, so confirm the manifest came from an independent authoritative source. This warning does not block the operation because a legitimate manifest can coincidentally match NQE.", + similarity*100, + configuredDifference*100, + ), + }} +} + +func makeStringSet(capacity int) map[string]struct{} { + return make(map[string]struct{}, capacity) +} + +func setJaccardSimilarity(left, right map[string]struct{}) float64 { + union := make(map[string]struct{}, len(left)+len(right)) + intersection := 0 + for value := range left { + union[value] = struct{}{} + if _, ok := right[value]; ok { + intersection++ + } + } + for value := range right { + union[value] = struct{}{} + } + if len(union) == 0 { + return 1 + } + return float64(intersection) / float64(len(union)) +} + +func setDifferenceFraction(candidate, configured map[string]struct{}) float64 { + if len(configured) == 0 { + return 0 + } + return 1 - setJaccardSimilarity(candidate, configured) } diff --git a/internal/app/account_manifest_test.go b/internal/app/account_manifest_test.go index 0d50ebc..9e60abf 100644 --- a/internal/app/account_manifest_test.go +++ b/internal/app/account_manifest_test.go @@ -51,6 +51,101 @@ func TestLoadAWSAccountManifestRejectsInvalidAndDuplicateIDs(t *testing.T) { } } +func TestSyncAWSAccountManifestWarnsWhenManifestMatchesNQEButNotConfiguredMembership(t *testing.T) { + server := newManifestSafeguardTestServer(t) + defer server.Close() + + summary, err := SyncAWSAccountManifest(context.Background(), Config{ + Host: server.URL, + Username: "user", + Password: "pass", + NetworkID: "network-1", + SetupIDs: []string{"setup-a"}, + APIPrefix: "/api", + Output: filepath.Join(t.TempDir(), "plan.json"), + }, manifestSafeguardAccounts()) + if err != nil { + t.Fatalf("SyncAWSAccountManifest() error = %v", err) + } + if len(summary.SafetyWarnings) != 1 || summary.SafetyWarnings[0].Code != "manifest_matches_nqe_observation" { + t.Fatalf("safety warnings = %#v", summary.SafetyWarnings) + } + if !strings.Contains(summary.SafetyWarnings[0].Message, "legitimate manifest can coincidentally match NQE") { + t.Fatalf("warning does not explain false-positive behavior: %q", summary.SafetyWarnings[0].Message) + } +} + +func TestSyncAWSAccountManifestSurfacesHighRemovalFraction(t *testing.T) { + server := newManifestSafeguardTestServer(t) + defer server.Close() + + summary, err := SyncAWSAccountManifest(context.Background(), Config{ + Host: server.URL, + Username: "user", + Password: "pass", + NetworkID: "network-1", + SetupIDs: []string{"setup-a"}, + APIPrefix: "/api", + Output: filepath.Join(t.TempDir(), "plan.json"), + MaxRemovals: 5, + MaxRemovalPercent: 100, + }, manifestSafeguardAccounts()) + if err != nil { + t.Fatalf("SyncAWSAccountManifest() error = %v", err) + } + if len(summary.RemovalImpacts) != 1 { + t.Fatalf("removal impacts = %#v", summary.RemovalImpacts) + } + impact := summary.RemovalImpacts[0] + if impact.RemovedCount != 3 || impact.ConfiguredCount != 5 || impact.RemovalPercent != 60 || + impact.Message != "This removes 3 of 5 accounts (60.00%) from setup setup-a." { + t.Fatalf("removal impact = %#v", impact) + } +} + +func newManifestSafeguardTestServer(t *testing.T) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodPost && r.URL.Path == "/api/nqe": + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"items":[ + {"Cloud Setup ID":"setup-a","Cloud Account ID":"111111111111","Cloud Account Name":"one","Collected?":true}, + {"Cloud Setup ID":"setup-a","Cloud Account ID":"222222222222","Cloud Account Name":"two","Collected?":true} + ]}`)) + case r.Method == http.MethodGet && r.URL.Path == "/api/networks/network-1/cloudAccounts": + infos := make([]api.AssumeRoleInfo, 0, 5) + for _, accountID := range []string{ + "111111111111", + "222222222222", + "333333333333", + "444444444444", + "555555555555", + } { + infos = append(infos, api.AssumeRoleInfo{ + AccountID: accountID, + RoleArn: "arn:aws:iam::" + accountID + ":role/ForwardRole", + Enabled: true, + }) + } + _ = json.NewEncoder(w).Encode([]api.CloudAccount{{ + Type: "AWS", + Name: "setup-a", + AssumeRoleInfos: infos, + }}) + default: + http.NotFound(w, r) + } + })) +} + +func manifestSafeguardAccounts() []AWSOrganizationAccount { + return []AWSOrganizationAccount{ + {ID: "111111111111", Name: "one"}, + {ID: "222222222222", Name: "two"}, + } +} + func TestRunAWSAccountManifestBuildsGovCloudRoleARNs(t *testing.T) { output := filepath.Join(t.TempDir(), "payload.json") summary, err := RunAWSAccountManifest(context.Background(), AWSOrganizationConfig{ @@ -109,13 +204,21 @@ func TestRunAWSAccountManifestRejectsPartitionRegionMismatch(t *testing.T) { func TestSyncAWSAccountManifestDryRunReportsRemovalAndApplyRequiresApproval(t *testing.T) { patchCount := 0 var patchedPayload api.PatchPayload + current := api.CloudAccount{ + Type: "AWS", + Name: "gov-prod", + Regions: map[string]api.RegionMeta{ + "us-gov-west-1": {TestInstant: 1}, + }, + AssumeRoleInfos: []api.AssumeRoleInfo{ + {AccountID: "111111111111", AccountName: "keep", RoleArn: "arn:aws-us-gov:iam::111111111111:role/ForwardRole", Enabled: true}, + {AccountID: "222222222222", AccountName: "remove", RoleArn: "arn:aws-us-gov:iam::222222222222:role/ForwardRole", Enabled: true}, + }, + } server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { case r.Method == http.MethodGet && r.URL.Path == "/api/networks/network-1/cloudAccounts": - _, _ = w.Write([]byte(`[{"type":"AWS","name":"gov-prod","regions":{"us-gov-west-1":{"testInstant":1}},"assumeRoleInfos":[ - {"accountId":"111111111111","accountName":"keep","roleArn":"arn:aws-us-gov:iam::111111111111:role/ForwardRole","enabled":true}, - {"accountId":"222222222222","accountName":"remove","roleArn":"arn:aws-us-gov:iam::222222222222:role/ForwardRole","enabled":true} - ]}]`)) + _ = json.NewEncoder(w).Encode([]api.CloudAccount{current}) case r.Method == http.MethodPatch && r.URL.Path == "/api/networks/network-1/cloudAccounts/gov-prod": if err := json.NewDecoder(r.Body).Decode(&patchedPayload); err != nil { t.Errorf("decode PATCH payload: %v", err) @@ -123,6 +226,7 @@ func TestSyncAWSAccountManifestDryRunReportsRemovalAndApplyRequiresApproval(t *t return } patchCount++ + current = testCloudAccountFromPatchPayload(patchedPayload) w.WriteHeader(http.StatusNoContent) default: http.NotFound(w, r) diff --git a/internal/app/apply_gateway.go b/internal/app/apply_gateway.go index 0bfd6b1..1f526c4 100644 --- a/internal/app/apply_gateway.go +++ b/internal/app/apply_gateway.go @@ -27,11 +27,28 @@ const ( // ApplyJournalEntry records the recoverable state of one planned setup. type ApplyJournalEntry struct { - SetupID string `json:"setup_id"` - Status ApplyStatus `json:"status"` - History []ApplyStatus `json:"history"` - HasChanges bool `json:"has_changes"` - Error string `json:"error,omitempty"` + SetupID string `json:"setup_id"` + Status ApplyStatus `json:"status"` + History []ApplyStatus `json:"history"` + HasChanges bool `json:"has_changes"` + Error string `json:"error,omitempty"` + PostPatchVerification *PostPatchVerificationRecord `json:"post_patch_verification,omitempty"` +} + +// PostPatchVerificationRecord records whether Forward's state matched the +// approved intent after a successful PATCH. +type PostPatchVerificationRecord struct { + Status string `json:"status"` + Message string `json:"message,omitempty"` +} + +// ApplyVerificationFailure is surfaced in command JSON as well as the durable +// journal so automation cannot mistake a sent-but-unverified PATCH for success. +type ApplyVerificationFailure struct { + SetupID string `json:"setup_id"` + Status string `json:"status"` + Message string `json:"message"` + RollbackOutput string `json:"rollback_output"` } // ApplyJournal is atomically rewritten after every setup disposition change. @@ -73,12 +90,13 @@ type ApplyAuthorization struct { // ApplyResult is returned even when an apply is partial or blocked. type ApplyResult struct { - PatchedCount int - Blocked bool - RollbackOutput string - RollbackSHA256 string - JournalOutput string - Journal ApplyJournal + PatchedCount int + Blocked bool + RollbackOutput string + RollbackSHA256 string + JournalOutput string + Journal ApplyJournal + VerificationFailures []ApplyVerificationFailure } // ApplyIntent is immutable after construction. Its state is private and every @@ -395,7 +413,7 @@ func GuardAndApply( _ = persistApplyJournal(&result) return result, err } - if !reflect.DeepEqual(setup.baseline, actual[setup.setupID]) { + if len(operatorControlledIntentDifferences(setup.baseline, actual[setup.setupID])) > 0 { conflict := fmt.Errorf( "selected Forward cloud setup state changed after planning for setup %s; no PATCH was sent for that setup; rerun the dry plan (the last-second re-read is only a weak mitigation because Forward provides no atomic compare-and-swap)", setup.setupID, @@ -410,6 +428,41 @@ func GuardAndApply( } result.PatchedCount++ setJournalStatus(entry, ApplyStatusApplied, "") + entry.PostPatchVerification = &PostPatchVerificationRecord{Status: "pending"} + + observedAccounts, err := client.CloudAccounts(ctx, state.networkID) + if err != nil { + failure := fmt.Errorf( + "CRITICAL: post-PATCH verification could not re-read setup %s: %w; Forward state is unknown and no automatic remediation was attempted; decide whose change should win, then use rollback artifact %s (SHA-256 %s) if rollback is appropriate", + setup.setupID, + err, + result.RollbackOutput, + result.RollbackSHA256, + ) + return failPostPatchVerification(result, entry, "read_failed", ApplyStatusFailed, failure) + } + observedPayloads, err := buildRollbackPayloads(observedAccounts, []string{setup.setupID}) + if err != nil { + failure := fmt.Errorf( + "CRITICAL: post-PATCH verification could not observe setup %s: %w; Forward state is unknown and no automatic remediation was attempted; decide whose change should win, then use rollback artifact %s (SHA-256 %s) if rollback is appropriate", + setup.setupID, + err, + result.RollbackOutput, + result.RollbackSHA256, + ) + return failPostPatchVerification(result, entry, "read_failed", ApplyStatusFailed, failure) + } + if differences := operatorControlledIntentDifferences(setup.target, observedPayloads[setup.setupID]); len(differences) > 0 { + failure := fmt.Errorf( + "CRITICAL: post-PATCH verification detected unexplained Forward state for setup %s (%s); a concurrent write may have been clobbered or interleaved; no automatic remediation was attempted; decide whose change should win, then use rollback artifact %s (SHA-256 %s) if rollback is appropriate", + setup.setupID, + strings.Join(differences, "; "), + result.RollbackOutput, + result.RollbackSHA256, + ) + return failPostPatchVerification(result, entry, "mismatch", ApplyStatusConflicted, failure) + } + entry.PostPatchVerification = &PostPatchVerificationRecord{Status: "matched"} if err := persistApplyJournal(&result); err != nil { return result, fmt.Errorf("setup %s was patched but its applied result could not be journaled: %w", setup.setupID, err) } @@ -417,6 +470,163 @@ func GuardAndApply( return result, nil } +func failPostPatchVerification( + result ApplyResult, + entry *ApplyJournalEntry, + verificationStatus string, + applyStatus ApplyStatus, + err error, +) (ApplyResult, error) { + entry.PostPatchVerification = &PostPatchVerificationRecord{ + Status: verificationStatus, + Message: err.Error(), + } + setJournalStatus(entry, applyStatus, err.Error()) + result.VerificationFailures = append(result.VerificationFailures, ApplyVerificationFailure{ + SetupID: entry.SetupID, + Status: verificationStatus, + Message: err.Error(), + RollbackOutput: result.RollbackOutput, + }) + if persistErr := persistApplyJournal(&result); persistErr != nil { + return result, fmt.Errorf("%w; additionally could not record post-PATCH verification in %s: %v", err, result.JournalOutput, persistErr) + } + return result, err +} + +func operatorControlledIntentDifferences(expected, observed api.PatchPayload) []string { + differences := make([]string, 0) + if expected.Type != observed.Type { + differences = append(differences, "setup type differs") + } + if expected.Name != observed.Name { + differences = append(differences, "setup name differs") + } + if expected.ProxyServerID != observed.ProxyServerID { + differences = append(differences, "proxy server differs") + } + if !stringMapsEqual(expected.RegionToProxyServerID, observed.RegionToProxyServerID) { + differences = append(differences, "region-to-proxy mapping differs") + } + if !stringSetsEqual(mapKeys(expected.Regions), mapKeys(observed.Regions)) { + differences = append(differences, "region membership differs") + } + + expectedAccounts, expectedAccountsInvalid := verificationAccountsByID(expected.AssumeRoleInfos) + observedAccounts, observedAccountsInvalid := verificationAccountsByID(observed.AssumeRoleInfos) + if expectedAccountsInvalid { + differences = append(differences, "approved intent contains duplicate or unidentifiable account entries") + } + if observedAccountsInvalid { + differences = append(differences, "observed state contains duplicate or unidentifiable account entries") + } + missing := make([]string, 0) + extra := make([]string, 0) + for accountID := range expectedAccounts { + if _, ok := observedAccounts[accountID]; !ok { + missing = append(missing, accountID) + } + } + for accountID := range observedAccounts { + if _, ok := expectedAccounts[accountID]; !ok { + extra = append(extra, accountID) + } + } + sort.Strings(missing) + sort.Strings(extra) + if len(missing) > 0 { + differences = append(differences, "missing account(s): "+strings.Join(missing, ", ")) + } + if len(extra) > 0 { + differences = append(differences, "extra account(s): "+strings.Join(extra, ", ")) + } + + shared := make([]string, 0, len(expectedAccounts)) + for accountID := range expectedAccounts { + if _, ok := observedAccounts[accountID]; ok { + shared = append(shared, accountID) + } + } + sort.Strings(shared) + for _, accountID := range shared { + expectedAccount := expectedAccounts[accountID] + observedAccount := observedAccounts[accountID] + fields := make([]string, 0, 4) + if expectedAccount.AccountName != observedAccount.AccountName { + fields = append(fields, "accountName") + } + if expectedAccount.RoleArn != observedAccount.RoleArn { + fields = append(fields, "roleArn") + } + if expectedAccount.ExternalID != observedAccount.ExternalID { + fields = append(fields, "externalId") + } + if expectedAccount.Enabled != observedAccount.Enabled { + fields = append(fields, "enabled") + } + if len(fields) > 0 { + differences = append(differences, fmt.Sprintf("account %s field(s) differ: %s", accountID, strings.Join(fields, ", "))) + } + } + return differences +} + +func verificationAccountsByID(accounts []api.AssumeRoleInfo) (map[string]api.AssumeRoleInfo, bool) { + result := make(map[string]api.AssumeRoleInfo, len(accounts)) + invalid := false + for _, account := range accounts { + accountID := assumeRoleAccountID(account) + if accountID == "" { + invalid = true + continue + } + if _, exists := result[accountID]; exists { + invalid = true + } + result[accountID] = account + } + return result, invalid +} + +func mapKeys[V any](values map[string]V) []string { + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + return keys +} + +func stringSetsEqual(left, right []string) bool { + if len(left) != len(right) { + return false + } + sort.Strings(left) + sort.Strings(right) + return reflect.DeepEqual(left, right) +} + +func stringMapsEqual(left, right map[string]string) bool { + if len(left) != len(right) { + return false + } + for key, value := range left { + rightValue, ok := right[key] + if !ok || rightValue != value { + return false + } + } + return true +} + +func journalEntryWasPatched(entry ApplyJournalEntry) bool { + for _, status := range entry.History { + if status == ApplyStatusApplied { + return true + } + } + return false +} + func validateApplyAuthorization(state *applyIntentState, authorization ApplyAuthorization) error { if !authorization.Approved { return fmt.Errorf("apply authorization is required") diff --git a/internal/app/apply_gateway_test.go b/internal/app/apply_gateway_test.go index 2fa8794..675c315 100644 --- a/internal/app/apply_gateway_test.go +++ b/internal/app/apply_gateway_test.go @@ -168,6 +168,147 @@ func TestGuardAndApplyDisableUsesDestructiveAuthorizationAndRemovalBudget(t *tes } } +func TestGuardAndApplyDetectsUnexplainedStateAfterPatch(t *testing.T) { + baseline := []api.AssumeRoleInfo{gatewayAssumeRole("111111111111", true)} + target := append(append([]api.AssumeRoleInfo(nil), baseline...), gatewayAssumeRole("222222222222", true)) + var ( + mu sync.Mutex + state = append([]api.AssumeRoleInfo(nil), baseline...) + ) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + mu.Lock() + accounts := append([]api.AssumeRoleInfo(nil), state...) + mu.Unlock() + _ = json.NewEncoder(w).Encode([]api.CloudAccount{{ + Type: "AWS", + Name: "setup-a", + AssumeRoleInfos: accounts, + }}) + case http.MethodPatch: + var payload api.PatchPayload + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Errorf("decode PATCH: %v", err) + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + mu.Lock() + state = append([]api.AssumeRoleInfo(nil), payload.AssumeRoleInfos...) + state = append(state, gatewayAssumeRole("333333333333", true)) + mu.Unlock() + w.WriteHeader(http.StatusNoContent) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + client, err := api.NewClient(server.URL, "/api", "alice", "secret", false, time.Second) + if err != nil { + t.Fatal(err) + } + intent := gatewayTestIntent(t, t.TempDir(), []gatewayTestSetup{{ + setupID: "setup-a", + baseline: baseline, + target: target, + changes: ChangeSet{Add: []AccountChange{{AccountID: AccountID("222222222222")}}}, + }}) + result, err := GuardAndApply(context.Background(), client, intent, ApplyAuthorization{ + PlanDigest: intent.Digest(), + Approved: true, + }) + if err == nil || !strings.Contains(err.Error(), "CRITICAL: post-PATCH verification") || + !strings.Contains(err.Error(), "extra account(s): 333333333333") || + !strings.Contains(err.Error(), result.RollbackOutput) { + t.Fatalf("GuardAndApply() error = %v; want prominent mismatch and rollback guidance", err) + } + if result.PatchedCount != 1 || len(result.VerificationFailures) != 1 { + t.Fatalf("verification result = %+v", result) + } + assertGatewayJournalEntry(t, result.Journal, "setup-a", ApplyStatusConflicted, + []ApplyStatus{ApplyStatusPlanned, ApplyStatusPending, ApplyStatusApplied, ApplyStatusConflicted}, + "concurrent write may have been clobbered or interleaved") + entry := journalEntry(&result.Journal, "setup-a") + if entry.PostPatchVerification == nil || entry.PostPatchVerification.Status != "mismatch" { + t.Fatalf("post-PATCH verification journal = %#v", entry.PostPatchVerification) + } + persisted := readGatewayJournal(t, result.JournalOutput) + persistedEntry := journalEntry(&persisted, "setup-a") + if persistedEntry.PostPatchVerification == nil || persistedEntry.PostPatchVerification.Status != "mismatch" { + t.Fatalf("persisted post-PATCH verification = %#v", persistedEntry.PostPatchVerification) + } +} + +func TestGuardAndApplyIgnoresNormalServerManagedPostPatchChanges(t *testing.T) { + baseline := []api.AssumeRoleInfo{gatewayAssumeRole("111111111111", true)} + target := append(append([]api.AssumeRoleInfo(nil), baseline...), gatewayAssumeRole("222222222222", true)) + getCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + getCount++ + accounts := baseline + testInstant := int64(456) + if getCount > 1 { + accounts = []api.AssumeRoleInfo{target[1], target[0]} + accounts[0].ErrorMsg = "server-managed collection status" + testInstant = 999 + } + _ = json.NewEncoder(w).Encode([]map[string]any{{ + "type": "AWS", + "name": "setup-a", + "regions": map[string]any{"us-east-1": map[string]any{"testInstant": testInstant}}, + "regionToProxyServerId": map[string]string{}, + "assumeRoleInfos": accounts, + "numVirtualizedDevices": 42, + }}) + case http.MethodPatch: + w.WriteHeader(http.StatusNoContent) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + client, err := api.NewClient(server.URL, "/api", "alice", "secret", false, time.Second) + if err != nil { + t.Fatal(err) + } + intent := gatewayTestIntent(t, t.TempDir(), []gatewayTestSetup{{ + setupID: "setup-a", + baseline: baseline, + target: target, + changes: ChangeSet{Add: []AccountChange{{AccountID: AccountID("222222222222")}}}, + }}) + for _, payloads := range []auditPayloads{intent.state.baselines, intent.state.targets} { + payload := payloads["setup-a"] + payload.Regions = map[string]int64{"us-east-1": 123} + payloads["setup-a"] = payload + } + intent.state.setups[0].baseline = clonePatchPayload(intent.state.baselines["setup-a"]) + intent.state.setups[0].target = clonePatchPayload(intent.state.targets["setup-a"]) + intent.state.digest, err = computeApplyIntentDigest(intent.state) + if err != nil { + t.Fatal(err) + } + + result, err := GuardAndApply(context.Background(), client, intent, ApplyAuthorization{ + PlanDigest: intent.Digest(), + Approved: true, + }) + if err != nil { + t.Fatalf("GuardAndApply() error = %v", err) + } + if len(result.VerificationFailures) != 0 { + t.Fatalf("normal server-managed changes flagged: %#v", result.VerificationFailures) + } + entry := journalEntry(&result.Journal, "setup-a") + if entry.PostPatchVerification == nil || entry.PostPatchVerification.Status != "matched" { + t.Fatalf("post-PATCH verification journal = %#v", entry.PostPatchVerification) + } +} + func TestApplyIntentDigestBindsBaselineSnapshotPolicyAndTarget(t *testing.T) { setup := gatewayTestSetup{ setupID: "setup-a", @@ -635,3 +776,18 @@ func gatewayAssumeRole(accountID string, enabled bool) api.AssumeRoleInfo { Enabled: enabled, } } + +func testCloudAccountFromPatchPayload(payload api.PatchPayload) api.CloudAccount { + regions := make(map[string]api.RegionMeta, len(payload.Regions)) + for region, instant := range payload.Regions { + regions[region] = api.RegionMeta{TestInstant: instant} + } + return api.CloudAccount{ + Type: payload.Type, + Name: payload.Name, + ProxyServerID: payload.ProxyServerID, + RegionToProxyServerID: cloneStringMap(payload.RegionToProxyServerID), + Regions: regions, + AssumeRoleInfos: append([]api.AssumeRoleInfo(nil), payload.AssumeRoleInfos...), + } +} diff --git a/internal/app/apply_plan.go b/internal/app/apply_plan.go index 9472c9d..09f97aa 100644 --- a/internal/app/apply_plan.go +++ b/internal/app/apply_plan.go @@ -30,16 +30,17 @@ type ApplyPlanConfig struct { } type ApplyPlanSummary struct { - Host string `json:"host"` - NetworkID string `json:"network_id"` - PlanPath string `json:"plan_path"` - PayloadSHA256 string `json:"payload_sha256"` - PlanDigest string `json:"plan_digest"` - RollbackOutput string `json:"rollback_output,omitempty"` - RollbackSHA256 string `json:"rollback_sha256,omitempty"` - ResultJournalOutput string `json:"result_journal_output"` - PatchedSetupCount int `json:"patched_setup_count"` - PatchedSetups []string `json:"patched_setups"` + Host string `json:"host"` + NetworkID string `json:"network_id"` + PlanPath string `json:"plan_path"` + PayloadSHA256 string `json:"payload_sha256"` + PlanDigest string `json:"plan_digest"` + RollbackOutput string `json:"rollback_output,omitempty"` + RollbackSHA256 string `json:"rollback_sha256,omitempty"` + ResultJournalOutput string `json:"result_journal_output"` + PatchedSetupCount int `json:"patched_setup_count"` + PatchedSetups []string `json:"patched_setups"` + ApplyVerificationFailures []ApplyVerificationFailure `json:"apply_verification_failures,omitempty"` } func ApplyPlan(ctx context.Context, cfg ApplyPlanConfig) (*ApplyPlanSummary, error) { @@ -166,21 +167,22 @@ func ApplyPlan(ctx context.Context, cfg ApplyPlanConfig) (*ApplyPlanSummary, err }) patchedSetups := make([]string, 0, applyResult.PatchedCount) for _, entry := range applyResult.Journal.Setups { - if entry.Status == ApplyStatusApplied { + if journalEntryWasPatched(entry) { patchedSetups = append(patchedSetups, entry.SetupID) } } summary := &ApplyPlanSummary{ - Host: cfg.Host, - NetworkID: cfg.NetworkID, - PlanPath: planPath, - PayloadSHA256: fmt.Sprintf("%x", sha256.Sum256(data)), - PlanDigest: intent.Digest(), - RollbackOutput: applyResult.RollbackOutput, - RollbackSHA256: applyResult.RollbackSHA256, - ResultJournalOutput: applyResult.JournalOutput, - PatchedSetupCount: applyResult.PatchedCount, - PatchedSetups: patchedSetups, + Host: cfg.Host, + NetworkID: cfg.NetworkID, + PlanPath: planPath, + PayloadSHA256: fmt.Sprintf("%x", sha256.Sum256(data)), + PlanDigest: intent.Digest(), + RollbackOutput: applyResult.RollbackOutput, + RollbackSHA256: applyResult.RollbackSHA256, + ResultJournalOutput: applyResult.JournalOutput, + PatchedSetupCount: applyResult.PatchedCount, + PatchedSetups: patchedSetups, + ApplyVerificationFailures: append([]ApplyVerificationFailure(nil), applyResult.VerificationFailures...), } if applyErr != nil { if applyResult.JournalOutput != "" { diff --git a/internal/app/apply_plan_test.go b/internal/app/apply_plan_test.go index 7cb1aa8..7dd14fc 100644 --- a/internal/app/apply_plan_test.go +++ b/internal/app/apply_plan_test.go @@ -18,6 +18,7 @@ import ( func TestApplyPlanPatchesReviewedPayload(t *testing.T) { var patched []string + current := api.CloudAccount{Type: "AWS", Name: "setup-a", AssumeRoleInfos: []api.AssumeRoleInfo{}} server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { user, pass, ok := r.BasicAuth() if !ok || user != "alice" || pass != "secret" { @@ -25,7 +26,7 @@ func TestApplyPlanPatchesReviewedPayload(t *testing.T) { return } if r.Method == http.MethodGet && r.URL.Path == "/api/networks/network-1/cloudAccounts" { - _, _ = w.Write([]byte(`[{"type":"AWS","name":"setup-a","assumeRoleInfos":[]}]`)) + _ = json.NewEncoder(w).Encode([]api.CloudAccount{current}) return } if r.Method != http.MethodPatch { @@ -33,6 +34,11 @@ func TestApplyPlanPatchesReviewedPayload(t *testing.T) { return } patched = append(patched, r.URL.Path) + var payload api.PatchPayload + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Fatalf("decode PATCH: %v", err) + } + current = testCloudAccountFromPatchPayload(payload) w.Header().Set("Content-Type", "application/json") _, _ = w.Write([]byte(`{}`)) })) @@ -125,6 +131,7 @@ func TestApplyPlanAcceptsPreBranchBinaryArtifacts(t *testing.T) { if err := json.NewDecoder(r.Body).Decode(&patched); err != nil { t.Fatalf("decode PATCH: %v", err) } + current = testCloudAccountFromPatchPayload(patched) _, _ = w.Write([]byte(`{}`)) default: http.NotFound(w, r) @@ -263,15 +270,25 @@ func TestApplyPlanDisableRequiresGatewayDestructiveAuthorization(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { patchCount := 0 + current := api.CloudAccount{ + Type: "AWS", + Name: "prod", + AssumeRoleInfos: []api.AssumeRoleInfo{ + {AccountID: "111111111111", RoleArn: "arn:aws:iam::111111111111:role/ForwardRole", Enabled: true}, + {AccountID: "222222222222", RoleArn: "arn:aws:iam::222222222222:role/ForwardRole", Enabled: true}, + }, + } server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { case r.Method == http.MethodGet && r.URL.Path == "/api/networks/network-1/cloudAccounts": - _, _ = w.Write([]byte(`[{"type":"AWS","name":"prod","assumeRoleInfos":[ - {"accountId":"111111111111","roleArn":"arn:aws:iam::111111111111:role/ForwardRole","enabled":true}, - {"accountId":"222222222222","roleArn":"arn:aws:iam::222222222222:role/ForwardRole","enabled":true} - ]}]`)) + _ = json.NewEncoder(w).Encode([]api.CloudAccount{current}) case r.Method == http.MethodPatch: patchCount++ + var payload api.PatchPayload + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Fatalf("decode PATCH: %v", err) + } + current = testCloudAccountFromPatchPayload(payload) w.WriteHeader(http.StatusNoContent) default: http.NotFound(w, r) diff --git a/internal/app/external_id.go b/internal/app/external_id.go index 841409c..e73d3ad 100644 --- a/internal/app/external_id.go +++ b/internal/app/external_id.go @@ -31,30 +31,31 @@ type ExternalIDConfig struct { } type ExternalIDSummary struct { - Host string `json:"host"` - NetworkID string `json:"network_id"` - SetupID string `json:"setup_id"` - Apply bool `json:"apply"` - Patched bool `json:"patched"` - Mode string `json:"mode"` - AccountCount int `json:"account_count"` - SelectedAccountCount int `json:"selected_account_count"` - ChangedAccountCount int `json:"changed_account_count"` - SetAccountCount int `json:"set_account_count"` - ClearedAccountCount int `json:"cleared_account_count"` - UnchangedAccountCount int `json:"unchanged_account_count"` - PreviousExternalIDConfigured bool `json:"previous_external_id_configured"` - PreviousExternalIDConsistent bool `json:"previous_external_id_consistent"` - TargetExternalIDConfigured bool `json:"target_external_id_configured"` - TargetExternalIDConsistent bool `json:"target_external_id_consistent"` - Changes []ExternalIDChange `json:"changes"` - Output string `json:"output"` - PayloadSHA256 string `json:"payload_sha256"` - PlanDigest string `json:"plan_digest,omitempty"` - RollbackOutput string `json:"rollback_output,omitempty"` - RollbackSHA256 string `json:"rollback_sha256,omitempty"` - ResultJournalOutput string `json:"result_journal_output,omitempty"` - Payload ExternalIDPatchPayload `json:"payload"` + Host string `json:"host"` + NetworkID string `json:"network_id"` + SetupID string `json:"setup_id"` + Apply bool `json:"apply"` + Patched bool `json:"patched"` + Mode string `json:"mode"` + AccountCount int `json:"account_count"` + SelectedAccountCount int `json:"selected_account_count"` + ChangedAccountCount int `json:"changed_account_count"` + SetAccountCount int `json:"set_account_count"` + ClearedAccountCount int `json:"cleared_account_count"` + UnchangedAccountCount int `json:"unchanged_account_count"` + PreviousExternalIDConfigured bool `json:"previous_external_id_configured"` + PreviousExternalIDConsistent bool `json:"previous_external_id_consistent"` + TargetExternalIDConfigured bool `json:"target_external_id_configured"` + TargetExternalIDConsistent bool `json:"target_external_id_consistent"` + Changes []ExternalIDChange `json:"changes"` + Output string `json:"output"` + PayloadSHA256 string `json:"payload_sha256"` + PlanDigest string `json:"plan_digest,omitempty"` + RollbackOutput string `json:"rollback_output,omitempty"` + RollbackSHA256 string `json:"rollback_sha256,omitempty"` + ResultJournalOutput string `json:"result_journal_output,omitempty"` + ApplyVerificationFailures []ApplyVerificationFailure `json:"apply_verification_failures,omitempty"` + Payload ExternalIDPatchPayload `json:"payload"` } type ExternalIDChange struct { @@ -303,6 +304,7 @@ func ChangeExternalID(ctx context.Context, cfg ExternalIDConfig) (*ExternalIDSum summary.RollbackOutput = applyResult.RollbackOutput summary.RollbackSHA256 = applyResult.RollbackSHA256 summary.ResultJournalOutput = applyResult.JournalOutput + summary.ApplyVerificationFailures = append([]ApplyVerificationFailure(nil), applyResult.VerificationFailures...) if applyErr != nil { if applyResult.JournalOutput != "" { return summary, fmt.Errorf("%w; apply result journal: %s", applyErr, applyResult.JournalOutput) diff --git a/internal/app/external_id_test.go b/internal/app/external_id_test.go index 5e89509..f9b6a3b 100644 --- a/internal/app/external_id_test.go +++ b/internal/app/external_id_test.go @@ -139,6 +139,11 @@ func TestChangeExternalIDConfirmsComputedDigestBeforeGatewayApply(t *testing.T) _ = json.NewEncoder(w).Encode([]api.CloudAccount{stored}) case http.MethodPatch: patchCount++ + var payload api.PatchPayload + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Fatalf("decode PATCH: %v", err) + } + stored = testCloudAccountFromPatchPayload(payload) w.WriteHeader(http.StatusNoContent) default: http.NotFound(w, r) @@ -175,8 +180,8 @@ func TestChangeExternalIDConfirmsComputedDigestBeforeGatewayApply(t *testing.T) if err != nil { t.Fatalf("ChangeExternalID() error = %v", err) } - if confirmationCount != 1 || getCount != 2 || patchCount != 1 { - t.Fatalf("confirmation/weak re-read/PATCH counts = %d/%d/%d, want 1/2/1", confirmationCount, getCount, patchCount) + if confirmationCount != 1 || getCount != 3 || patchCount != 1 { + t.Fatalf("confirmation/pre-read/post-read/PATCH counts = %d/%d/%d, want 1/3/1", confirmationCount, getCount, patchCount) } if summary.PlanDigest == "" || summary.ResultJournalOutput == "" { t.Fatalf("missing gateway digest or result journal: %#v", summary) diff --git a/internal/app/run.go b/internal/app/run.go index d3571aa..43d8629 100644 --- a/internal/app/run.go +++ b/internal/app/run.go @@ -130,48 +130,68 @@ func prepareReconcileConfig(cfg Config, planningInstant time.Time) Config { } type Summary struct { - Host string `json:"host"` - NetworkID string `json:"network_id"` - SnapshotID string `json:"snapshot_id,omitempty"` - Source string `json:"source,omitempty"` - AWSOrganizationID string `json:"aws_organization_id,omitempty"` - AWSManagementID string `json:"aws_management_account_id,omitempty"` - AWSAccountCount int `json:"aws_account_count,omitempty"` - AWSSkippedCount int `json:"aws_skipped_account_count,omitempty"` - CredentialMode string `json:"credential_mode,omitempty"` - Regions []string `json:"regions,omitempty"` - CreatePayloadReady bool `json:"create_payload_ready,omitempty"` - PostedSetupCount int `json:"posted_setup_count,omitempty"` - CreatePayload *api.CreateAWSPayload `json:"create_payload,omitempty"` - ManualAccountData []ManualAccountData `json:"manual_account_data,omitempty"` - QueryID string `json:"query_id,omitempty"` - QueryOverride bool `json:"query_override"` - QuerySetupParam string `json:"query_setup_param,omitempty"` - SetupIDs []string `json:"setup_ids,omitempty"` - SelectedSetupIDs []string `json:"selected_setup_ids,omitempty"` - Output string `json:"output"` - PayloadSHA256 string `json:"payload_sha256,omitempty"` - ManualOutput string `json:"manual_output,omitempty"` - ManualPayloadSHA256 string `json:"manual_payload_sha256,omitempty"` - ManualPayloads map[string][]api.AssumeRoleInfo `json:"manual_payloads,omitempty"` - RollbackOutput string `json:"rollback_output,omitempty"` - RollbackSHA256 string `json:"rollback_sha256,omitempty"` - PlanDigest string `json:"plan_digest,omitempty"` - ResultJournalOutput string `json:"result_journal_output,omitempty"` - ApplyJournal *ApplyJournal `json:"apply_journal,omitempty"` - Apply bool `json:"apply"` - FetchedItemCount int `json:"fetched_item_count"` - IgnoredNQEItemCount int `json:"ignored_nqe_item_count,omitempty"` - IgnoredNQEAccounts []AccountSummary `json:"ignored_nqe_accounts,omitempty"` - SkippedNQERows []MalformedNQERowSummary `json:"skipped_nqe_rows,omitempty"` - PlannedSetupCount int `json:"planned_setup_count"` - PatchedSetupCount int `json:"patched_setup_count"` - SkippedSetupCount int `json:"skipped_setup_count"` - PlannedSetups []SetupSummary `json:"planned_setups"` - SkippedSetups []SkipSummary `json:"skipped_setups,omitempty"` - CandidateCheck []CandidateCheck `json:"candidate_check,omitempty"` - RemovalBlocked bool `json:"removal_blocked,omitempty"` - RemovalBlockReason string `json:"removal_block_reason,omitempty"` + Host string `json:"host"` + NetworkID string `json:"network_id"` + SnapshotID string `json:"snapshot_id,omitempty"` + Source string `json:"source,omitempty"` + AWSOrganizationID string `json:"aws_organization_id,omitempty"` + AWSManagementID string `json:"aws_management_account_id,omitempty"` + AWSAccountCount int `json:"aws_account_count,omitempty"` + AWSSkippedCount int `json:"aws_skipped_account_count,omitempty"` + CredentialMode string `json:"credential_mode,omitempty"` + Regions []string `json:"regions,omitempty"` + CreatePayloadReady bool `json:"create_payload_ready,omitempty"` + PostedSetupCount int `json:"posted_setup_count,omitempty"` + CreatePayload *api.CreateAWSPayload `json:"create_payload,omitempty"` + ManualAccountData []ManualAccountData `json:"manual_account_data,omitempty"` + QueryID string `json:"query_id,omitempty"` + QueryOverride bool `json:"query_override"` + QuerySetupParam string `json:"query_setup_param,omitempty"` + SetupIDs []string `json:"setup_ids,omitempty"` + SelectedSetupIDs []string `json:"selected_setup_ids,omitempty"` + Output string `json:"output"` + PayloadSHA256 string `json:"payload_sha256,omitempty"` + ManualOutput string `json:"manual_output,omitempty"` + ManualPayloadSHA256 string `json:"manual_payload_sha256,omitempty"` + ManualPayloads map[string][]api.AssumeRoleInfo `json:"manual_payloads,omitempty"` + RollbackOutput string `json:"rollback_output,omitempty"` + RollbackSHA256 string `json:"rollback_sha256,omitempty"` + PlanDigest string `json:"plan_digest,omitempty"` + ResultJournalOutput string `json:"result_journal_output,omitempty"` + ApplyJournal *ApplyJournal `json:"apply_journal,omitempty"` + ApplyVerificationFailures []ApplyVerificationFailure `json:"apply_verification_failures,omitempty"` + Apply bool `json:"apply"` + FetchedItemCount int `json:"fetched_item_count"` + IgnoredNQEItemCount int `json:"ignored_nqe_item_count,omitempty"` + IgnoredNQEAccounts []AccountSummary `json:"ignored_nqe_accounts,omitempty"` + SkippedNQERows []MalformedNQERowSummary `json:"skipped_nqe_rows,omitempty"` + PlannedSetupCount int `json:"planned_setup_count"` + PatchedSetupCount int `json:"patched_setup_count"` + SkippedSetupCount int `json:"skipped_setup_count"` + PlannedSetups []SetupSummary `json:"planned_setups"` + SkippedSetups []SkipSummary `json:"skipped_setups,omitempty"` + CandidateCheck []CandidateCheck `json:"candidate_check,omitempty"` + RemovalBlocked bool `json:"removal_blocked,omitempty"` + RemovalBlockReason string `json:"removal_block_reason,omitempty"` + SafetyWarnings []SafetyWarning `json:"safety_warnings,omitempty"` + RemovalImpacts []RemovalImpact `json:"removal_impacts,omitempty"` +} + +// SafetyWarning is a non-blocking operator warning with a stable code for JSON +// consumers. +type SafetyWarning struct { + Code string `json:"code"` + Message string `json:"message"` +} + +// RemovalImpact makes destructive cardinality explicit instead of requiring +// operators or automation to derive it from account arrays. +type RemovalImpact struct { + SetupID string `json:"setup_id"` + RemovedCount int `json:"removed_count"` + ConfiguredCount int `json:"configured_count"` + RemovalPercent float64 `json:"removal_percent"` + Message string `json:"message"` } type CandidateCheck struct { @@ -975,6 +995,7 @@ func buildSummary( patchedCount int, ) *Summary { setupSummaries := make([]SetupSummary, 0, len(plan.Setups)) + removalImpacts := make([]RemovalImpact, 0, len(plan.Setups)) for _, setup := range plan.Setups { regions := make([]string, 0, len(setup.Payload.Regions)) for region := range setup.Payload.Regions { @@ -1013,6 +1034,26 @@ func buildSummary( UnchangedAccountCount: len(setup.UnchangedAccounts), ApplyStatus: ApplyStatusPlanned, }) + if removedCount := len(setup.RemovedAccounts); removedCount > 0 { + configuredCount := len(setup.CurrentAccounts) + percent := 0.0 + if configuredCount > 0 { + percent = float64(removedCount) * 100 / float64(configuredCount) + } + removalImpacts = append(removalImpacts, RemovalImpact{ + SetupID: setup.SetupID, + RemovedCount: removedCount, + ConfiguredCount: configuredCount, + RemovalPercent: percent, + Message: fmt.Sprintf( + "This removes %d of %d accounts (%.2f%%) from setup %s.", + removedCount, + configuredCount, + percent, + setup.SetupID, + ), + }) + } } return &Summary{ @@ -1041,6 +1082,7 @@ func buildSummary( PlannedSetups: setupSummaries, SkippedSetups: plan.Skips, CandidateCheck: plan.CandidateChecks, + RemovalImpacts: removalImpacts, } } @@ -1050,6 +1092,7 @@ func applyResultToSummary(summary *Summary, result ApplyResult) { summary.RollbackSHA256 = result.RollbackSHA256 summary.ResultJournalOutput = result.JournalOutput summary.RemovalBlocked = result.Blocked + summary.ApplyVerificationFailures = append([]ApplyVerificationFailure(nil), result.VerificationFailures...) journal := result.Journal summary.ApplyJournal = &journal entries := make(map[string]ApplyJournalEntry, len(result.Journal.Setups)) @@ -1063,7 +1106,7 @@ func applyResultToSummary(summary *Summary, result ApplyResult) { } summary.PlannedSetups[index].ApplyStatus = entry.Status summary.PlannedSetups[index].ApplyError = entry.Error - summary.PlannedSetups[index].Patched = entry.Status == ApplyStatusApplied + summary.PlannedSetups[index].Patched = journalEntryWasPatched(entry) } } diff --git a/internal/app/run_test.go b/internal/app/run_test.go index 9fdaaf6..951b974 100644 --- a/internal/app/run_test.go +++ b/internal/app/run_test.go @@ -820,6 +820,17 @@ func TestBuildPlanRejectsMalformedNQEAccountID(t *testing.T) { func TestRunWritesPayloadAndPatchesWhenApplyEnabled(t *testing.T) { var patched []string + current := api.CloudAccount{ + Type: "AWS", + Name: "setup-a", + Regions: map[string]api.RegionMeta{"us-east-1": {TestInstant: 123}}, + ProxyServerID: "proxy-1", + AssumeRoleInfos: []api.AssumeRoleInfo{{ + RoleArn: "arn:aws:iam::111111111111:role/ForwardRole", + ExternalID: "Org:99", + Enabled: true, + }}, + } server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { user, pass, ok := r.BasicAuth() if !ok || user != "alice" || pass != "secret" { @@ -832,9 +843,14 @@ func TestRunWritesPayloadAndPatchesWhenApplyEnabled(t *testing.T) { _, _ = w.Write([]byte(`{"items":[{"Setup ID":"setup-a","Cloud Account ID":"111111111111","Cloud Account Name":"acct-a"}]}`)) case r.Method == http.MethodGet && r.URL.Path == "/api/networks/network-1/cloudAccounts": w.Header().Set("Content-Type", "application/json") - _, _ = w.Write([]byte(`[{"name":"setup-a","regions":{"us-east-1":{"testInstant":123}},"assumeRoleInfos":[{"roleArn":"arn:aws:iam::111111111111:role/ForwardRole","externalId":"Org:99","enabled":true}],"proxyServerId":"proxy-1"}]`)) + _ = json.NewEncoder(w).Encode([]api.CloudAccount{current}) case r.Method == http.MethodPatch && r.URL.Path == "/api/networks/network-1/cloudAccounts/setup-a": patched = append(patched, r.URL.Path) + var payload api.PatchPayload + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Fatalf("decode PATCH: %v", err) + } + current = testCloudAccountFromPatchPayload(payload) w.Header().Set("Content-Type", "application/json") _, _ = w.Write([]byte(`{}`)) default: diff --git a/internal/webhook/durable_queue_test.go b/internal/webhook/durable_queue_test.go index 5a43c7f..917d17b 100644 --- a/internal/webhook/durable_queue_test.go +++ b/internal/webhook/durable_queue_test.go @@ -44,6 +44,9 @@ func TestAcceptedEventSurvivesCrashBeforeCompletion(t *testing.T) { // A real crash discards process-local admission signals and the channel. finishProcessAdmission(statePath, eventDedupeKey(event)) + if err := server.Close(); err != nil { + t.Fatalf("release crashed daemon state lock: %v", err) + } var recoveredRuns atomic.Int32 restarted := newDurableQueueServer(t, statePath, func(_ context.Context, cfg app.Config) (*app.Summary, error) { recoveredRuns.Add(1) @@ -78,6 +81,9 @@ func TestInFlightEventIsReplayedAfterCrash(t *testing.T) { t.Fatalf("markJobInFlight() exists=%v error=%v", exists, err) } finishProcessAdmission(statePath, eventDedupeKey(event)) + if err := server.Close(); err != nil { + t.Fatalf("release crashed daemon state lock: %v", err) + } restartedRuns := make(chan struct{}, 1) restarted := newDurableQueueServer(t, statePath, func(_ context.Context, cfg app.Config) (*app.Summary, error) { @@ -360,6 +366,11 @@ func newDurableQueueServer(t *testing.T, statePath string, run RunFunc) *Server if err != nil { t.Fatalf("New() error = %v", err) } + t.Cleanup(func() { + if err := server.Close(); err != nil { + t.Errorf("close webhook server: %v", err) + } + }) return server } diff --git a/internal/webhook/server.go b/internal/webhook/server.go index 7fdb227..691e592 100644 --- a/internal/webhook/server.go +++ b/internal/webhook/server.go @@ -57,6 +57,9 @@ type Server struct { maxAttempts int retryBaseDelay time.Duration retryMaxDelay time.Duration + stateLock *webhookStateLock + stateLockCloseOnce sync.Once + stateLockCloseErr error } func New(cfg Config) (*Server, error) { @@ -104,8 +107,13 @@ func New(cfg Config) (*Server, error) { return nil, err } cfg.StatePath = statePath + stateLock, err := lockWebhookState(statePath) + if err != nil { + return nil, err + } state, err := loadWebhookState(statePath) if err != nil { + _ = stateLock.close() return nil, err } @@ -124,14 +132,17 @@ func New(cfg Config) (*Server, error) { maxAttempts: webhookMaxAttempts, retryBaseDelay: webhookRetryBaseDelay, retryMaxDelay: webhookRetryMaxDelay, + stateLock: stateLock, } if err := server.recoverInFlightJobs(); err != nil { + _ = server.Close() return nil, err } return server, nil } func (s *Server) Run(ctx context.Context) error { + defer s.Close() mux := http.NewServeMux() mux.HandleFunc("/healthz", s.handleHealthz) mux.HandleFunc(s.cfg.Path, s.handleEvent) @@ -156,6 +167,18 @@ func (s *Server) Run(ctx context.Context) error { return nil } +// Close releases the interprocess ownership of the configured webhook state +// path. Run calls it automatically when the daemon stops. +func (s *Server) Close() error { + if s == nil { + return nil + } + s.stateLockCloseOnce.Do(func() { + s.stateLockCloseErr = s.stateLock.close() + }) + return s.stateLockCloseErr +} + func (s *Server) handleHealthz(w http.ResponseWriter, _ *http.Request) { s.stateMu.Lock() pendingDepth := len(s.state.PendingEvents) diff --git a/internal/webhook/server_test.go b/internal/webhook/server_test.go index a978126..1a6b803 100644 --- a/internal/webhook/server_test.go +++ b/internal/webhook/server_test.go @@ -91,6 +91,33 @@ func TestNewRequiresConfiguredNetworkWhenApplyEnabled(t *testing.T) { } } +func TestNewRefusesSecondDaemonForSameStateFile(t *testing.T) { + statePath := filepath.Join(t.TempDir(), "webhook-state.json") + cfg := Config{ + StatePath: statePath, + Logger: log.New(io.Discard, "", 0), + Run: func(context.Context, app.Config) (*app.Summary, error) { + return &app.Summary{}, nil + }, + App: app.Config{ + Host: "https://fwd.example", + Username: "alice", + Password: "secret", + }, + } + first, err := New(cfg) + if err != nil { + t.Fatalf("first New() error = %v", err) + } + defer first.Close() + + second, err := New(cfg) + if second != nil || err == nil || !strings.Contains(err.Error(), + "webhook state file "+statePath+" is already locked by another process; only one webhook daemon may use a state file") { + t.Fatalf("second New() server=%v error=%v; want state-file lock refusal", second, err) + } +} + func TestHandleEventQueuesExactSnapshot(t *testing.T) { var ( mu sync.Mutex @@ -230,6 +257,11 @@ func newTestServer(t *testing.T, cfg Config) (*httptest.Server, *Server) { if err != nil { t.Fatalf("New() error = %v", err) } + t.Cleanup(func() { + if err := server.Close(); err != nil { + t.Errorf("close webhook server: %v", err) + } + }) t.Cleanup(func() { waitForWebhookStateIdle(t, server) }) diff --git a/internal/webhook/state.go b/internal/webhook/state.go index 703f4a1..aaa4afc 100644 --- a/internal/webhook/state.go +++ b/internal/webhook/state.go @@ -3,16 +3,22 @@ package webhook import ( "context" "encoding/json" + "errors" "fmt" "os" "path/filepath" "strings" "sync" + "syscall" "time" "github.com/forwardnetworks/aws-sync/internal/api" ) +type webhookStateLock struct { + file *os.File +} + const ( previousWebhookStateVersion = 1 webhookStateVersion = 2 @@ -81,6 +87,48 @@ func resolveStatePath(configured string) (string, error) { return filepath.Join(configDir, "awssync", "webhook-state.json"), nil } +func lockWebhookState(path string) (*webhookStateLock, error) { + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, fmt.Errorf("create webhook state directory for lock: %w", err) + } + // The state itself is atomically replaced during persistence, so its inode + // cannot carry a lifetime lock. A stable sidecar represents ownership of the + // configured state path across those renames. + lockPath := path + ".lock" + file, err := os.OpenFile(lockPath, os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + return nil, fmt.Errorf("open webhook state lock %s: %w", lockPath, err) + } + if err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + _ = file.Close() + if errors.Is(err, syscall.EWOULDBLOCK) || errors.Is(err, syscall.EAGAIN) { + return nil, fmt.Errorf( + "webhook state file %s is already locked by another process; only one webhook daemon may use a state file", + path, + ) + } + return nil, fmt.Errorf("lock webhook state file %s: %w", path, err) + } + return &webhookStateLock{file: file}, nil +} + +func (lock *webhookStateLock) close() error { + if lock == nil || lock.file == nil { + return nil + } + unlockErr := syscall.Flock(int(lock.file.Fd()), syscall.LOCK_UN) + closeErr := lock.file.Close() + lock.file = nil + if unlockErr != nil { + return fmt.Errorf("unlock webhook state file: %w", unlockErr) + } + if closeErr != nil { + return fmt.Errorf("close webhook state lock: %w", closeErr) + } + return nil +} + func loadWebhookState(path string) (webhookState, error) { state := newWebhookState() data, err := os.ReadFile(path) From 4ad06c1d88178cc953f23ec44b59e82ba6e7091b Mon Sep 17 00:00:00 2001 From: captainpacket Date: Mon, 27 Jul 2026 11:28:14 -0500 Subject: [PATCH 2/2] docs: retarget release notes template at v3.1.0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The template is rendered verbatim for whatever tag triggers a release, so shipping v3.1.0 against the v3.0.0 text would republish that release's breaking changes as though they were new — telling operators to act on migrations they already completed, and burying what actually changed. Rewrites it for this release: additive hardening, no breaking changes. v2.x migration is now a pointer to docs/upgrading.md rather than the full text, and the known-limitation section states precisely what post-apply verification does and does not detect, so nobody reads it as closing the concurrent-edit gap. --- .github/RELEASE_NOTES_TEMPLATE.md | 112 +++++++++--------------------- 1 file changed, 34 insertions(+), 78 deletions(-) diff --git a/.github/RELEASE_NOTES_TEMPLATE.md b/.github/RELEASE_NOTES_TEMPLATE.md index 8aaf24d..c72de07 100644 --- a/.github/RELEASE_NOTES_TEMPLATE.md +++ b/.github/RELEASE_NOTES_TEMPLATE.md @@ -1,95 +1,51 @@ ## aws-sync {{VERSION}} -### Breaking changes +### This release -Read this section before upgrading. These changes fail closed, so an automated -deployment that does not act on them will stop working rather than degrade. +Additive safety hardening. **No breaking changes** — upgrading from v3.0.0 +requires no configuration changes. -**1. NQE-based account removal has been removed.** +- **Post-apply verification.** Each setup is re-read after a successful PATCH + and compared against the approved target. Unexplained divergence stops the + apply, is recorded as `conflicted` in the result journal, and points at the + rollback artifact. No automatic remediation is attempted: a corrective PATCH + would repeat the same overwrite, and deciding whose change wins is a human + call. -`--prune-missing` remains a recognized option so existing automation receives -an actionable error, but it always refuses before credentials, NQE, planning, -or PATCH work. NQE is observed snapshot inventory, not an authoritative account -manifest, so an absent row cannot prove an account should be deleted. Replace -NQE prune workflows with `sync-accounts` and a complete human-reviewed manifest. -Manifest removals still require `--allow-removals` and both nonzero removal -ceilings. + This narrows the concurrent-edit gap; it does not close it. See *Known + limitation* below for exactly what remains undetectable. -**2. `serve-webhook --apply` now requires authentication and an explicit network.** +- **Manifest quality warnings.** `sync-accounts` warns when a manifest closely + matches observed NQE inventory while diverging from configured membership — + the signature of a manifest derived from NQE, which reproduces by hand the + unsound inference retired in v3.0.0. It warns rather than blocks. -The server previously accepted unauthenticated requests when no webhook -credentials were configured, and let an event select any network or setup. It -now refuses to start in apply mode without all three of: +- **Removal impact is always stated.** Previews now say "removes 968 of 978 + accounts (98.98%)" regardless of what the configured ceilings permit. -```bash -awssync serve-webhook --apply --yes \ - --webhook-basic-username \ - --webhook-basic-password \ - --network-id -``` - -Configure Forward to send matching credentials (`awssync configure-webhook`). -Event scope is now intersected with configured scope: an event naming a -different network, or a setup outside `--setup-id`, is rejected with `403` -instead of being honored. - -The server also persists dedupe and snapshot-ordering state to -`$UserConfigDir/awssync/webhook-state.json`. Ensure the service user can write -that directory, or set `--webhook-state-file`. - -**3. Destructive applies in unattended contexts now require an explicit flag.** - -Forward's API exposes no compare-and-swap token, so a concurrent edit in the UI -cannot be detected before a full-list PATCH overwrites it. Removals and disables -requested without a human present now require `--allow-unattended-destructive`: +- **Webhook state locking.** A second daemon sharing a state file now fails + fast instead of silently corrupting dedupe records and snapshot watermarks. -```bash -awssync sync-accounts --apply --yes --allow-unattended-destructive ... -awssync apply-plan --allow-unattended-destructive ... # when removing/disabling -awssync serve-webhook --apply --yes --allow-unattended-destructive ... -``` - -`--yes` counts as unattended even in a terminal. The flag does not bypass -`--allow-removals`, evidence rules, or either removal ceiling — it is an -additional acknowledgement, not a replacement. `safe-sync` is unaffected, -being additive-only. Non-destructive applies are unaffected. +### Upgrading from v2.x -### Safety changes - -- NQE reconciliation is unconditionally additive. Pagination completeness checks remain to diagnose truncated observed data, but completeness no longer authorizes absence-based deletion. -- A malformed account ID now fails the plan instead of being silently skipped, since skipping rows is how a partial inventory becomes a deletion. Use `--allow-malformed-rows` to skip and report them; doing so marks the inventory incomplete and therefore blocks removals. -- Setting an account to `enabled: false` is now classified as destructive. It consumes the same authorization and removal ceilings as deletion, closing a path where `apply-plan` could disable every account in a setup without tripping any removal guard. -- All account-list writes go through one guarded apply path, enforced by a test that fails if any other caller appears. -- External ID rotation now writes a pre-change rollback artifact, re-reads before PATCH, and binds confirmation to the computed payload. -- A partial multi-setup apply reports per-setup disposition (applied, pending, conflicted, failed) and a result-journal path instead of a bare error. -- Planning is deterministic: preview and apply produce identical digests for identical inputs. -- Cross-setup account moves are refused. Sequential per-setup PATCHes cannot guarantee an account ends up in exactly one setup if the run fails midway. +v3.0.0 removed NQE-based account removal and made webhook authentication and +unattended destructive authorization mandatory. If you are coming from v2.x, +read [docs/upgrading.md](https://github.com/forwardnetworks/aws-sync/blob/main/docs/upgrading.md) +before installing — three previously-working invocations now fail closed. ### Known limitation Forward's cloud-account API provides no ETag, version field, or other -compare-and-swap token. A concurrent edit made in the Forward UI between this -tool's final read and its PATCH will be overwritten, and this is deterministic -rather than a narrow race. The pre-PATCH re-read narrows the window but does not -close it. Prefer `safe-sync` for routine work, and avoid unattended destructive -runs on setups that people also edit by hand. - -### Highlights - -- New `awssync safe-sync` command provides a one-command routine workflow: 24-hour snapshot freshness, preflight, compact preview, additive-only enforcement, one confirmation, rollback, and apply. -- `safe-sync` does not expose prune or removal controls and stops before PATCH if preflight is not ready or the reviewed payload changes. -- A zero-change `safe-sync` exits successfully without PATCHing Forward or refreshing setup test timestamps. -- The README is now novice-first, with the routine workflow, count definitions, expected output, common stop conditions, and a short decision diagram before expert features. -- A one-page routine operator handoff is available at `docs/routine-safe-sync.md`. -- NQE reconciliation is additive by default: configured accounts missing from the current NQE result remain in the setup, while discovered disabled accounts are re-enabled. -- NQE-based deletion is retired; `--prune-missing` returns an actionable refusal and reviewed manifest removal remains available through `sync-accounts`. -- Every apply writes a pre-change `.rollback.json` PATCH payload containing the account list and PATCHable setup fields, not a full setup backup, and verifies that the selected setup state has not changed before the first PATCH. -- CLI runs pin the latest processed snapshot so planning and apply use one immutable NQE inventory. -- Invalid NQE account-ID placeholders are ignored and reported instead of becoming AWS accounts. -- Human-readable output is now the default; use `--json` or `--format json` for automation. -- Regression coverage includes 0, 1, 10, half, and all-enabled account states; additive NQE and authoritative-manifest paths; multi-setup isolation; concurrent setup changes; rollback; and snapshot pinning. -- Per-account External ID selection and CSV workflows from v2.3.0 remain supported. -- Release assets remain available for Linux and macOS on amd64 and arm64 with SHA-256 checksums and GitHub build-provenance attestations. +compare-and-swap token. + +Post-apply verification detects unexplained divergence after a write. It cannot +detect a concurrent edit that arrived between this tool's final read and its +PATCH: the full-list write erases it, and the resulting state then matches the +approved intent exactly, leaving nothing observable. + +Combined with the pre-PATCH comparison, what remains uncovered is an edit +landing inside the PATCH execution window. Prefer `safe-sync` for routine work, +and avoid unattended destructive runs on setups that people also edit by hand. ### Download and verify