tegwick
d96aab2321
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02929-244b-7391-b933-c04010e8eedb
445 lines
16 KiB
Python
445 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Compose a deterministic, plan-only NetKingdom security scenario."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
TOOL_DIR = Path(__file__).resolve().parent
|
|
CONTRACT_TOOL_DIR = TOOL_DIR.parent / "playbook-capability-contract"
|
|
sys.path.insert(0, str(CONTRACT_TOOL_DIR))
|
|
|
|
import playbook_contract_validator as contract # noqa: E402
|
|
|
|
|
|
PLAN_API_VERSION = "netkingdom.io/security-scenario-composition/v0.1"
|
|
PLAN_KIND = "SecurityScenarioComposition"
|
|
SCENARIO_ID = re.compile(r"^scenario:[a-z0-9][a-z0-9._:-]*$")
|
|
SCENARIO_KEYS = {
|
|
"id",
|
|
"authority",
|
|
"initial_trust",
|
|
"requires",
|
|
"providers",
|
|
"parameter_overrides",
|
|
}
|
|
|
|
Issue = contract.Issue
|
|
Declaration = contract.Declaration
|
|
|
|
|
|
def error(path: str, message: str) -> Issue:
|
|
return Issue("ERROR", path, message)
|
|
|
|
|
|
def has_errors(issues: list[Issue]) -> bool:
|
|
return any(item.level == "ERROR" for item in issues)
|
|
|
|
|
|
def validate_scenario(scenario: dict[str, Any]) -> list[Issue]:
|
|
issues: list[Issue] = []
|
|
unknown_keys = sorted(set(scenario) - SCENARIO_KEYS)
|
|
if unknown_keys:
|
|
issues.append(error("scenario", f"unknown fields: {unknown_keys}"))
|
|
|
|
scenario_id = scenario.get("id")
|
|
if not isinstance(scenario_id, str) or not SCENARIO_ID.fullmatch(scenario_id):
|
|
issues.append(error("scenario.id", "must match 'scenario:<lowercase-id>'"))
|
|
|
|
authority = scenario.get("authority")
|
|
if authority not in contract.SCENARIO_AUTHORITIES:
|
|
issues.append(error("scenario.authority", f"unknown authority {authority!r}"))
|
|
|
|
requires = scenario.get("requires")
|
|
if not isinstance(requires, dict):
|
|
issues.append(error("scenario.requires", "must be an object"))
|
|
required_caps: list[Any] = []
|
|
else:
|
|
unknown_requires = sorted(set(requires) - {"capabilities"})
|
|
if unknown_requires:
|
|
issues.append(error("scenario.requires", f"unknown fields: {unknown_requires}"))
|
|
required_caps = requires.get("capabilities", [])
|
|
if not isinstance(required_caps, list) or not required_caps:
|
|
issues.append(error("scenario.requires.capabilities", "must be a non-empty list"))
|
|
required_caps = []
|
|
|
|
seen_caps: set[str] = set()
|
|
for index, cap_id in enumerate(required_caps):
|
|
path = f"scenario.requires.capabilities[{index}]"
|
|
if cap_id not in contract.CAPABILITIES:
|
|
issues.append(error(path, f"unknown capability id {cap_id!r}"))
|
|
if cap_id in seen_caps:
|
|
issues.append(error(path, f"duplicate capability id {cap_id!r}"))
|
|
if isinstance(cap_id, str):
|
|
seen_caps.add(cap_id)
|
|
|
|
initial_trust = scenario.get("initial_trust", [])
|
|
if not isinstance(initial_trust, list):
|
|
issues.append(error("scenario.initial_trust", "must be a list"))
|
|
initial_trust = []
|
|
seen_trust: set[str] = set()
|
|
for index, state in enumerate(initial_trust):
|
|
path = f"scenario.initial_trust[{index}]"
|
|
if state not in contract.TRUST_STATES:
|
|
issues.append(error(path, f"unknown trust state {state!r}"))
|
|
if state in seen_trust:
|
|
issues.append(error(path, f"duplicate trust state {state!r}"))
|
|
if isinstance(state, str):
|
|
seen_trust.add(state)
|
|
|
|
providers = scenario.get("providers", {})
|
|
if not isinstance(providers, dict):
|
|
issues.append(error("scenario.providers", "must be an object"))
|
|
providers = {}
|
|
for cap_id, declaration_id in providers.items():
|
|
path = f"scenario.providers.{cap_id}"
|
|
if cap_id not in seen_caps:
|
|
issues.append(error(path, "provider pin targets an unrequested capability"))
|
|
if not isinstance(declaration_id, str) or not declaration_id:
|
|
issues.append(error(path, "provider declaration id must be a non-empty string"))
|
|
|
|
overrides = scenario.get("parameter_overrides", {})
|
|
if not isinstance(overrides, dict):
|
|
issues.append(error("scenario.parameter_overrides", "must be an object"))
|
|
else:
|
|
for declaration_id, values in overrides.items():
|
|
if not isinstance(declaration_id, str) or not declaration_id:
|
|
issues.append(error("scenario.parameter_overrides", "declaration ids must be non-empty strings"))
|
|
if not isinstance(values, dict):
|
|
issues.append(error(f"scenario.parameter_overrides.{declaration_id}", "must be an object"))
|
|
|
|
return issues
|
|
|
|
|
|
def validate_declarations(declarations: list[Declaration]) -> list[Issue]:
|
|
issues: list[Issue] = []
|
|
ids: dict[str, list[str]] = {}
|
|
for declaration in declarations:
|
|
for item in contract.validate_declaration(declaration):
|
|
issues.append(
|
|
Issue(
|
|
item.level,
|
|
f"declarations[{declaration.path}].{item.path}",
|
|
item.message,
|
|
)
|
|
)
|
|
ids.setdefault(declaration.id, []).append(str(declaration.path))
|
|
|
|
for declaration_id, paths in sorted(ids.items()):
|
|
if not declaration_id:
|
|
continue
|
|
if len(paths) > 1:
|
|
issues.append(
|
|
error(
|
|
"declarations",
|
|
f"duplicate declaration id {declaration_id!r}: {sorted(paths)}",
|
|
)
|
|
)
|
|
return issues
|
|
|
|
|
|
def declaration_trust(declaration: Declaration, section: str) -> list[dict[str, Any]]:
|
|
trust = declaration.data.get("spec", {}).get("trust", {})
|
|
values = trust.get(section, []) if isinstance(trust, dict) else []
|
|
return values if isinstance(values, list) else []
|
|
|
|
|
|
def trust_states(declaration: Declaration, section: str) -> set[str]:
|
|
return {
|
|
str(item.get("state"))
|
|
for item in declaration_trust(declaration, section)
|
|
if isinstance(item, dict) and item.get("state")
|
|
}
|
|
|
|
|
|
def stable_digest(scenario: dict[str, Any], selected: dict[str, Declaration]) -> str:
|
|
payload = {
|
|
"scenario": scenario,
|
|
"declarations": {
|
|
declaration_id: selected[declaration_id].data
|
|
for declaration_id in sorted(selected)
|
|
},
|
|
}
|
|
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
return f"sha256:{hashlib.sha256(encoded).hexdigest()}"
|
|
|
|
|
|
def select_declarations(
|
|
declarations: list[Declaration],
|
|
scenario: dict[str, Any],
|
|
) -> tuple[list[Issue], dict[str, Declaration], dict[str, str]]:
|
|
issues: list[Issue] = []
|
|
required_caps = scenario["requires"]["capabilities"]
|
|
pins = scenario.get("providers", {})
|
|
selected: dict[str, Declaration] = {}
|
|
capability_providers: dict[str, str] = {}
|
|
|
|
for cap_id in required_caps:
|
|
matches = sorted(
|
|
(declaration for declaration in declarations if cap_id in declaration.capabilities),
|
|
key=lambda declaration: declaration.id,
|
|
)
|
|
pin = pins.get(cap_id)
|
|
if pin is not None:
|
|
pinned = [declaration for declaration in matches if declaration.id == pin]
|
|
if not pinned:
|
|
available = [declaration.id for declaration in matches]
|
|
issues.append(
|
|
error(
|
|
f"scenario.providers.{cap_id}",
|
|
f"pinned declaration {pin!r} does not provide {cap_id!r}; available={available}",
|
|
)
|
|
)
|
|
continue
|
|
chosen = pinned[0]
|
|
elif not matches:
|
|
issues.append(
|
|
error(
|
|
"scenario.requires.capabilities",
|
|
f"no declaration provides {cap_id!r}",
|
|
)
|
|
)
|
|
continue
|
|
elif len(matches) > 1:
|
|
issues.append(
|
|
error(
|
|
f"scenario.providers.{cap_id}",
|
|
f"ambiguous providers {[item.id for item in matches]}; pin one explicitly",
|
|
)
|
|
)
|
|
continue
|
|
else:
|
|
chosen = matches[0]
|
|
|
|
selected[chosen.id] = chosen
|
|
capability_providers[str(cap_id)] = chosen.id
|
|
|
|
return issues, selected, capability_providers
|
|
|
|
|
|
def effective_parameters(
|
|
declaration: Declaration,
|
|
overrides: dict[str, Any],
|
|
authority: str,
|
|
) -> tuple[list[Issue], dict[str, Any]]:
|
|
issues: list[Issue] = []
|
|
params_out: dict[str, Any] = {}
|
|
for name in sorted(overrides):
|
|
if name not in declaration.parameters:
|
|
issues.append(
|
|
error(
|
|
f"scenario.parameter_overrides.{declaration.id}.{name}",
|
|
"unknown parameter override",
|
|
)
|
|
)
|
|
|
|
for name, param in sorted(declaration.parameters.items()):
|
|
overridden, value = contract.effective_parameter_value(param, overrides, declaration.id)
|
|
path = f"scenario.parameter_overrides.{declaration.id}.{name}"
|
|
if param.get("required") is True and value is None:
|
|
issues.append(error(path, "required parameter has no default or override"))
|
|
if overridden:
|
|
issues.extend(contract.validate_override_allowed(param, value, authority, path))
|
|
params_out[name] = {
|
|
"value": value,
|
|
"source": "override" if overridden else "default",
|
|
"sensitivity": param.get("sensitivity"),
|
|
"tuning_authority": param.get("tuning_authority"),
|
|
}
|
|
return issues, params_out
|
|
|
|
|
|
def sequence_declarations(
|
|
selected: dict[str, Declaration],
|
|
initial_trust: set[str],
|
|
) -> tuple[list[Issue], list[Declaration], set[str]]:
|
|
issues: list[Issue] = []
|
|
established = set(initial_trust)
|
|
remaining = dict(selected)
|
|
ordered: list[Declaration] = []
|
|
|
|
while remaining:
|
|
eligible = sorted(
|
|
(
|
|
declaration
|
|
for declaration in remaining.values()
|
|
if trust_states(declaration, "requires") <= established
|
|
),
|
|
key=lambda declaration: declaration.id,
|
|
)
|
|
if not eligible:
|
|
for declaration_id, declaration in sorted(remaining.items()):
|
|
missing = sorted(trust_states(declaration, "requires") - established)
|
|
issues.append(
|
|
error(
|
|
f"composition.execution_steps.{declaration_id}",
|
|
f"cannot sequence declaration; unresolved trust states: {missing}",
|
|
)
|
|
)
|
|
break
|
|
chosen = eligible[0]
|
|
ordered.append(chosen)
|
|
established.update(trust_states(chosen, "satisfies"))
|
|
del remaining[chosen.id]
|
|
|
|
return issues, ordered, established
|
|
|
|
|
|
def compose_scenario(
|
|
declarations: list[Declaration],
|
|
scenario: dict[str, Any],
|
|
) -> tuple[list[Issue], dict[str, Any] | None]:
|
|
issues = validate_scenario(scenario)
|
|
issues.extend(validate_declarations(declarations))
|
|
if has_errors(issues):
|
|
return issues, None
|
|
|
|
selection_issues, selected, capability_providers = select_declarations(declarations, scenario)
|
|
issues.extend(selection_issues)
|
|
if has_errors(issues):
|
|
return issues, None
|
|
|
|
overrides = scenario.get("parameter_overrides", {})
|
|
for declaration_id in sorted(overrides):
|
|
if declaration_id not in selected:
|
|
issues.append(
|
|
error(
|
|
f"scenario.parameter_overrides.{declaration_id}",
|
|
"override targets an unselected declaration",
|
|
)
|
|
)
|
|
|
|
parameters: dict[str, dict[str, Any]] = {}
|
|
authority = str(scenario["authority"])
|
|
for declaration_id, declaration in sorted(selected.items()):
|
|
param_issues, values = effective_parameters(
|
|
declaration,
|
|
overrides.get(declaration_id, {}),
|
|
authority,
|
|
)
|
|
issues.extend(param_issues)
|
|
parameters[declaration_id] = values
|
|
|
|
sequence_issues, ordered, planned_trust = sequence_declarations(
|
|
selected,
|
|
set(scenario.get("initial_trust", [])),
|
|
)
|
|
issues.extend(sequence_issues)
|
|
if has_errors(issues):
|
|
return issues, None
|
|
|
|
execution_steps: list[dict[str, Any]] = []
|
|
responsibilities: list[dict[str, Any]] = []
|
|
for index, declaration in enumerate(ordered, start=1):
|
|
metadata = declaration.data["metadata"]
|
|
spec = declaration.data["spec"]
|
|
required = declaration_trust(declaration, "requires")
|
|
satisfied = declaration_trust(declaration, "satisfies")
|
|
readiness = [
|
|
{
|
|
"state": item["state"],
|
|
"checks": item.get("readiness_checks", []),
|
|
}
|
|
for item in satisfied
|
|
]
|
|
execution_steps.append(
|
|
{
|
|
"order": index,
|
|
"declaration_id": declaration.id,
|
|
"executor_owner": metadata["owner"],
|
|
"repo": metadata["repo"],
|
|
"capabilities": sorted(
|
|
cap_id
|
|
for cap_id, provider in capability_providers.items()
|
|
if provider == declaration.id
|
|
),
|
|
"entry_point": spec["playbook"],
|
|
"parameters": parameters[declaration.id],
|
|
"requires_trust": required,
|
|
"satisfies_trust": satisfied,
|
|
"readiness_obligations": readiness,
|
|
}
|
|
)
|
|
for responsibility in spec["responsibilities"]:
|
|
responsibilities.append(
|
|
{
|
|
"declaration_id": declaration.id,
|
|
**responsibility,
|
|
}
|
|
)
|
|
|
|
composition = {
|
|
"apiVersion": PLAN_API_VERSION,
|
|
"kind": PLAN_KIND,
|
|
"scenario": scenario["id"],
|
|
"authority": authority,
|
|
"composition_digest": stable_digest(scenario, selected),
|
|
"requested_capabilities": list(scenario["requires"]["capabilities"]),
|
|
"capability_providers": capability_providers,
|
|
"initial_trust": sorted(scenario.get("initial_trust", [])),
|
|
"planned_final_trust": sorted(planned_trust),
|
|
"execution": {
|
|
"mode": "plan-only",
|
|
"permitted": False,
|
|
"reason": "Composition does not authorize or perform provider execution.",
|
|
},
|
|
"execution_steps": execution_steps,
|
|
"responsibility_map": responsibilities,
|
|
}
|
|
return issues, composition
|
|
|
|
|
|
def load_declarations(paths: list[str]) -> tuple[list[Declaration], list[Issue]]:
|
|
declarations: list[Declaration] = []
|
|
issues: list[Issue] = []
|
|
for raw_path in paths:
|
|
path = Path(raw_path)
|
|
try:
|
|
data = contract.load_yaml(path)
|
|
except Exception as exc:
|
|
issues.append(error(f"declarations[{path}]", f"failed to load: {exc}"))
|
|
continue
|
|
declarations.append(Declaration(path=path, data=data))
|
|
return declarations, issues
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
description="Compose a deterministic, non-executing NetKingdom security scenario."
|
|
)
|
|
parser.add_argument("declarations", nargs="+", help="Playbook capability declaration YAML files")
|
|
parser.add_argument("--scenario", required=True, help="Security scenario YAML file")
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
declarations, issues = load_declarations(args.declarations)
|
|
try:
|
|
scenario = contract.load_yaml(Path(args.scenario))
|
|
except Exception as exc:
|
|
issues.append(error("scenario", f"failed to load {args.scenario}: {exc}"))
|
|
scenario = {}
|
|
|
|
composition: dict[str, Any] | None = None
|
|
if not has_errors(issues):
|
|
composition_issues, composition = compose_scenario(declarations, scenario)
|
|
issues.extend(composition_issues)
|
|
|
|
payload = {
|
|
"ok": not has_errors(issues),
|
|
"issues": [item.__dict__ for item in issues],
|
|
"composition": composition,
|
|
}
|
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
|
return 1 if has_errors(issues) else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|