Validate policy yaml with unittest and pydantic
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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..."
|
||||
|
||||
@@ -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)
|
||||
@@ -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")
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user