Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions doc/cli/cloud-security.md
Original file line number Diff line number Diff line change
Expand Up @@ -541,5 +541,6 @@ remediation runs.
digests, deadline, generation) and a confirmation token.
- Repeating the command with `--confirm <token>` 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.
13 changes: 11 additions & 2 deletions limacharlie/commands/cloudsec.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand All @@ -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
Expand Down Expand Up @@ -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))


Expand Down
18 changes: 18 additions & 0 deletions tests/unit/test_cloudsec_evidence_chain.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Loading