coulomb/net-kingdom
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

NK-WP-0048: canon accepted; SecurityCaseType schemas, validator and template (T01, T03)
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 4s
Details

Browse source 
security-case-types_v0.1 accepted by the founder 2026-10-08. Adds JSON schemas
for types and cases, tools/security-case-types/security_case_validator.py
(validate-type, validate-case, render via the scenario composer, plan-only),
and a runnable _template type with README sections, an example case and tests.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 5457@bnt-lap001
Assistant-Session: 792bf84a-3783-4f5d-a811-cded61dd2012
This commit is contained in:
tegwick 2026-10-08 22:29:13 +02:00
parent dcacdf57f3
commit d8cf42bab6
11 changed files with 1122 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

34
tools/security-case-types/README.md Normal file
View file

@ -0,0 +1,34 @@
# security-case-types tools
Validator and renderer for SecurityCaseTypes and SecurityCases
(`canon/standards/security-case-types_v0.1.md`, schemas
`canon/schemas/security-case-type_v0.1.schema.json` and
`security-case_v0.1.schema.json`).
```bash
# a type directory: type.yaml, README.md sections, tests, example cases
python3 tools/security-case-types/security_case_validator.py validate-type security-case-types/<type-id>
# a case against its type (security-case-types/<type-id>/type.yaml)
python3 tools/security-case-types/security_case_validator.py validate-case <case.yaml> --types security-case-types
# render a case into the composer's plan plus materialization, follow-through,
# verification and rollback; execution is never permitted
python3 tools/security-case-types/security_case_validator.py render <case.yaml> \
--types security-case-types capabilities/playbooks/*.yaml
```
Only `implemented` types accept cases. `--allow-unimplemented` exists for a type's
own example cases while the type is being written and acknowledged.
The validator refuses, among other things: missing sections, unknown or unpinned
capabilities, attended steps without justification, steps or materialized items
without a rollback, verification without a negative check or that reads a value,
templates with unknown parameters, credential-shaped values anywhere, secret
parameters that are not custody references, and high-risk types without founder
acceptance.
To start a type: copy `security-case-types/_template/`, rename the directory to
the type id, and rewrite every section. The template's tests travel with it.
Tests: `python3 -m pytest tools/security-case-types/tests security-case-types/*/tests`.

500
tools/security-case-types/security_case_validator.py Normal file
View file

