acdl/tests/test_iam_policy_baseline.py

"""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