107 lines
4.6 KiB
Python
107 lines
4.6 KiB
Python
"""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"}
|
|
|
|
|
|
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."""
|
|
parts = PurePosixPath(policy_file).parts
|
|
# a policy at the policies/ root is located at the root, so its scope is the whole tree
|
|
scope = PurePosixPath(*parts[parts.index("policies") + 1:]).parent
|
|
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"),
|
|
("policies/global-root.yaml", "*"), # a root-level policy is scoped to the whole tree
|
|
]:
|
|
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_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)
|