"""Tests for the IAM policy baseline (REQ-116, v1.11 Phase 56). Asserts that terraform/bootstrap/spike_runner_policy.json grants the minimum permissions required for CAP-017..022 + the Cost Explorer query (REQ-119). This is the regression-testable surface for the IAM re- bootstrap: any future drift (a permission removed) surfaces as a test failure at milestone COMPLETE (D-091 gate). """ import json import sys from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) ROOT = Path(__file__).resolve().parent.parent POLICY_PATH = ROOT / "terraform" / "bootstrap" / "spike_runner_policy.json" REQUIRED_ACTIONS = { "cloudfront": [ "cloudfront:Create*", "cloudfront:Get*", "cloudfront:List*", "cloudfront:Update*", "cloudfront:Delete*", ], "waf": [ "wafv2:Create*", "wafv2:Get*", "wafv2:List*", "wafv2:Update*", "wafv2:Delete*", ], "lambda": [ "lambda:Create*", "lambda:Get*", "lambda:List*", "lambda:Update*", "lambda:Delete*", "lambda:InvokeFunction", "lambda:InvokeFunctionUrl", ], "dynamodb_contracts": [ "dynamodb:Create*", "dynamodb:Describe*", "dynamodb:Get*", "dynamodb:Put*", "dynamodb:Update*", "dynamodb:Delete*", "dynamodb:Query", "dynamodb:Scan", ], "secretsmanager": [ "secretsmanager:GetSecretValue", "secretsmanager:DescribeSecret", "secretsmanager:ListSecrets", ], "sns": [ "sns:CreateTopic", "sns:Publish", "sns:ListTopics", ], "cost_explorer": [ "ce:GetCostAndUsage", "ce:GetCostForecast", "ce:GetCostAndUsageWithResources", "ce:GetDimensionValues", "ce:GetTags", ], "kms": [ "kms:CreateKey", "kms:CreateAlias", "kms:Describe*", "kms:Get*", "kms:List*", "kms:ScheduleKeyDeletion", ], "iam_oidc": [ "iam:CreateOpenIDConnectProvider", "iam:GetOpenIDConnectProvider", "iam:ListOpenIDConnectProviders", "iam:CreateRole", "iam:GetRole", "iam:ListRoles", ], } def _all_actions(policy): actions = set() for stmt in policy["Statement"]: if stmt.get("Effect") != "Allow": continue stmt_actions = stmt.get("Action", []) if isinstance(stmt_actions, str): stmt_actions = [stmt_actions] for a in stmt_actions: actions.add(a) return actions def _has_action(all_actions, required): if required.endswith("*"): prefix = required[:-1] return any(a.startswith(prefix) for a in all_actions) return required in all_actions class TestIAMPolicyBaseline: """REQ-116: the spike_runner_policy.json grants the v1.11 minimum.""" @pytest.fixture(scope="class") def policy(self): return json.loads(POLICY_PATH.read_text()) def test_policy_file_exists_and_is_valid_json(self, policy): assert "Statement" in policy assert isinstance(policy["Statement"], list) assert len(policy["Statement"]) >= 15 def test_all_statements_are_allow_or_have_effect(self, policy): for stmt in policy["Statement"]: assert "Effect" in stmt assert stmt["Effect"] in {"Allow", "Deny"} @pytest.mark.parametrize("group", sorted(REQUIRED_ACTIONS)) def test_required_actions_present(self, policy, group): all_actions = _all_actions(policy) missing = [a for a in REQUIRED_ACTIONS[group] if not _has_action(all_actions, a)] assert not missing, f"missing required {group} actions: {missing}" def test_dynamodb_contracts_table_in_resource(self, policy): contracts_stmts = [ s for s in policy["Statement"] if any("nova-contracts" in r for r in ( s.get("Resource") if isinstance(s.get("Resource"), list) else [s.get("Resource", "")] )) ] assert contracts_stmts, "no statement references the nova-contracts table" def test_lambda_scoped_to_nova_functions(self, policy): lambda_stmts = [s for s in policy["Statement"] if any( a.startswith("lambda:") for a in ( s.get("Action") if isinstance(s.get("Action"), list) else [s.get("Action", "")] ) )] assert lambda_stmts, "no lambda statement" for s in lambda_stmts: res = s.get("Resource", "") if isinstance(res, list): res = " ".join(res) assert "function:nova-*" in res or res == "*", \ "lambda actions not scoped to nova-* functions" def test_cost_explorer_is_read_only(self, policy): ce_actions = set() for s in policy["Statement"]: acts = s.get("Action", []) if isinstance(acts, str): acts = [acts] for a in acts: if a.startswith("ce:"): ce_actions.add(a) for a in ce_actions: assert a.startswith("ce:Get") or a.startswith("ce:List") or a.startswith("ce:Describe"), \ f"non-read Cost Explorer action granted: {a}" def test_no_statement_uses_iam_passrole_to_star(self, policy): for s in policy["Statement"]: acts = s.get("Action", []) if isinstance(acts, str): acts = [acts] if "iam:PassRole" in acts: res = s.get("Resource", "") if isinstance(res, list): res = " ".join(res) assert res != "*", "iam:PassRole must not be granted to Resource: *" def test_iam_role_creation_scoped_to_nova_prefix(self, policy): """G-104: iam:CreateRole must be scoped to role/nova-* (not Resource: *).""" for s in policy["Statement"]: acts = s.get("Action", []) if isinstance(acts, str): acts = [acts] if "iam:CreateRole" in acts: res = s.get("Resource", "") if isinstance(res, list): res = " ".join(res) assert "nova-*" in res, f"iam:CreateRole must be scoped to nova-* (got: {res})" def test_kms_scoped_to_nova_alias(self, policy): """G-104: kms:CreateKey etc. must be scoped to alias/nova-* (not Resource: *).""" for s in policy["Statement"]: acts = s.get("Action", []) if isinstance(acts, str): acts = [acts] if any(a.startswith("kms:") for a in acts): res = s.get("Resource", "") if isinstance(res, list): res = " ".join(res) assert "nova-*" in res, f"kms actions must be scoped to nova-* (got: {res})" def test_cloudfront_waf_remain_global(self, policy): """G-104: CloudFront + WAFv2 (CloudFront scope) ARNs are global; Resource: * is acceptable here (documented constraint, not a defect).""" global_actions = {"cloudfront:", "wafv2:"} for s in policy["Statement"]: acts = s.get("Action", []) if isinstance(acts, str): acts = [acts] if any(any(a.startswith(g) for g in global_actions) for a in acts): res = s.get("Resource", "") if isinstance(res, list): res = res[0] if res else "" # CloudFront/WAFv2 are allowed to be * (global ARNs) assert res == "*" or "nova" in res