From 740ba53d3d9f5cbf8cf275d7dcbda852fe8b25f6 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 28 Sep 2026 15:59:18 +0000 Subject: [PATCH] ci(iam): fail closed on widened policies (PLAT-234) Compare new and removed SCPs, and fail when a Deny shrinks or a Condition changes. Run CheckNoNewAccess on bootstrap templates from the base repo. Install the base worktree's own dependencies and warn when analyzer credentials are skipped. Co-authored-by: Adam Moussa --- .github/workflows/ci.yaml | 9 +- scripts/check_iam_policies.py | 486 +++++++++++++++++++++++++++------- 2 files changed, 401 insertions(+), 94 deletions(-) diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index c7bc682..264f5f8 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -39,15 +39,20 @@ jobs: run: | git fetch origin main git worktree add --detach /tmp/iam-base origin/main - ln -s "$GITHUB_WORKSPACE/node_modules" /tmp/iam-base/node_modules + npm ci --prefix /tmp/iam-base (cd /tmp/iam-base && npx cdk synth org-governance -o /tmp/iam-base-out --quiet) - name: Configure AWS credentials + id: aws-creds continue-on-error: true uses: aws-actions/configure-aws-credentials@e1253824e5c10ff9df46874f81ed3ec929e19cfd # v6.3.0 with: role-to-assume: arn:aws:iam::328440206208:role/githubdeploy-seahaven-org-baseline-policy-check aws-region: us-east-1 # pragma: allowlist secret + - name: Note skipped analyzer credentials + if: steps.aws-creds.outcome != 'success' + run: echo "::warning title=Access Analyzer skipped::OIDC assume-role did not succeed, so ValidatePolicy and CheckNoNewAccess did not run. The skip stays until githubdeploy-seahaven-org-baseline-policy-check is deployed." + - name: Check IAM policies - run: python3 scripts/check_iam_policies.py --cdk-out cdk.out --base-cdk-out /tmp/iam-base-out --self-test + run: python3 scripts/check_iam_policies.py --cdk-out cdk.out --base-cdk-out /tmp/iam-base-out --base-repo /tmp/iam-base --self-test diff --git a/scripts/check_iam_policies.py b/scripts/check_iam_policies.py index 8f89b41..4aad5ae 100755 --- a/scripts/check_iam_policies.py +++ b/scripts/check_iam_policies.py @@ -6,12 +6,16 @@ Local invariants always run: - the plan refresh template grants no lambda write, including lambda:* When --cdk-out is set, synthesized service control policies are collected. -When AWS credentials can call sts:GetCallerIdentity, each document is sent to -IAM Access Analyzer ValidatePolicy. CheckNoNewAccess compares identity -policies to the base document. It rejects SERVICE_CONTROL_POLICY and any -document with no Allow, so SCP diffs compare Allow actions and Deny -NotAction lists instead. Missing credentials skip the AWS calls and still -pass the local invariants. +SCP diffs run locally. A new or renamed SCP is compared with an empty +baseline, and a base SCP missing from head is treated as deleted. The diff +fails when access widens: a new Allow, a smaller Deny, a larger Deny +NotAction list, or any Condition change. When AWS credentials can call +sts:GetCallerIdentity, each document is sent to IAM Access Analyzer +ValidatePolicy. A new SCP with a ValidatePolicy ERROR fails, because that +error cannot already exist on main. CheckNoNewAccess compares bootstrap +identity policies with the templates in --base-repo. It rejects +SERVICE_CONTROL_POLICY, so SCPs use the local diff. Missing credentials +skip the AWS calls, emit a warning, and still run the local checks. The substrate template is not scanned. Its afi plan role still has lambda:* until the import apply replaces it. @@ -104,14 +108,19 @@ def check_plan_refresh() -> None: print(f"invariant ok: {path.name} has no lambda write") -def bootstrap_documents() -> list[tuple[str, str, dict]]: - documents: list[tuple[str, str, dict]] = [] - for path in sorted(BOOTSTRAP.glob("*.json.tmpl")): +def bootstrap_documents(root: Path | None = None) -> dict[str, dict]: + bootstrap = (root or ROOT) / "lib" / "hcptf-bootstrap" + if not bootstrap.is_dir(): + fail(f"bootstrap template directory does not exist: {bootstrap}") + documents: dict[str, dict] = {} + for path in sorted(bootstrap.glob("*.json.tmpl")): # Trust documents are resource policies without a Resource element. # ValidatePolicy rejects that shape. The StringEquals invariant covers them. if path.name.startswith("trust-"): continue - documents.append((path.name, "IDENTITY_POLICY", load_json(path))) + documents[path.name] = load_json(path) + if not documents: + fail(f"no bootstrap policy templates in {bootstrap}") return documents @@ -141,7 +150,11 @@ def aws_available() -> bool: text=True, ) if result.returncode != 0: - print("AWS checks skipped: sts get-caller-identity failed") + print( + "::warning title=Access Analyzer skipped::sts get-caller-identity failed. " + "ValidatePolicy and CheckNoNewAccess did not run." + ) + print("AWS checks skipped: sts get-caller-identity failed", file=sys.stderr) return False identity = json.loads(result.stdout) print(f"AWS checks using account {identity.get('Account')}") @@ -221,43 +234,323 @@ def as_list(value: object) -> list[str]: return [str(item) for item in value] -def allow_surface(document: dict) -> set[tuple[str, ...]]: - """Actions a policy newly permits. - - CheckNoNewAccess rejects SERVICE_CONTROL_POLICY and any document with no - Allow statement. For those documents, new access is an added Allow or a - larger Deny NotAction list (the deny then skips more actions). - """ - surface: set[tuple[str, ...]] = set() +def statement_list(document: dict) -> list[dict]: statements = document.get("Statement", []) if isinstance(statements, dict): - statements = [statements] - for statement in statements: - sid = str(statement.get("Sid", "")) - effect = statement.get("Effect") - if effect == "Allow": - for action in statement_actions(statement): - for resource in as_list(statement.get("Resource")) or ["*"]: - surface.add(("allow", sid, action, resource)) - elif effect == "Deny" and "NotAction" in statement: - for action in as_list(statement.get("NotAction")): - surface.add(("not-action", sid, action)) - return surface + return [statements] + return list(statements) -def compare_documents(name: str, policy_type: str, existing: dict, new: dict) -> None: - if policy_type != "IDENTITY_POLICY": - added = sorted(allow_surface(new) - allow_surface(existing)) +def normalize(value: object) -> object: + if isinstance(value, dict): + return {key: normalize(value[key]) for key in sorted(value)} + if isinstance(value, list): + items = [normalize(item) for item in value] + return sorted(items, key=lambda item: json.dumps(item, sort_keys=True, default=str)) + return value + + +def stable(value: object) -> str: + return json.dumps(normalize(value), sort_keys=True, separators=(",", ":"), default=str) + + +def name_set(statement: dict, field: str) -> set[str]: + return set(as_list(statement.get(field))) + + +def drop_exact(old: list[dict], new: list[dict]) -> tuple[list[dict], list[dict]]: + used = [False] * len(new) + new_keys = [stable(item) for item in new] + remaining_old: list[dict] = [] + for statement in old: + key = stable(statement) + matched = False + for index, candidate in enumerate(new_keys): + if not used[index] and candidate == key: + used[index] = True + matched = True + break + if not matched: + remaining_old.append(statement) + remaining_new = [item for index, item in enumerate(new) if not used[index]] + return remaining_old, remaining_new + + +def pair_widenings(old: dict, new: dict) -> list[str]: + """Ways a matched statement grants more access than it used to.""" + reasons: list[str] = [] + label = str(new.get("Sid") or old.get("Sid") or "statement") + if old.get("Effect") != new.get("Effect"): + return [f"{label}: effect changed"] + if stable(old.get("Condition")) != stable(new.get("Condition")): + reasons.append(f"{label}: condition changed") + if stable(old.get("Principal")) != stable(new.get("Principal")): + reasons.append(f"{label}: principal changed") + if ("Action" in old) != ("Action" in new) or ("NotAction" in old) != ("NotAction" in new): + reasons.append(f"{label}: action form changed") + return reasons + if ("Resource" in old) != ("Resource" in new) or ("NotResource" in old) != ("NotResource" in new): + reasons.append(f"{label}: resource form changed") + return reasons + effect = old.get("Effect") + old_actions, new_actions = name_set(old, "Action"), name_set(new, "Action") + old_not, new_not = name_set(old, "NotAction"), name_set(new, "NotAction") + old_resources, new_resources = name_set(old, "Resource"), name_set(new, "Resource") + old_not_resources = name_set(old, "NotResource") + new_not_resources = name_set(new, "NotResource") + if effect == "Allow": + added = sorted(new_actions - old_actions) if added: - fail(f"SCP allows new access {name}: {added[:8]}") + reasons.append(f"{label}: allow actions added {added}") + removed_exceptions = sorted(old_not - new_not) + if removed_exceptions: + reasons.append(f"{label}: allow NotAction shrank {removed_exceptions}") + if new_resources - old_resources: + reasons.append(f"{label}: allow resources added") + if old_not_resources - new_not_resources: + reasons.append(f"{label}: allow NotResource shrank") + elif effect == "Deny": + removed = sorted(old_actions - new_actions) + if removed: + reasons.append(f"{label}: deny actions removed {removed}") + grown = sorted(new_not - old_not) + if grown: + reasons.append(f"{label}: deny NotAction grew {grown}") + if old_resources - new_resources: + reasons.append(f"{label}: deny resources removed") + if new_not_resources - old_not_resources: + reasons.append(f"{label}: deny NotResource grew") + return reasons + + +def scp_widenings(existing: dict, new: dict) -> list[str]: + """Access added relative to existing. + + CheckNoNewAccess rejects SERVICE_CONTROL_POLICY. A new Allow, a removed + or smaller Deny, a larger Deny NotAction list, or any Condition change + is new access. A new Deny, or a larger Deny Action list, is not. + """ + reasons: list[str] = [] + + def grouped(document: dict) -> tuple[dict[str, list[dict]], list[dict]]: + keyed: dict[str, list[dict]] = {} + loose: list[dict] = [] + for statement in statement_list(document): + sid = statement.get("Sid") + if isinstance(sid, str) and sid: + keyed.setdefault(sid, []).append(statement) + else: + loose.append(statement) + return keyed, loose + + old_keyed, old_loose = grouped(existing) + new_keyed, new_loose = grouped(new) + for sid in sorted(set(old_keyed) | set(new_keyed)): + old_group, new_group = drop_exact(old_keyed.get(sid, []), new_keyed.get(sid, [])) + count = min(len(old_group), len(new_group)) + for old, statement in zip(old_group[:count], new_group[:count]): + reasons.extend(pair_widenings(old, statement)) + for statement in old_group[count:]: + if statement.get("Effect") == "Deny": + reasons.append(f"{sid}: deny statement removed") + for statement in new_group[count:]: + if statement.get("Effect") == "Allow": + reasons.append(f"{sid}: allow statement added") + old_loose, new_loose = drop_exact(old_loose, new_loose) + for statement in old_loose: + if statement.get("Effect") == "Deny": + reasons.append("deny statement removed") + for statement in new_loose: + if statement.get("Effect") == "Allow": + reasons.append("allow statement added") + return reasons + + +def compare_documents( + name: str, + policy_type: str, + existing: dict, + new: dict, + checker=check_no_new_access, +) -> None: + if policy_type != "IDENTITY_POLICY": + reasons = scp_widenings(existing, new) + if reasons: + fail(f"SCP allows new access {name}: {reasons[:8]}") print(f"SCP allow check ok: {name}") return - result = check_no_new_access(existing, new, policy_type) + result = checker(existing, new, policy_type) if result != "PASS": fail(f"CheckNoNewAccess {name}: {result or 'empty result'}") print(f"CheckNoNewAccess ok: {name}") +def expect_widening(existing: dict, new: dict, label: str) -> None: + if not scp_widenings(existing, new): + fail(f"SCP allow check missed {label}") + + +def expect_same(existing: dict, new: dict, label: str) -> None: + reasons = scp_widenings(existing, new) + if reasons: + fail(f"SCP allow check flagged {label}: {reasons}") + + +def local_self_test() -> None: + deny_only = { + "Statement": [ + {"Sid": "A", "Effect": "Deny", "NotAction": ["iam:*"], "Resource": "*"} + ] + } + larger_not_action = { + "Statement": [ + { + "Sid": "A", + "Effect": "Deny", + "NotAction": ["iam:*", "s3:*"], + "Resource": "*", + } + ] + } + expect_widening(deny_only, larger_not_action, "a larger Deny NotAction list") + expect_widening(deny_only, {"Statement": []}, "a removed Deny") + expect_widening( + { + "Statement": [ + { + "Sid": "A", + "Effect": "Deny", + "Action": ["iam:CreateRole", "iam:DeleteRole"], + "Resource": "*", + } + ] + }, + { + "Statement": [ + {"Sid": "A", "Effect": "Deny", "Action": "iam:CreateRole", "Resource": "*"} + ] + }, + "a smaller Deny Action list", + ) + expect_widening( + { + "Statement": [ + { + "Sid": "A", + "Effect": "Allow", + "Action": "s3:GetObject", + "Resource": "*", + "Condition": {"StringEquals": {"aws:RequestedRegion": "home"}}, + } + ] + }, + { + "Statement": [ + {"Sid": "A", "Effect": "Allow", "Action": "s3:GetObject", "Resource": "*"} + ] + }, + "a removed Allow Condition", + ) + expect_widening( + { + "Statement": [ + { + "Sid": "A", + "Effect": "Deny", + "Action": "iam:CreateRole", + "Resource": "*", + "Condition": { + "ArnNotLike": { + "aws:PrincipalArn": "arn:aws:iam::*:role/OrganizationAccountAccessRole" + } + }, + } + ] + }, + { + "Statement": [ + { + "Sid": "A", + "Effect": "Deny", + "Action": "iam:CreateRole", + "Resource": "*", + "Condition": { + "ArnNotLike": {"aws:PrincipalArn": "arn:aws:iam::*:role/other"} + }, + } + ] + }, + "a changed Deny Condition", + ) + expect_widening( + {"Statement": []}, + { + "Statement": [ + {"Sid": "Wide", "Effect": "Allow", "Action": "*", "Resource": "*"} + ] + }, + "a new SCP Allow *", + ) + expect_widening( + { + "Statement": [ + {"Sid": "A", "Effect": "Allow", "Action": "s3:GetObject", "Resource": "*"} + ] + }, + { + "Statement": [ + {"Sid": "A", "Effect": "Allow", "NotAction": "iam:CreateRole", "Resource": "*"} + ] + }, + "an Allow rewritten as NotAction", + ) + expect_same(deny_only, deny_only, "an identical document") + expect_same( + { + "Statement": [ + {"Sid": "A", "Effect": "Deny", "Action": "iam:CreateRole", "Resource": "*"} + ] + }, + { + "Statement": [ + { + "Sid": "A", + "Effect": "Deny", + "Action": ["iam:CreateRole", "iam:DeleteRole"], + "Resource": "*", + } + ] + }, + "a larger Deny Action list", + ) + plan = load_json(BOOTSTRAP / "plan-refresh-policy.json.tmpl") + widened_plan = json.loads(json.dumps(plan)) + widened_plan["Statement"].append( + {"Sid": "Wide", "Effect": "Allow", "Action": "s3:PutObject", "Resource": "*"} + ) + try: + compare_documents( + "plan-refresh-policy.json.tmpl", + "IDENTITY_POLICY", + plan, + widened_plan, + checker=lambda _existing, _new, _policy_type: "FAIL", + ) + except CheckFailure as exc: + if "CheckNoNewAccess" not in str(exc): + raise + else: + fail("widened plan-refresh template was accepted") + compare_documents( + "plan-refresh-policy.json.tmpl", + "IDENTITY_POLICY", + plan, + plan, + checker=lambda _existing, _new, _policy_type: "PASS", + ) + print("self-test ok: SCP widenings and plan-refresh comparison") + + def self_test() -> None: existing = { "Version": "2012-10-17", @@ -279,83 +572,92 @@ def self_test() -> None: fail("self-test expected FAIL when s3:PutObject is added") if check_no_new_access(existing, existing, "IDENTITY_POLICY") != "PASS": fail("self-test expected PASS for an identical policy") + plan = load_json(BOOTSTRAP / "plan-refresh-policy.json.tmpl") + widened_plan = json.loads(json.dumps(plan)) + widened_plan["Statement"].append( + {"Effect": "Allow", "Action": "s3:PutObject", "Resource": "*"} + ) + if check_no_new_access(plan, widened_plan, "IDENTITY_POLICY") != "FAIL": + fail("self-test expected FAIL when plan refresh gains s3:PutObject") print("self-test ok: CheckNoNewAccess distinguishes added access") +def validate_document( + name: str, + policy_type: str, + document: dict, + base_documents: dict[str, dict], + require_identity_base: bool, +) -> None: + head_errors = error_findings(name, policy_type, document) + if name in base_documents: + base_errors = set(error_findings(name, policy_type, base_documents[name])) + introduced = [item for item in head_errors if item not in base_errors] + if introduced: + fail(f"ValidatePolicy new ERROR {name}: {describe_findings(introduced)}") + if head_errors: + print(f"ValidatePolicy existing ERROR {name}: {describe_findings(head_errors)}") + else: + print(f"ValidatePolicy ok: {name} ({policy_type})") + if policy_type == "IDENTITY_POLICY": + compare_documents(name, policy_type, base_documents[name], document) + return + if head_errors: + fail(f"ValidatePolicy ERROR {name}: {describe_findings(head_errors)}") + if policy_type == "IDENTITY_POLICY" and require_identity_base: + fail(f"new bootstrap policy has no base document: {name}") + print(f"ValidatePolicy ok: {name} ({policy_type})") + + def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--cdk-out", type=Path) parser.add_argument("--base-cdk-out", type=Path) + parser.add_argument("--base-repo", type=Path) parser.add_argument("--self-test", action="store_true") args = parser.parse_args() try: check_trust_templates() check_plan_refresh() - deny_only = { - "Statement": [ - {"Sid": "A", "Effect": "Deny", "NotAction": ["iam:*"], "Resource": "*"} - ] - } - widened = { - "Statement": [ - { - "Sid": "A", - "Effect": "Deny", - "NotAction": ["iam:*", "s3:*"], - "Resource": "*", - }, - { - "Sid": "B", - "Effect": "Allow", - "Action": "s3:GetObject", - "Resource": "*", - }, - ] - } - if not (allow_surface(widened) - allow_surface(deny_only)): - fail("SCP allow check missed a widened NotAction and a new Allow") - if allow_surface(deny_only) - allow_surface(deny_only): - fail("SCP allow check flagged an identical document") - documents: list[tuple[str, str, dict]] = list(bootstrap_documents()) + local_self_test() + documents: list[tuple[str, str, dict]] = [ + (name, "IDENTITY_POLICY", document) + for name, document in bootstrap_documents().items() + ] base_documents: dict[str, dict] = {} + if args.base_repo: + base_documents.update(bootstrap_documents(args.base_repo)) + head_scps: dict[str, dict] = {} if args.cdk_out: - for name, document in synthesized_scps(args.cdk_out).items(): + head_scps = synthesized_scps(args.cdk_out) + for name, document in head_scps.items(): documents.append((name, "SERVICE_CONTROL_POLICY", document)) + base_scps: dict[str, dict] = {} if args.base_cdk_out: - base_documents = synthesized_scps(args.base_cdk_out) + base_scps = synthesized_scps(args.base_cdk_out) + base_documents.update(base_scps) + empty: dict = {"Statement": []} + for name, document in head_scps.items(): + compare_documents( + name, + "SERVICE_CONTROL_POLICY", + base_scps.get(name, empty), + document, + ) + for name in sorted(set(base_scps) - set(head_scps)): + compare_documents(name, "SERVICE_CONTROL_POLICY", base_scps[name], empty) if aws_available(): if args.self_test: self_test() for name, policy_type, document in documents: - head_errors = error_findings(name, policy_type, document) - if name in base_documents: - base_errors = set(error_findings(name, policy_type, base_documents[name])) - introduced = [item for item in head_errors if item not in base_errors] - if introduced: - fail( - f"ValidatePolicy new ERROR {name}: {describe_findings(introduced)}" - ) - if head_errors: - print( - f"ValidatePolicy existing ERROR {name}: {describe_findings(head_errors)}" - ) - else: - print(f"ValidatePolicy ok: {name} ({policy_type})") - compare_documents( - name, - policy_type, - base_documents[name], - document, - ) - elif head_errors and policy_type == "IDENTITY_POLICY": - fail(f"ValidatePolicy ERROR {name}: {describe_findings(head_errors)}") - elif head_errors: - print( - f"ValidatePolicy existing ERROR {name}: {describe_findings(head_errors)}" - ) - else: - print(f"ValidatePolicy ok: {name} ({policy_type})") + validate_document( + name, + policy_type, + document, + base_documents, + require_identity_base=args.base_repo is not None, + ) print(f"checked {len(documents)} documents") except CheckFailure as exc: print(f"FAIL: {exc}", file=sys.stderr)