feat: inventory schema 0.3 and close WP-0006 T01 T02 T04

v0.2 records stay valid. v0.3 requires the five facets. Validators
reject inline Scaleway endpoints and secret-looking strings. The
backup record is the first 0.3 object. Reef views already met T02.
This commit is contained in:
tegwick 2026-08-15 02:40:43 +02:00
parent a9830f3109
commit 34a014a896
8 changed files with 140 additions and 12 deletions

View file

@ -12,6 +12,24 @@ from pathlib import Path
from entities import association_ok
RESOURCE_ID = re.compile(r"^resource:[a-z0-9][a-z0-9:_-]+$")
INLINE_ENDPOINT = re.compile(
r"(https?://\S*s3\S*|https?://\S+\.scw\.cloud|\bs3://)",
re.IGNORECASE,
)
INLINE_SECRET = re.compile(
r"(BEGIN [A-Z ]*PRIVATE KEY|AGE-SECRET-KEY-|AKIA[0-9A-Z]{16}|sk_live_|SCW[A-Z0-9]{17})"
)
V03_FACETS = (
"description", "decision", "operational_refs", "credential_handles", "consumers",
)
# Evidence and decision refs may cite provider HTTPS docs. Those are not
# operating attributes and must not be treated as inline endpoints.
SKIP_INLINE_PATHS = {
("evidence", "ref"),
("decision", "ref"),
("requirements", "ref"),
("cost", "price_evidence"),
}
STATUSES = {
"proposed", "ordered", "commissioning", "active", "suspended",
"retiring", "retired", "rejected",
@ -39,9 +57,27 @@ def validate_transition(current: str, target: str) -> None:
raise ValueError(f"invalid lifecycle transition {current} -> {target}")
def _walk_strings(node, path=()):
if isinstance(node, dict):
for key, value in node.items():
yield from _walk_strings(value, path + (key,))
elif isinstance(node, list):
for value in node:
yield from _walk_strings(value, path)
elif isinstance(node, str):
yield path, node
def _skip_inline(path: tuple) -> bool:
if path and path[0] in {"operational_refs", "credential_handles"}:
return True
return len(path) >= 2 and (path[0], path[-1]) in SKIP_INLINE_PATHS
def validate_record(record: dict) -> None:
if record.get("schema_version") != "0.2":
raise ValueError("portfolio records must use schema_version 0.2")
version = record.get("schema_version")
if version not in {"0.2", "0.3"}:
raise ValueError("portfolio records must use schema_version 0.2 or 0.3")
if not RESOURCE_ID.fullmatch(record.get("id", "")):
raise ValueError("invalid resource id")
if record.get("status") not in STATUSES:
@ -51,6 +87,14 @@ def validate_record(record: dict) -> None:
association_ok(record)
if version == "0.3":
missing = [name for name in V03_FACETS if name not in record]
if missing:
raise ValueError("schema 0.3 requires five facets: " + ",".join(missing))
consumers = record.get("consumers") or {}
if "potential" not in consumers or "actual" not in consumers:
raise ValueError("schema 0.3 consumers must list potential and actual")
decision = record.get("decision")
if record.get("status") in {"ordered", "commissioning"} and (
not decision or decision.get("status") != "approved"
@ -63,6 +107,20 @@ def validate_record(record: dict) -> None:
if not str(ref).startswith("secret:"):
raise ValueError(f"credential_handles must be secret: references: {ref}")
for path, value in _walk_strings(record):
if _skip_inline(path):
continue
if INLINE_ENDPOINT.search(value):
raise ValueError(
"inline provider endpoint must be a reef: reference, not "
+ ".".join(str(part) for part in path)
)
if INLINE_SECRET.search(value):
raise ValueError(
"inline secret material is forbidden; use a secret: handle at "
+ ".".join(str(part) for part in path)
)
allocation = record["ownership"]["allocation"]
if allocation["mode"] == "unattributed":
if allocation["cost_attribution_key"] is not None:

View file

@ -41,11 +41,16 @@ def main() -> int:
assert forecast["record_type"] == "forecast"
assert len({row["period"] for row in forecast["rows"]}) == len(forecast["rows"])
assert all(row["total_eur"] == round(row["infrastructure_eur"] + row["internal_labor_eur"], 2) for row in forecast["rows"])
assert schema["properties"]["schema_version"]["const"] == "0.2"
assert set(schema["properties"]["schema_version"]["enum"]) == {"0.2", "0.3"}
required = set(schema["required"])
v03 = {"description", "decision", "operational_refs", "credential_handles", "consumers"}
for resource in resources:
assert not required - resource.keys(), f"missing fields: {required - resource.keys()}"
need = set(required)
if resource.get("schema_version") == "0.3":
need |= v03
assert not need - resource.keys(), f"missing fields: {need - resource.keys()}"
validate_record(resource)
assert any(resource.get("schema_version") == "0.3" for resource in resources)
operational_records = [load(str(path)) for path in Path("data/control-cycle").glob("*.json")]
control_records += operational_records
record_ids = {record["record_id"] for record in control_records}
@ -126,6 +131,8 @@ def main() -> int:
if view["reef_id"] == "reef-railiance":
assert view["role"] == "compute_substrate"
assert any("reef-storage" in note for note in view.get("notes") or [])
if view["reef_id"] == "reef-storage":
assert view["role"] == "storage_substrate"
print("resource-control declarations: valid")
return 0