From b944c61e7ea4cb6c8ba220326c510b4714422b13 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 27 Sep 2026 17:55:44 +1000 Subject: [PATCH] Validate policy yaml with unittest and pydantic --- .pre-commit-config.yaml | 11 +-- Makefile | 6 +- scripts/check_policy_paths.py | 61 ------------- scripts/test_check_policy_paths.py | 56 ------------ tests/__init__.py | 0 tests/test_policies.py | 133 +++++++++++++++++++++++++++++ 6 files changed, 142 insertions(+), 125 deletions(-) delete mode 100644 scripts/check_policy_paths.py delete mode 100644 scripts/test_check_policy_paths.py create mode 100644 tests/__init__.py create mode 100644 tests/test_policies.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index bc29faa..03a38a4 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -24,9 +24,10 @@ repos: ] - repo: local hooks: - - id: policy-path-scope - name: policy rule paths stay within their own directory - entry: python3 scripts/check_policy_paths.py + - id: vault-yaml-tests + name: vault yaml definitions pass their unit tests + entry: python3 -m unittest discover -s tests -t . language: python - additional_dependencies: [pyyaml] - files: ^policies/.*\.yaml$ + additional_dependencies: [pyyaml, pydantic] + pass_filenames: false + always_run: true diff --git a/Makefile b/Makefile index ffe9cd9..da57e04 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: init plan apply format check-policies +.PHONY: init plan apply format test VAULT_AUTH_METHOD ?= approle VAULT_K8S_ROLE ?= woodpecker_terraform_vault @@ -28,8 +28,8 @@ apply: init @$(call vault_env) && \ terragrunt run --all --parallelism 2 --non-interactive apply -check-policies: - @python3 scripts/check_policy_paths.py +test: + @uv run --with pyyaml --with pydantic python -m unittest discover -s tests -t . format: @echo "Formatting OpenTofu files..." diff --git a/scripts/check_policy_paths.py b/scripts/check_policy_paths.py deleted file mode 100644 index 415883f..0000000 --- a/scripts/check_policy_paths.py +++ /dev/null @@ -1,61 +0,0 @@ -#!/usr/bin/env python3 -"""Fail when a rule path in policies/**/*.yaml grants outside the policy's own directory.""" -import glob -import sys -from pathlib import PurePosixPath - -import yaml - -# kv-v2 inserts one of these directly after the mount; it is not part of the scope. -KV_API_SEGMENTS = {"data", "metadata", "delete", "undelete", "destroy"} - -ALLOWED = { - ("policies/global-root.yaml", "*"), # root policy - ("policies/gpg/admin.yaml", "sys/plugins/catalog/secret/vault-plugin-secrets-gpg"), # plugin catalog registration - ("policies/kv/service/terraform/authentik.yaml", "kv/data/kubernetes/namespace/+/default/oauth-credentials"), # terraform writes k8s namespace secrets - ("policies/kv/service/terraform/authentik.yaml", "kv/data/kubernetes/namespace/logging/default/vlogs-oauth-credentials"), # terraform writes k8s namespace secrets - ("policies/kv/service/terraform/enc-encapi-environment.yaml", "kv/data/kubernetes/namespace/encapi/default/environment"), # terraform writes k8s namespace secrets - ("policies/kv/service/terraform/rancher.yaml", "kv/data/kubernetes/namespace/cattle-system/default/oauth-credentials"), # terraform writes k8s namespace secrets - ("policies/identity/group/admin.yaml", "identity/group-alias"), # group-alias endpoint for the same groups - ("policies/identity/group/admin.yaml", "identity/group-alias/*"), # group-alias endpoint for the same groups - ("policies/identity/group/admin.yaml", "identity/lookup/group"), # group lookup endpoint - ("policies/sys/policy/admin.yaml", "sys/policies/acl"), # dir is policy, API path is policies - ("policies/sys/policy/admin.yaml", "sys/policies/acl/*"), # dir is policy, API path is policies - ("policies/sys/mounts/admin.yaml", "sys/mounts-tune/*"), # sibling API path of the mounts endpoint -} - - -def policy_scope(path): - parts = PurePosixPath(path).parts - rel = PurePosixPath(*parts[parts.index("policies") + 1:]) - # a policy at the policies/ root has no directory, so its own name is the scope - return "policies" / rel, rel.parent if rel.parent.parts else PurePosixPath(rel.stem) - - -def rule_scope(rule_path): - path = PurePosixPath(rule_path) - if len(path.parts) > 1 and path.parts[1] in KV_API_SEGMENTS: - return PurePosixPath(path.parts[0], *path.parts[2:]) - return path - - -def violations(files): - for f in files: - key, scope = policy_scope(f) - rules = (yaml.safe_load(open(f)) or {}).get("rules") - if not rules: - yield f"{f}: no rules" - continue - for rule in rules: - path = rule.get("path") - if not path: - yield f"{f}: rule without a path" - elif (str(key), path) not in ALLOWED and not rule_scope(path).is_relative_to(scope): - yield f'{f}: rule path "{path}" escapes policy scope "{scope}"' - - -if __name__ == "__main__": - found = list(violations(sys.argv[1:] or sorted(glob.glob("policies/**/*.yaml", recursive=True)))) - for v in found: - print(v) - sys.exit(1 if found else 0) diff --git a/scripts/test_check_policy_paths.py b/scripts/test_check_policy_paths.py deleted file mode 100644 index 2fd6f26..0000000 --- a/scripts/test_check_policy_paths.py +++ /dev/null @@ -1,56 +0,0 @@ -#!/usr/bin/env python3 -"""Run with: python3 scripts/test_check_policy_paths.py""" -import sys -import tempfile -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).parent)) -from check_policy_paths import violations - -TMP = Path(tempfile.mkdtemp()) - - -def check(rel, *rule_paths): - f = TMP / rel - f.parent.mkdir(parents=True, exist_ok=True) - f.write_text("rules:\n" + "".join(f' - path: "{p}"\n' for p in rule_paths)) - return [v.replace(f"{TMP}/", "") for v in violations([str(f)])] - - -# in scope, including kv-v2 data/metadata segments and any kv-v2 mount -assert check("policies/sys/mounts/admin.yaml", "sys/mounts", "sys/mounts/*") == [] -assert check("policies/rundeck/rundeck.yaml", "rundeck/data/*", "rundeck/metadata/*") == [] -assert check("policies/kv/service/authentik/oidc-vault/read.yaml", "kv/data/service/authentik/oidc-vault") == [] - -# + and * below the directory are fine -assert check("policies/kv/service/vault/read.yaml", "kv/data/service/vault/+/+/auth_backend/*") == [] -assert check("policies/kubernetes/au/admin.yaml", "kubernetes/au/+/config") == [] - -# + standing in for a literal directory segment is not -assert check("policies/kv/service/vault/au/syd1/ghp/write.yaml", "kv/data/service/vault/+/+/ghp/config") == [ - 'policies/kv/service/vault/au/syd1/ghp/write.yaml: rule path ' - '"kv/data/service/vault/+/+/ghp/config" escapes policy scope "kv/service/vault/au/syd1/ghp"' -] - -# a sibling API path is not a prefix match -assert check("policies/sys/thing/admin.yaml", "sys/thing-tune/*") == [ - 'policies/sys/thing/admin.yaml: rule path "sys/thing-tune/*" escapes policy scope "sys/thing"' -] - -# out of tree entirely -assert check("policies/kv/foo/bar/baz.yaml", "kv/data/foo/baz/bar") == [ - 'policies/kv/foo/bar/baz.yaml: rule path "kv/data/foo/baz/bar" escapes policy scope "kv/foo/bar"' -] - -# allowlisted outliers pass -assert check("policies/global-root.yaml", "*") == [] -assert check("policies/sys/policy/admin.yaml", "sys/policies/acl", "sys/policies/acl/*") == [] -assert check("policies/sys/mounts/admin.yaml", "sys/mounts-tune/*") == [] - -# malformed policies fail -(TMP / "policies/empty.yaml").write_text("auth:\n approle:\n - x\n") -assert list(violations([str(TMP / "policies/empty.yaml")]))[0].endswith(": no rules") -(TMP / "policies/nopath.yaml").write_text("rules:\n - capabilities:\n - read\n") -assert list(violations([str(TMP / "policies/nopath.yaml")]))[0].endswith(": rule without a path") - -print("ok") diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_policies.py b/tests/test_policies.py new file mode 100644 index 0000000..31cf9e8 --- /dev/null +++ b/tests/test_policies.py @@ -0,0 +1,133 @@ +"""Validate the Vault policy definitions under policies/.""" +import unittest +from pathlib import Path, PurePosixPath +from typing import Literal + +import yaml +from pydantic import BaseModel, ConfigDict, Field, ValidationError + +REPO_ROOT = Path(__file__).resolve().parent.parent + +# kv-v2 inserts one of these directly after the mount; it is not part of the scope. +KV_API_SEGMENTS = {"data", "metadata", "delete", "undelete", "destroy"} + +ALLOWED = { + ("policies/global-root.yaml", "*"), # root policy + ("policies/gpg/admin.yaml", "sys/plugins/catalog/secret/vault-plugin-secrets-gpg"), # plugin catalog registration + ("policies/kv/service/terraform/authentik.yaml", "kv/data/kubernetes/namespace/+/default/oauth-credentials"), # terraform writes k8s namespace secrets + ("policies/kv/service/terraform/authentik.yaml", "kv/data/kubernetes/namespace/logging/default/vlogs-oauth-credentials"), # terraform writes k8s namespace secrets + ("policies/kv/service/terraform/enc-encapi-environment.yaml", "kv/data/kubernetes/namespace/encapi/default/environment"), # terraform writes k8s namespace secrets + ("policies/kv/service/terraform/rancher.yaml", "kv/data/kubernetes/namespace/cattle-system/default/oauth-credentials"), # terraform writes k8s namespace secrets + ("policies/identity/group/admin.yaml", "identity/group-alias"), # group-alias endpoint for the same groups + ("policies/identity/group/admin.yaml", "identity/group-alias/*"), # group-alias endpoint for the same groups + ("policies/identity/group/admin.yaml", "identity/lookup/group"), # group lookup endpoint + ("policies/sys/policy/admin.yaml", "sys/policies/acl"), # dir is policy, API path is policies + ("policies/sys/policy/admin.yaml", "sys/policies/acl/*"), # dir is policy, API path is policies + ("policies/sys/mounts/admin.yaml", "sys/mounts-tune/*"), # sibling API path of the mounts endpoint +} + + +class Rule(BaseModel): + model_config = ConfigDict(extra="forbid") + + path: str = Field(min_length=1) + capabilities: list[ + Literal["create", "read", "update", "patch", "delete", "list", "sudo", "deny"] + ] = Field(min_length=1) + + +class Policy(BaseModel): + model_config = ConfigDict(extra="forbid") + + rules: list[Rule] = Field(min_length=1) + auth: dict[str, list[str]] + + +def policy_files(): + return sorted(REPO_ROOT.glob("policies/**/*.yaml")) + + +def escaping_scope(policy_file, rule_path): + """The policy directory a rule path reaches outside of, or None when in scope.""" + if (policy_file, rule_path) in ALLOWED: + return None + parts = PurePosixPath(policy_file).parts + below_policies = PurePosixPath(*parts[parts.index("policies") + 1:]) + # a policy at the policies/ root has no directory, so its own name is the scope + scope = below_policies.parent if below_policies.parent.parts else PurePosixPath(below_policies.stem) + path = PurePosixPath(rule_path) + if len(path.parts) > 1 and path.parts[1] in KV_API_SEGMENTS: + path = PurePosixPath(path.parts[0], *path.parts[2:]) + return None if path.is_relative_to(scope) else str(scope) + + +class PolicyFileTests(unittest.TestCase): + def setUp(self): + self.files = policy_files() + self.assertTrue(self.files, f"no policy files discovered under {REPO_ROOT}/policies") + + def test_schema(self): + for f in self.files: + with self.subTest(path=f.relative_to(REPO_ROOT).as_posix()): + Policy.model_validate(yaml.safe_load(f.read_text())) + + def test_paths(self): + for f in self.files: + rel = f.relative_to(REPO_ROOT).as_posix() + for rule in yaml.safe_load(f.read_text()).get("rules") or []: + rule_path = rule.get("path") + with self.subTest(path=rel, rule=rule_path): + self.assertIsNotNone(rule_path, "rule has no path") + scope = escaping_scope(rel, rule_path) + self.assertIsNone(scope, f'rule path "{rule_path}" escapes policy scope "{scope}"') + + +class RuleScopeTests(unittest.TestCase): + """The scope rule against fixtures, so a clean tree cannot hide a broken check.""" + + def test_in_scope(self): + for policy_file, rule_path in [ + ("policies/sys/mounts/admin.yaml", "sys/mounts"), + ("policies/sys/mounts/admin.yaml", "sys/mounts/*"), + ("policies/rundeck/rundeck.yaml", "rundeck/data/*"), + ("policies/rundeck/rundeck.yaml", "rundeck/metadata/*"), + ("policies/kv/service/authentik/oidc-vault/read.yaml", "kv/data/service/authentik/oidc-vault"), + ("policies/kv/service/vault/read.yaml", "kv/data/service/vault/+/+/auth_backend/*"), + ("policies/kubernetes/au/admin.yaml", "kubernetes/au/+/config"), + ]: + with self.subTest(policy=policy_file, rule=rule_path): + self.assertIsNone(escaping_scope(policy_file, rule_path)) + + def test_escapes(self): + for policy_file, rule_path, scope in [ + ("policies/sys/thing/admin.yaml", "sys/thing-tune/*", "sys/thing"), + ("policies/kv/foo/bar/baz.yaml", "kv/data/foo/baz/bar", "kv/foo/bar"), + ("policies/kv/service/vault/au/syd1/ghp/w.yaml", "kv/data/service/vault/+/+/ghp/config", "kv/service/vault/au/syd1/ghp"), + ]: + with self.subTest(policy=policy_file, rule=rule_path): + self.assertEqual(escaping_scope(policy_file, rule_path), scope) + + def test_allowlisted(self): + for policy_file, rule_path in [ + ("policies/global-root.yaml", "*"), + ("policies/sys/policy/admin.yaml", "sys/policies/acl"), + ("policies/sys/policy/admin.yaml", "sys/policies/acl/*"), + ("policies/sys/mounts/admin.yaml", "sys/mounts-tune/*"), + ]: + with self.subTest(policy=policy_file, rule=rule_path): + self.assertIsNone(escaping_scope(policy_file, rule_path)) + + def test_schema_rejects_malformed(self): + auth = {"approle": ["tf_vault"]} + rule = {"path": "kv/data/x", "capabilities": ["read"]} + for doc in [ + {"auth": auth}, # no rules + {"rules": [], "auth": auth}, # empty rules + {"rules": [{"capabilities": ["read"]}], "auth": auth}, # rule without a path + {"rules": [{"path": "kv/data/x", "capabilities": []}], "auth": auth}, # rule without capabilities + {"rules": [{"path": "kv/data/x", "capabilities": ["write"]}], "auth": auth}, # not a Vault capability + {"rules": [rule]}, # no auth + {"rules": [rule], "auth": auth, "rulez": []}, # typo'd top-level key + ]: + with self.subTest(doc=doc), self.assertRaises(ValidationError): + Policy.model_validate(doc)