mirror of
https://github.com/Sea-Haven-Industries/seahaven-org-baseline.git
synced 2026-10-02 15:33:23 +00:00
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 <amoussa1229@users.noreply.github.com>
This commit is contained in:
Cursor Agent
parent
5907cd3467
commit
740ba53d3d
2 changed files with 401 additions and 94 deletions
9
.github/workflows/ci.yaml
vendored
9
.github/workflows/ci.yaml
vendored
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue