From b7b6ea0c8f7c38a6c72a091ee9e6b3a20f3c2860 Mon Sep 17 00:00:00 2001 From: Maxime Lamothe-Brassard Date: Wed, 23 Sep 2026 15:38:03 -0700 Subject: [PATCH] cloudsec remediation: fail cleanly on a run without a reviewable generation A run whose generation is missing or not an integer now fails with a clear message instead of a TypeError, and the refusal text names the decision correctly (approved, rejected, cancelled). The docs state that the confirmation token is a review step and the server is the gate. Co-Authored-By: Claude Opus 5.5 (1M context) --- doc/cli/cloud-security.md | 5 +++-- limacharlie/commands/cloudsec.py | 13 +++++++++++-- tests/unit/test_cloudsec_evidence_chain.py | 18 ++++++++++++++++++ 3 files changed, 32 insertions(+), 4 deletions(-) diff --git a/doc/cli/cloud-security.md b/doc/cli/cloud-security.md index 9ed5577f..4e4f9dd9 100644 --- a/doc/cli/cloud-security.md +++ b/doc/cli/cloud-security.md @@ -541,5 +541,6 @@ remediation runs. digests, deadline, generation) and a confirmation token. - Repeating the command with `--confirm ` sends the decision. The token is derived from the run's generation and target digest, so it stops matching when - the run or its targets change. The server applies the same check again and - refuses a stale decision. + the run or its targets change. The token is a review step, not a secret. The + server is the gate: it requires `cloudsec.respond` and refuses a decision whose + generation or target digest no longer matches the run. diff --git a/limacharlie/commands/cloudsec.py b/limacharlie/commands/cloudsec.py index eab571ce..caba8a19 100644 --- a/limacharlie/commands/cloudsec.py +++ b/limacharlie/commands/cloudsec.py @@ -3614,6 +3614,9 @@ def remediation_create(ctx, finding_id, action, idempotency_key) -> None: } +_DECIDED = {"approve": "approved", "reject": "rejected", "cancel": "cancelled"} + + def _decision_token(oid: str, run: dict[str, Any], decision: str) -> str: """Bind a confirmation to exactly what was reviewed. @@ -3634,7 +3637,13 @@ def _remediation_decide(ctx, run_id: str, decision: str, confirm: str | None) -> run = detail.get("run") or {} state = run.get("state", "") if state not in _DECIDABLE[decision]: - raise click.ClickException(f"run {run_id} is '{state}'; it cannot be {decision}d now") + raise click.ClickException(f"run {run_id} is '{state}'; it cannot be {_DECIDED[decision]} now") + generation = run.get("generation") + if isinstance(generation, bool) or not isinstance(generation, int) or generation < 0 or \ + (decision == "approve" and not run.get("scope_digest")): + raise click.ClickException( + f"run {run_id} does not carry a reviewable generation and target digest; " + "it cannot be decided from here") token = _decision_token(cs.oid, run, decision) if not confirm: # Review only: nothing is sent. The operator reads exactly what the decision @@ -3663,7 +3672,7 @@ def _remediation_decide(ctx, run_id: str, decision: str, confirm: str | None) -> raise click.ClickException( "confirmation does not match the run as it is now (it changed since the review, " "or the token is for another run or decision); review it again") - _output(ctx, cs.decide_remediation(run_id, decision, int(run.get("generation", -1)), + _output(ctx, cs.decide_remediation(run_id, decision, generation, run.get("scope_digest") if decision == "approve" else None)) diff --git a/tests/unit/test_cloudsec_evidence_chain.py b/tests/unit/test_cloudsec_evidence_chain.py index 10eda1b8..454b7385 100644 --- a/tests/unit/test_cloudsec_evidence_chain.py +++ b/tests/unit/test_cloudsec_evidence_chain.py @@ -297,3 +297,21 @@ def test_a_run_that_cannot_take_the_decision_is_refused_locally(decision, state) get_remediation=_run(state=state)) assert result.exit_code != 0 inst.decide_remediation.assert_not_called() + + +@pytest.mark.parametrize("generation", [None, "3", -1, True]) +def test_a_run_without_a_reviewable_generation_fails_cleanly(generation): + run = _run() + run["result"]["run"]["generation"] = generation + for args in (["cloudsec", "remediation", "cancel", RID], ["cloudsec", "remediation", "cancel", RID, "--confirm", "0" * 16]): + result, inst = _invoke(args, get_remediation=run) + assert result.exit_code == 1, result.output + assert "reviewable generation" in result.output + assert not isinstance(result.exception, TypeError) + inst.decide_remediation.assert_not_called() + + +@pytest.mark.parametrize("decision,word", [("approve", "approved"), ("reject", "rejected"), ("cancel", "cancelled")]) +def test_the_refusal_names_the_decision_correctly(decision, word): + result, _ = _invoke(["cloudsec", "remediation", decision, RID], get_remediation=_run(state="verified")) + assert f"cannot be {word} now" in result.output