@ -0,0 +1,500 @@
#!/usr/bin/env python3
"""Validate SecurityCaseTypes and SecurityCases, and render a case into a plan.
Canon: canon/standards/security-case-types_v0.1.md. A type is a NetKingdom-owned
composition of declared playbook capabilities; a case fills in the type's
parameters. Rendering produces the security-scenario composer's deterministic,
plan-only output (execution is never permitted here) plus the type's
materialization and follow-through, with parameters substituted.
security_case_validator.py validate-type security-case-types/<type-id>
security_case_validator.py validate-case <case.yaml> --types security-case-types
security_case_validator.py render <case.yaml> --types security-case-types <declarations...>
"""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import string
import sys
from pathlib import Path
from typing import Any
import yaml
TOOL_DIR = Path(__file__).resolve().parent
COMPOSER_DIR = TOOL_DIR.parent / "security-scenario-composer"
sys.path.insert(0, str(COMPOSER_DIR))
import security_scenario_composer as composer # noqa: E402
contract = composer.contract
Issue = contract.Issue
TYPE_API_VERSION = "netkingdom.io/security-case-type/v0.1"
CASE_API_VERSION = "netkingdom.io/security-case/v0.1"
RENDER_API_VERSION = "netkingdom.io/security-case-render/v0.1"
ID = re.compile(r"^[a-z0-9][a-z0-9-]*$")
VERSION = re.compile(r"^\d+\.\d+$")
TYPE_REF = re.compile(r"^([a-z0-9][a-z0-9-]*)@(\d+\.\d+)$")
TYPE_STATUSES = ["draft", "proposed", "acknowledged", "implemented", "deprecated", "retired"]
CASE_STATUSES = ["proposed", "accepted", "materialized", "live", "retired"]
RISK_CLASSES = {"low", "medium", "high"}
ACCEPTERS = {"owner", "founder"}
MATERIALIZATION_KINDS = {"openbao_policy", "openbao_role", "kv_path", "catalog_entry"}
TYPE_SECTIONS = [
"id", "version", "status", "title", "purpose", "parameters", "composition",
"materialization", "follow_through", "verification", "rollback", "acceptance",
]
REQUIRED_README_HEADINGS = ["## Threat model", "## Alternatives considered", "## Known limits"]
# Values that look like credentials are refused anywhere in a case. A
# secret_reference parameter must hold a custody reference, never a value.
SECRET_SHAPES = [
re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
re.compile(r"\b(?:hvs|hvb|hvr)\.[A-Za-z0-9_-]{8,}"),
re.compile(r"\b(?:sk|pk|rk)-[A-Za-z0-9_-]{16,}"),
re.compile(r"\bAKIA[0-9A-Z]{16}\b"),
re.compile(r"\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\."),
re.compile(r"\bgh[pousr]_[A-Za-z0-9]{20,}"),
]
CUSTODY_REFERENCE = re.compile(r"^(?:bao|openbao|vault|env|file):[^\s]+$")
def err(path: str, message: str) -> Issue:
return Issue("ERROR", path, message)
def has_errors(issues: list[Issue]) -> bool:
return any(i.level == "ERROR" for i in issues)
def placeholders(text: str) -> set[str]:
return {f[1] for f in string.Formatter().parse(text) if f[1]}
def _walk_strings(value: Any, path: str):
if isinstance(value, str):
yield path, value
elif isinstance(value, dict):
for k, v in value.items():
yield from _walk_strings(v, f"{path}.{k}")
elif isinstance(value, list):
for i, v in enumerate(value):
yield from _walk_strings(v, f"{path}[{i}]")
def secret_shaped(text: str) -> bool:
return any(p.search(text) for p in SECRET_SHAPES)
# ── Types ────────────────────────────────────────────────────────────────────
def validate_type(data: dict[str, Any], type_dir: Path | None = None) -> list[Issue]:
issues: list[Issue] = []
if data.get("apiVersion") != TYPE_API_VERSION:
issues.append(err("type.apiVersion", f"must be {TYPE_API_VERSION}"))
if data.get("kind") != "SecurityCaseType":
issues.append(err("type.kind", "must be SecurityCaseType"))
spec = data.get("spec")
if not isinstance(spec, dict):
return issues + [err("type.spec", "missing")]
for section in TYPE_SECTIONS:
if section not in spec:
issues.append(err(f"type.spec.{section}", "required section is missing"))
if has_errors(issues):
return issues
if not ID.fullmatch(str(spec["id"])):
issues.append(err("type.spec.id", "must be lowercase letters, digits and hyphens"))
if not VERSION.fullmatch(str(spec["version"])):
issues.append(err("type.spec.version", "must be <major>.<minor>"))
if spec["status"] not in TYPE_STATUSES:
issues.append(err("type.spec.status", f"must be one of {TYPE_STATUSES}"))
params = _validate_type_parameters(spec["parameters"], issues)
step_ids = _validate_follow_through(spec["follow_through"], issues)
_validate_composition(spec["composition"], params, issues)
_validate_materialization(spec["materialization"], params, issues)
_validate_verification(spec["verification"], params, issues)
_validate_rollback(spec["rollback"], step_ids, spec["materialization"], issues)
_validate_acceptance(spec["acceptance"], issues)
for path, text in _walk_strings(spec, "type.spec"):
if secret_shaped(text):
issues.append(err(path, "contains a credential-shaped value"))
if type_dir is not None:
readme = type_dir / "README.md"
if not readme.is_file():
issues.append(err("type.README.md", "missing"))
else:
text = readme.read_text()
for heading in REQUIRED_README_HEADINGS:
if heading not in text:
issues.append(err("type.README.md", f"missing section {heading!r}"))
tests = type_dir / "tests"
if not tests.is_dir() or not any(tests.glob("test_*.py")):
issues.append(err("type.tests", "a type needs runnable tests (tests/test_*.py)"))
examples = sorted((type_dir / "examples").glob("*.yaml")) if (type_dir / "examples").is_dir() else []
if not examples:
issues.append(err("type.examples", "a type needs at least one example case"))
return issues
def _validate_type_parameters(params: Any, issues: list[Issue]) -> dict[str, dict]:
out: dict[str, dict] = {}
if not isinstance(params, list) or not params:
issues.append(err("type.spec.parameters", "must be a non-empty list"))
return out
for i, p in enumerate(params):
path = f"type.spec.parameters[{i}]"
if not isinstance(p, dict) or not ID.fullmatch(str(p.get("name", "")).replace("_", "-")):
issues.append(err(path, "needs a name of lowercase letters, digits, '-' or '_'"))
continue
name = p["name"]
if name in out:
issues.append(err(path, f"duplicate parameter {name!r}"))
if p.get("type") not in contract.PARAM_TYPES:
issues.append(err(f"{path}.type", f"must be one of {sorted(contract.PARAM_TYPES)}"))
if p.get("sensitivity") not in contract.SENSITIVITIES:
issues.append(err(f"{path}.sensitivity", f"must be one of {sorted(contract.SENSITIVITIES)}"))
if p.get("sensitivity") == "secret_reference" and p.get("type") != "string":
issues.append(err(path, "a secret_reference parameter must be a string custody reference"))
if not isinstance(p.get("case_settable"), bool):
issues.append(err(f"{path}.case_settable", "must be true or false"))
if not p.get("description"):
issues.append(err(f"{path}.description", "required"))
if not p.get("case_settable") and "default" not in p:
issues.append(err(path, "a parameter cases cannot set needs a default"))
out[name] = p
return out
def _validate_composition(comp: Any, params: dict[str, dict], issues: list[Issue]) -> None:
path = "type.spec.composition"
if not isinstance(comp, dict):
issues.append(err(path, "must be a mapping"))
return
if comp.get("authority") not in contract.SCENARIO_AUTHORITIES:
issues.append(err(f"{path}.authority", f"must be one of {sorted(contract.SCENARIO_AUTHORITIES)}"))
trust = comp.get("initial_trust")
if not isinstance(trust, list) or not trust or not set(trust) <= contract.TRUST_STATES:
issues.append(err(f"{path}.initial_trust", "must be a non-empty list of known trust states"))
caps = comp.get("capabilities")
if not isinstance(caps, list) or not caps:
issues.append(err(f"{path}.capabilities", "must list at least one capability"))
caps = []
for c in caps:
if c not in contract.CAPABILITIES:
issues.append(err(f"{path}.capabilities", f"unknown capability {c!r}"))
providers = comp.get("providers") or {}
for c in caps:
if not providers.get(c):
issues.append(err(f"{path}.providers", f"capability {c!r} needs an exact provider pin"))
for target, source in (comp.get("parameter_bindings") or {}).items():
if not re.fullmatch(r"[a-z0-9][a-z0-9._-]*\.[a-z_][a-z0-9_]*", str(target)):
issues.append(err(f"{path}.parameter_bindings", f"{target!r} must be <declaration-id>.<parameter>"))
if source not in params:
issues.append(err(f"{path}.parameter_bindings.{target}", f"binds unknown case parameter {source!r}"))
def _validate_templates(items: Any, path: str, params: dict[str, dict], issues: list[Issue]) -> None:
for i, item in enumerate(items):
for spath, text in _walk_strings(item, f"{path}[{i}]"):
for name in placeholders(text):
if name not in params:
issues.append(err(spath, f"template uses unknown parameter {{{name}}}"))
def _validate_materialization(items: Any, params: dict[str, dict], issues: list[Issue]) -> None:
path = "type.spec.materialization"
if not isinstance(items, list) or not items:
issues.append(err(path, "must list what ops-mason builds per case"))
return
seen = set()
for i, item in enumerate(items):
if not isinstance(item, dict) or item.get("kind") not in MATERIALIZATION_KINDS:
issues.append(err(f"{path}[{i}].kind", f"must be one of {sorted(MATERIALIZATION_KINDS)}"))
continue
if not item.get("id") or item["id"] in seen:
issues.append(err(f"{path}[{i}].id", "needs a unique id"))
seen.add(item.get("id"))
if not item.get("template"):
issues.append(err(f"{path}[{i}].template", "required"))
_validate_templates(items, path, params, issues)
def _validate_follow_through(steps: Any, issues: list[Issue]) -> set[str]:
path = "type.spec.follow_through"
ids: set[str] = set()
if not isinstance(steps, list) or not steps:
issues.append(err(path, "must list the ops-warden steps"))
return ids
for i, s in enumerate(steps):
p = f"{path}[{i}]"
if not isinstance(s, dict) or not s.get("id"):
issues.append(err(p, "needs an id"))
continue
if s["id"] in ids:
issues.append(err(p, f"duplicate step {s['id']!r}"))
ids.add(s["id"])
for key in ("owner", "entry_point", "auth"):
if not s.get(key):
issues.append(err(f"{p}.{key}", "required"))
if not isinstance(s.get("warden_executes"), bool):
issues.append(err(f"{p}.warden_executes", "must be true or false"))
if s.get("attended") is True and not s.get("attended_justification"):
issues.append(err(f"{p}.attended_justification", "an attended step must say why it cannot be unattended"))
return ids
def _validate_verification(checks: Any, params: dict[str, dict], issues: list[Issue]) -> None:
path = "type.spec.verification"
if not isinstance(checks, list) or not checks:
issues.append(err(path, "must list checks that prove a case is live"))
return
expects = {c.get("expect") for c in checks if isinstance(c, dict)}
if "deny" not in expects:
issues.append(err(path, "needs at least one negative check (expect: deny)"))
if not expects & {"allow", "present"}:
issues.append(err(path, "needs at least one positive check (expect: allow or present)"))
for i, c in enumerate(checks):
if not isinstance(c, dict) or not c.get("id") or not c.get("check"):
issues.append(err(f"{path}[{i}]", "needs id, check and expect"))
if isinstance(c, dict) and c.get("reads_secret_value") is not False:
issues.append(err(f"{path}[{i}].reads_secret_value", "must be false: verification never reads a value"))
_validate_templates(checks, path, params, issues)
def _validate_rollback(steps: Any, step_ids: set[str], materialization: Any, issues: list[Issue]) -> None:
path = "type.spec.rollback"
if not isinstance(steps, list) or not steps:
issues.append(err(path, "must give the exact inverse of each step"))
return
covered = {s.get("reverses") for s in steps if isinstance(s, dict)}
mat_ids = {m.get("id") for m in materialization if isinstance(m, dict)} if isinstance(materialization, list) else set()
for missing in sorted((step_ids | mat_ids) - covered):
issues.append(err(path, f"no rollback reverses {missing!r}"))
def _validate_acceptance(acc: Any, issues: list[Issue]) -> None:
path = "type.spec.acceptance"
if not isinstance(acc, dict):
issues.append(err(path, "must be a mapping"))
return
who = acc.get("accepted_by")
if not isinstance(who, list) or not who or not set(who) <= ACCEPTERS:
issues.append(err(f"{path}.accepted_by", f"must be a non-empty subset of {sorted(ACCEPTERS)}"))
if acc.get("risk_class") not in RISK_CLASSES:
issues.append(err(f"{path}.risk_class", f"must be one of {sorted(RISK_CLASSES)}"))
if acc.get("risk_class") == "high" and "founder" not in (who or []):
issues.append(err(f"{path}.accepted_by", "a high-risk type needs founder acceptance of each case"))
# ── Cases ────────────────────────────────────────────────────────────────────
def load_type(types_root: Path, ref: str) -> tuple[list[Issue], dict[str, Any] | None, Path | None]:
m = TYPE_REF.fullmatch(ref or "")
if not m:
return [err("case.spec.type", "must be <type-id>@<major>.<minor>")], None, None
type_dir = types_root / m.group(1)
type_file = type_dir / "type.yaml"
if not type_file.is_file():
return [err("case.spec.type", f"unknown type {m.group(1)!r}")], None, None
data = yaml.safe_load(type_file.read_text()) or {}
if str(data.get("spec", {}).get("version")) != m.group(2):
return [err("case.spec.type", f"type {m.group(1)!r} is at version "
f"{data.get('spec', {}).get('version')}, not {m.group(2)}")], None, None
return [], data, type_dir
def validate_case(case: dict[str, Any], type_data: dict[str, Any], *, require_implemented: bool = True) -> list[Issue]:
issues: list[Issue] = []
if case.get("apiVersion") != CASE_API_VERSION:
issues.append(err("case.apiVersion", f"must be {CASE_API_VERSION}"))
if case.get("kind") != "SecurityCase":
issues.append(err("case.kind", "must be SecurityCase"))
spec = case.get("spec") or {}
if not ID.fullmatch(str(spec.get("id", ""))):
issues.append(err("case.spec.id", "must be lowercase letters, digits and hyphens"))
if spec.get("status") not in CASE_STATUSES:
issues.append(err("case.spec.status", f"must be one of {CASE_STATUSES}"))
owner = spec.get("owner") or {}
if not owner.get("repo") or not owner.get("person"):
issues.append(err("case.spec.owner", "needs repo and person"))
tspec = type_data.get("spec", {})
if require_implemented and tspec.get("status") != "implemented":
issues.append(err("case.spec.type", f"type {tspec.get('id')!r} is {tspec.get('status')!r}; "
"only implemented types accept cases"))
if spec.get("status") != "proposed":
acc = spec.get("acceptance") or {}
needed = set(tspec.get("acceptance", {}).get("accepted_by", []))
given = {a.get("role") for a in acc.get("by", []) if isinstance(a, dict)}
if not acc.get("at") or not acc.get("record") or not needed <= given:
issues.append(err("case.spec.acceptance", f"needs at, record and acceptance by {sorted(needed)}"))
params = {p["name"]: p for p in tspec.get("parameters", [])}
values = spec.get("parameters") or {}
for name in sorted(set(values) - set(params)):
issues.append(err(f"case.spec.parameters.{name}", "not a parameter of this type"))
for name, p in params.items():
path = f"case.spec.parameters.{name}"
if name not in values:
if p.get("case_settable") and p.get("required", True) and "default" not in p:
issues.append(err(path, "required"))
continue
if not p.get("case_settable"):
issues.append(err(path, "this type does not let cases set it"))
continue
issues.extend(_check_value(path, values[name], p))
for path, text in _walk_strings(spec, "case.spec"):
if secret_shaped(text):
issues.append(err(path, "contains a credential-shaped value; cases hold references, never secrets"))
return issues
def _check_value(path: str, value: Any, p: dict[str, Any]) -> list[Issue]:
issues: list[Issue] = []
kinds = {"string": str, "integer": int, "number": (int, float), "boolean": bool,
"array": list, "object": dict}
if not isinstance(value, kinds[p["type"]]) or (p["type"] in {"integer", "number"} and isinstance(value, bool)):
return [err(path, f"must be {p['type']}")]
c = p.get("constraints") or {}
items = value if isinstance(value, list) else [value]
for item in items:
if isinstance(item, str):
if "pattern" in c and not re.fullmatch(c["pattern"], item):
issues.append(err(path, f"{item!r} does not match {c['pattern']}"))
if p.get("sensitivity") == "secret_reference" and not CUSTODY_REFERENCE.fullmatch(item):
issues.append(err(path, "must be a custody reference such as bao:<mount>/<path>"))
if "enum" in c and item not in c["enum"]:
issues.append(err(path, f"{item!r} is not one of {c['enum']}"))
if isinstance(value, list):
if "min_items" in c and len(value) < c["min_items"]:
issues.append(err(path, f"needs at least {c['min_items']} items"))
if c.get("unique") and len(set(map(str, value))) != len(value):
issues.append(err(path, "items must be unique"))
return issues
# ── Rendering ────────────────────────────────────────────────────────────────
def effective_values(case: dict[str, Any], type_data: dict[str, Any]) -> dict[str, Any]:
values = {p["name"]: p.get("default") for p in type_data["spec"]["parameters"] if "default" in p}
values.update(case["spec"].get("parameters") or {})
return values
def _fill(value: Any, values: dict[str, Any]) -> Any:
if isinstance(value, str):
return value.format_map({k: (",".join(map(str, v)) if isinstance(v, list) else v) for k, v in values.items()})
if isinstance(value, dict):
return {k: _fill(v, values) for k, v in value.items()}
if isinstance(value, list):
return [_fill(v, values) for v in value]
return value
def render_case(case: dict[str, Any], type_data: dict[str, Any], declarations: list,
*, require_implemented: bool = True) -> tuple[list[Issue], dict[str, Any] | None]:
issues = validate_case(case, type_data, require_implemented=require_implemented)
issues += validate_type(type_data)
if has_errors(issues):
return issues, None
tspec, cspec = type_data["spec"], case["spec"]
values = effective_values(case, type_data)
comp = tspec["composition"]
overrides: dict[str, dict[str, Any]] = {}
for target, source in (comp.get("parameter_bindings") or {}).items():
decl_id, param = target.rsplit(".", 1)
overrides.setdefault(decl_id, {})[param] = values[source]
scenario = {
"id": f"scenario:case:{cspec['id']}",
"authority": comp["authority"],
"initial_trust": list(comp["initial_trust"]),
"requires": {"capabilities": list(comp["capabilities"])},
"providers": dict(comp["providers"]),
}
if overrides:
scenario["parameter_overrides"] = overrides
scenario_issues, plan = composer.compose_scenario(declarations, scenario)
issues.extend(scenario_issues)
if plan is None or has_errors(issues):
return issues, None
rendered = {
"apiVersion": RENDER_API_VERSION,
"kind": "SecurityCaseRender",
"case": cspec["id"],
"type": f"{tspec['id']}@{tspec['version']}",
"composition": plan,
"materialization": _fill(tspec["materialization"], values),
"follow_through": _fill(tspec["follow_through"], values),
"verification": _fill(tspec["verification"], values),
"rollback": _fill(tspec["rollback"], values),
"execution": {"permitted": False},
}
rendered["digest"] = "sha256:" + hashlib.sha256(
json.dumps(rendered, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
return issues, rendered
# ── CLI ──────────────────────────────────────────────────────────────────────
def _report(issues: list[Issue]) -> int:
for i in issues:
print(f"{i.level} {i.path}: {i.message}", file=sys.stderr)
return 1 if has_errors(issues) else 0
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
sub = ap.add_subparsers(dest="cmd", required=True)
vt = sub.add_parser("validate-type")
vt.add_argument("type_dir")
vc = sub.add_parser("validate-case")
vc.add_argument("case")
vc.add_argument("--types", required=True)
vc.add_argument("--allow-unimplemented", action="store_true",
help="validate an example case against a type that is not implemented yet")
rd = sub.add_parser("render")
rd.add_argument("case")
rd.add_argument("declarations", nargs="+")
rd.add_argument("--types", required=True)
rd.add_argument("--allow-unimplemented", action="store_true")
args = ap.parse_args(argv)
if args.cmd == "validate-type":
d = Path(args.type_dir)
data = yaml.safe_load((d / "type.yaml").read_text()) or {}
code = _report(validate_type(data, d))
if code == 0:
print(f"OK {data['spec']['id']}@{data['spec']['version']} ({data['spec']['status']})")
return code
case = yaml.safe_load(Path(args.case).read_text()) or {}
issues, type_data, _ = load_type(Path(args.types), (case.get("spec") or {}).get("type", ""))
if type_data is None:
return _report(issues)
if args.cmd == "validate-case":
code = _report(validate_case(case, type_data, require_implemented=not args.allow_unimplemented))
if code == 0:
print(f"OK case {case['spec']['id']}")
return code
declarations, load_issues = composer.load_declarations(args.declarations)
issues, rendered = render_case(case, type_data, declarations,
require_implemented=not args.allow_unimplemented)
issues = load_issues + issues
if rendered is None:
return _report(issues)
json.dump(rendered, sys.stdout, indent=2, sort_keys=True)
print()
return _report(issues)
if __name__ == "__main__":
raise SystemExit(main())

159
tools/security-case-types/tests/test_security_case_validator.py Normal file
View file

@ -0,0 +1,159 @@
import copy
import importlib.util
import sys
from pathlib import Path
import yaml
TOOL_DIR = Path(__file__).resolve().parents[1]
REPO = TOOL_DIR.parents[1]
SPEC = importlib.util.spec_from_file_location("security_case_validator", TOOL_DIR / "security_case_validator.py")
v = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = v
SPEC.loader.exec_module(v)
TEMPLATE_DIR = REPO / "security-case-types" / "_template"
TYPE = yaml.safe_load((TEMPLATE_DIR / "type.yaml").read_text())
CASE = yaml.safe_load((TEMPLATE_DIR / "examples" / "example-operator.yaml").read_text())
DECLS, _ = v.composer.load_declarations([str(REPO / "capabilities/playbooks/net-kingdom.local-identity.yaml")])
def msgs(issues):
return [f"{i.path}: {i.message}" for i in issues if i.level == "ERROR"]
def implemented(t=None):
t = copy.deepcopy(t or TYPE)
t["spec"]["status"] = "implemented"
return t
def test_template_type_validates():
assert msgs(v.validate_type(TYPE, TEMPLATE_DIR)) == []
def test_missing_section_is_refused():
t = copy.deepcopy(TYPE)
del t["spec"]["rollback"]
assert any("rollback" in m for m in msgs(v.validate_type(t)))
def test_unpinned_provider_is_refused():
t = copy.deepcopy(TYPE)
t["spec"]["composition"]["providers"] = {}
assert any("provider pin" in m for m in msgs(v.validate_type(t)))
def test_unknown_capability_is_refused():
t = copy.deepcopy(TYPE)
t["spec"]["composition"]["capabilities"] = ["c9.invented"]
assert any("unknown capability" in m for m in msgs(v.validate_type(t)))
def test_attended_step_needs_justification():
t = copy.deepcopy(TYPE)
del t["spec"]["follow_through"][0]["attended_justification"]
assert any("attended" in m for m in msgs(v.validate_type(t)))
def test_every_step_needs_a_rollback():
t = copy.deepcopy(TYPE)
t["spec"]["rollback"] = [r for r in t["spec"]["rollback"] if r["reverses"] != "init-store"]
assert any("init-store" in m for m in msgs(v.validate_type(t)))
def test_verification_needs_a_negative_check_and_never_reads_values():
t = copy.deepcopy(TYPE)
t["spec"]["verification"] = [c for c in t["spec"]["verification"] if c["expect"] != "deny"]
t["spec"]["verification"][0]["reads_secret_value"] = True
found = msgs(v.validate_type(t))
assert any("negative check" in m for m in found)
assert any("never reads a value" in m for m in found)
def test_template_with_unknown_parameter_is_refused():
t = copy.deepcopy(TYPE)
t["spec"]["materialization"][1]["template"] = "policy-{nonexistent}"
assert any("unknown parameter" in m for m in msgs(v.validate_type(t)))
def test_high_risk_type_needs_founder_acceptance():
t = copy.deepcopy(TYPE)
t["spec"]["acceptance"]["risk_class"] = "high"
assert any("founder" in m for m in msgs(v.validate_type(t)))
def test_credential_shaped_value_in_type_is_refused():
t = copy.deepcopy(TYPE)
t["spec"]["purpose"] = "token hvs.CAESIabcdefghijklmnop"
assert any("credential-shaped" in m for m in msgs(v.validate_type(t)))
def test_case_against_unimplemented_type_is_refused():
assert any("only implemented types" in m for m in msgs(v.validate_case(CASE, TYPE)))
assert msgs(v.validate_case(CASE, implemented())) == []
def test_case_cannot_carry_a_secret_value():
c = copy.deepcopy(CASE)
c["spec"]["parameters"]["signing_key_ref"] = "-----BEGIN PRIVATE KEY-----"
found = msgs(v.validate_case(c, implemented()))
assert any("custody reference" in m for m in found)
assert any("credential-shaped" in m for m in found)
def test_case_parameter_rules():
c = copy.deepcopy(CASE)
c["spec"]["parameters"]["operator_username"] = "Not Valid"
c["spec"]["parameters"]["store_home"] = "/tmp/elsewhere"
c["spec"]["parameters"]["surprise"] = 1
del c["spec"]["parameters"]["operator_email"]
found = msgs(v.validate_case(c, implemented()))
assert any("does not match" in m for m in found)
assert any("does not let cases set it" in m for m in found)
assert any("not a parameter of this type" in m for m in found)
assert any("operator_email: required" in m for m in found)
def test_accepted_case_needs_acceptance_record():
c = copy.deepcopy(CASE)
c["spec"]["status"] = "accepted"
assert any("acceptance" in m for m in msgs(v.validate_case(c, implemented())))
c["spec"]["acceptance"] = {"at": "2026-10-08", "record": "decision:x", "by": [{"role": "owner", "name": "a"}]}
assert msgs(v.validate_case(c, implemented())) == []
def test_render_is_deterministic_plan_only_and_substitutes_parameters():
issues, r = v.render_case(CASE, implemented(), DECLS)
assert msgs(issues) == []
assert r["execution"] == {"permitted": False}
assert r["materialization"][1]["template"] == "local-identity-bootstrap-admin-read"
assert r["rollback"][1]["action"].endswith("bao:platform/workloads/example/local-identity-signing")
_, again = v.render_case(CASE, implemented(), DECLS)
assert r["digest"] == again["digest"]
def test_load_type_checks_reference_and_version(tmp_path):
root = tmp_path / "types"
(root / "template-example").mkdir(parents=True)
(root / "template-example" / "type.yaml").write_text(yaml.safe_dump(TYPE))
issues, data, _ = v.load_type(root, "template-example@0.1")
assert data is not None and not msgs(issues)
issues, data, _ = v.load_type(root, "template-example@1.0")
assert data is None and any("version" in m for m in msgs(issues))
issues, data, _ = v.load_type(root, "nope")
assert data is None
def test_template_conforms_to_the_json_schemas():
import json
import pytest
jsonschema = pytest.importorskip("jsonschema")
type_schema = json.loads((REPO / "canon/schemas/security-case-type_v0.1.schema.json").read_text())
case_schema = json.loads((REPO / "canon/schemas/security-case_v0.1.schema.json").read_text())
jsonschema.validate(TYPE, type_schema)
jsonschema.validate(CASE, case_schema)
bad = copy.deepcopy(CASE)
bad["spec"]["surprise"] = True
with pytest.raises(jsonschema.ValidationError):
jsonschema.validate(bad, case_schema)