unkin-agent
9c1f05ed60
ci/woodpecker/push/apply Pipeline was successful
Identity OIDC (token roles, provider clients, scopes, assignments) is not managed from YAML yet, and role/client names share one global credential path namespace that is easy to collide on.
- add identity_oidc{,_key,_role,_client,_scope,_assignment,_provider} modules and config discovery; roles/clients live under config/<kind>/<key>/
- allow each role/client client_id on its key via vault_identity_oidc_key_allowed_client_id
- resolve provider allowed_clients from client names
- add a test rejecting duplicate role/client names and references to missing keys, clients, scopes or assignments
Reviewed-on: #166
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
121 lines
4.5 KiB
Python
121 lines
4.5 KiB
Python
"""Validate identity OIDC role/client definitions under config/."""
|
|
import tempfile
|
|
import unittest
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def problems(config_dir):
|
|
"""Duplicate role/client names and references to missing keys, clients, scopes or assignments."""
|
|
config_dir = Path(config_dir)
|
|
keys = {f.stem for f in config_dir.glob("identity_oidc_key/*.yaml")}
|
|
names = defaultdict(list)
|
|
found = []
|
|
for kind in ("identity_oidc_role", "identity_oidc_client"):
|
|
for f in sorted(config_dir.glob(f"{kind}/**/*.yaml")):
|
|
rel = f.relative_to(config_dir).as_posix()
|
|
names[f.stem].append(rel)
|
|
key = f.parent.relative_to(config_dir / kind).as_posix()
|
|
if key not in keys:
|
|
found.append(f'{rel}: key "{key}" has no config/identity_oidc_key/{key}.yaml')
|
|
for name, files in sorted(names.items()):
|
|
if len(files) > 1:
|
|
found.append(f'role/client name "{name}" is used by {", ".join(files)}')
|
|
|
|
def stems(kind):
|
|
return {f.stem for f in config_dir.glob(f"{kind}/*.yaml")}
|
|
|
|
clients = {f.stem for f in config_dir.glob("identity_oidc_client/**/*.yaml")}
|
|
refs = [
|
|
("identity_oidc_provider", "allowed_clients", clients | {"*"}),
|
|
("identity_oidc_provider", "scopes_supported", stems("identity_oidc_scope")),
|
|
# allow_all is Vault's built-in assignment.
|
|
("identity_oidc_client", "assignments", stems("identity_oidc_assignment") | {"allow_all"}),
|
|
]
|
|
for kind, field, known in refs:
|
|
for f in sorted(config_dir.glob(f"{kind}/**/*.yaml")):
|
|
for ref in (yaml.safe_load(f.read_text()) or {}).get(field, []):
|
|
if ref not in known:
|
|
found.append(f'{f.relative_to(config_dir).as_posix()}: {field} "{ref}" does not exist')
|
|
return found
|
|
|
|
|
|
class RepoTests(unittest.TestCase):
|
|
def test_repo_config(self):
|
|
self.assertEqual(problems(REPO_ROOT / "config"), [])
|
|
|
|
|
|
class ProblemsTests(unittest.TestCase):
|
|
def tree(self, *paths, **content):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
for p in (*paths, *content):
|
|
f = Path(tmp.name, p)
|
|
f.parent.mkdir(parents=True, exist_ok=True)
|
|
f.write_text(yaml.safe_dump(content.get(p, {})))
|
|
return tmp.name
|
|
|
|
def test_clean(self):
|
|
d = self.tree(
|
|
"identity_oidc_key/a.yaml",
|
|
"identity_oidc_key/b.yaml",
|
|
"identity_oidc_role/a/r1.yaml",
|
|
"identity_oidc_role/b/r2.yaml",
|
|
"identity_oidc_client/a/c1.yaml",
|
|
)
|
|
self.assertEqual(problems(d), [])
|
|
|
|
def test_role_client_collision(self):
|
|
d = self.tree(
|
|
"identity_oidc_key/a.yaml",
|
|
"identity_oidc_role/a/x.yaml",
|
|
"identity_oidc_client/a/x.yaml",
|
|
)
|
|
self.assertEqual(problems(d), [
|
|
'role/client name "x" is used by identity_oidc_role/a/x.yaml, identity_oidc_client/a/x.yaml'
|
|
])
|
|
|
|
def test_collision_across_keys(self):
|
|
d = self.tree(
|
|
"identity_oidc_key/a.yaml",
|
|
"identity_oidc_key/b.yaml",
|
|
"identity_oidc_role/a/x.yaml",
|
|
"identity_oidc_role/b/x.yaml",
|
|
)
|
|
self.assertEqual(problems(d), [
|
|
'role/client name "x" is used by identity_oidc_role/a/x.yaml, identity_oidc_role/b/x.yaml'
|
|
])
|
|
|
|
def test_missing_key(self):
|
|
d = self.tree("identity_oidc_client/nope/c.yaml")
|
|
self.assertEqual(problems(d), [
|
|
'identity_oidc_client/nope/c.yaml: key "nope" has no config/identity_oidc_key/nope.yaml'
|
|
])
|
|
|
|
def test_references(self):
|
|
d = self.tree(
|
|
"identity_oidc_key/a.yaml",
|
|
"identity_oidc_scope/groups.yaml",
|
|
"identity_oidc_assignment/admins.yaml",
|
|
**{
|
|
"identity_oidc_client/a/c.yaml": {"assignments": ["admins", "allow_all", "ghost"]},
|
|
"identity_oidc_provider/p.yaml": {
|
|
"allowed_clients": ["c", "*", "nope"],
|
|
"scopes_supported": ["groups", "email"],
|
|
},
|
|
},
|
|
)
|
|
self.assertEqual(problems(d), [
|
|
'identity_oidc_provider/p.yaml: allowed_clients "nope" does not exist',
|
|
'identity_oidc_provider/p.yaml: scopes_supported "email" does not exist',
|
|
'identity_oidc_client/a/c.yaml: assignments "ghost" does not exist',
|
|
])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|