716 lines
27 KiB
Python
716 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate the versioned Crank Community Capability Inventory contract."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import heapq
|
|
import json
|
|
import re
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Any, Iterable, Iterator
|
|
|
|
|
|
SCHEMA_DIALECT = "https://json-schema.org/draft/2020-12/schema"
|
|
MAX_DOCUMENT_BYTES = 4_194_304
|
|
MAX_FLOWS = 10_000
|
|
MAX_REQUIREMENTS_PER_FLOW = 64
|
|
MAX_CAPABILITIES_PER_FLOW = 16
|
|
MAX_EVIDENCE_LINKS_PER_KIND = 64
|
|
MAX_ID_LENGTH = 128
|
|
MAX_REQUIREMENT_LENGTH = 128
|
|
MAX_CAPABILITY_LENGTH = 128
|
|
MAX_OWNER_LENGTH = 256
|
|
MAX_OUTCOME_LENGTH = 4_096
|
|
MAX_NOTES_LENGTH = 4_096
|
|
MAX_EVIDENCE_PATH_LENGTH = 1_024
|
|
MAX_DIAGNOSTICS = 1_000
|
|
MAX_REPORT_BYTES = 65_536
|
|
|
|
FLOW_TYPES = ("ui", "api", "mcp")
|
|
FLOW_STATUSES = ("implemented", "planned", "gap", "blocked")
|
|
ALLOWED_CAPABILITIES = ("tools", "resources", "prompts", "tasks", "load_runs")
|
|
FORBIDDEN_CAPABILITIES = (
|
|
"multi_workspace", # community-scope: allow=multi-workspace
|
|
"enterprise_rbac", # community-scope: allow=enterprise
|
|
"sso", # community-scope: allow=sso
|
|
"non_rest_upstream", # community-scope: allow=non-rest-upstream
|
|
"arbitrary_distributed_load_targets", # community-scope: allow=distributed-load-targets
|
|
)
|
|
|
|
TOP_LEVEL_FIELDS = frozenset({"schema_version", "product", "flows"})
|
|
FLOW_FIELDS = frozenset(
|
|
{
|
|
"id",
|
|
"type",
|
|
"requirements",
|
|
"user_outcome",
|
|
"owner",
|
|
"status",
|
|
"capabilities",
|
|
"evidence",
|
|
"notes",
|
|
}
|
|
)
|
|
EVIDENCE_FIELDS = frozenset({"automated", "manual"})
|
|
|
|
SCHEMA_ALLOWED_KEYS: dict[tuple[str, ...], frozenset[str]] = {
|
|
(): frozenset(
|
|
{
|
|
"$schema",
|
|
"$id",
|
|
"title",
|
|
"type",
|
|
"additionalProperties",
|
|
"required",
|
|
"properties",
|
|
"$defs",
|
|
}
|
|
),
|
|
("properties",): frozenset({"schema_version", "product", "flows"}),
|
|
("properties", "schema_version"): frozenset({"const"}),
|
|
("properties", "product"): frozenset({"const"}),
|
|
("properties", "flows"): frozenset({"type", "minItems", "maxItems", "items"}),
|
|
("properties", "flows", "items"): frozenset({"$ref"}),
|
|
("$defs",): frozenset({"flow", "evidence", "evidencePaths"}),
|
|
("$defs", "flow"): frozenset(
|
|
{"type", "additionalProperties", "required", "properties"}
|
|
),
|
|
("$defs", "flow", "properties"): FLOW_FIELDS,
|
|
("$defs", "flow", "properties", "id"): frozenset(
|
|
{"type", "minLength", "maxLength", "pattern"}
|
|
),
|
|
("$defs", "flow", "properties", "type"): frozenset({"enum"}),
|
|
("$defs", "flow", "properties", "requirements"): frozenset(
|
|
{"type", "minItems", "maxItems", "uniqueItems", "items"}
|
|
),
|
|
("$defs", "flow", "properties", "requirements", "items"): frozenset(
|
|
{"type", "minLength", "maxLength", "pattern"}
|
|
),
|
|
("$defs", "flow", "properties", "user_outcome"): frozenset(
|
|
{"type", "minLength", "maxLength", "pattern"}
|
|
),
|
|
("$defs", "flow", "properties", "owner"): frozenset(
|
|
{"type", "minLength", "maxLength", "pattern"}
|
|
),
|
|
("$defs", "flow", "properties", "status"): frozenset({"enum"}),
|
|
("$defs", "flow", "properties", "capabilities"): frozenset(
|
|
{"type", "minItems", "maxItems", "uniqueItems", "items"}
|
|
),
|
|
("$defs", "flow", "properties", "capabilities", "items"): frozenset({"enum"}),
|
|
("$defs", "flow", "properties", "evidence"): frozenset({"$ref"}),
|
|
("$defs", "flow", "properties", "notes"): frozenset({"type", "maxLength"}),
|
|
("$defs", "evidence"): frozenset(
|
|
{"type", "additionalProperties", "required", "properties"}
|
|
),
|
|
("$defs", "evidence", "properties"): EVIDENCE_FIELDS,
|
|
("$defs", "evidence", "properties", "automated"): frozenset({"$ref"}),
|
|
("$defs", "evidence", "properties", "manual"): frozenset({"$ref"}),
|
|
("$defs", "evidencePaths"): frozenset(
|
|
{"type", "minItems", "maxItems", "uniqueItems", "items"}
|
|
),
|
|
("$defs", "evidencePaths", "items"): frozenset(
|
|
{"type", "minLength", "maxLength", "pattern"}
|
|
),
|
|
}
|
|
|
|
FLOW_ID = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
|
|
REQUIREMENT_ID = re.compile(r"^FR-[1-9][0-9]*$")
|
|
NON_WHITESPACE_PATTERN = r"[\s\S]*\S[\s\S]*"
|
|
EVIDENCE_PATH_PATTERN = r"^(?!/)(?!.*//)(?!.*(?:^|/)\.\.?($|/))(?!.*\\).+$"
|
|
|
|
|
|
@dataclass(frozen=True, order=True)
|
|
class Diagnostic:
|
|
code: str
|
|
pointer: str
|
|
message: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _ReverseDiagnostic:
|
|
diagnostic: Diagnostic
|
|
|
|
def __lt__(self, other: "_ReverseDiagnostic") -> bool:
|
|
return self.diagnostic > other.diagnostic
|
|
|
|
|
|
class DiagnosticCollector:
|
|
"""Retain only the lexicographically first diagnostics with an exact omitted count."""
|
|
|
|
def __init__(self) -> None:
|
|
self._heap: list[_ReverseDiagnostic] = []
|
|
self.total = 0
|
|
|
|
def append(self, diagnostic: Diagnostic) -> None:
|
|
self.total += 1
|
|
wrapped = _ReverseDiagnostic(diagnostic)
|
|
if len(self._heap) < MAX_DIAGNOSTICS:
|
|
heapq.heappush(self._heap, wrapped)
|
|
elif diagnostic < self._heap[0].diagnostic:
|
|
heapq.heapreplace(self._heap, wrapped)
|
|
|
|
def extend(self, diagnostics: Iterable[Diagnostic] | "DiagnosticCollector") -> None:
|
|
if isinstance(diagnostics, DiagnosticCollector):
|
|
for diagnostic in diagnostics.ordered():
|
|
self.append(diagnostic)
|
|
self.total += diagnostics.omitted
|
|
return
|
|
for diagnostic in diagnostics:
|
|
self.append(diagnostic)
|
|
|
|
@property
|
|
def omitted(self) -> int:
|
|
return self.total - len(self._heap)
|
|
|
|
def ordered(self) -> list[Diagnostic]:
|
|
return sorted(item.diagnostic for item in self._heap)
|
|
|
|
def __bool__(self) -> bool:
|
|
return self.total > 0
|
|
|
|
|
|
class DuplicateJsonMemberError(ValueError):
|
|
pass
|
|
|
|
|
|
def reject_duplicate_json_members(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
|
|
value: dict[str, Any] = {}
|
|
for key, item in pairs:
|
|
if key in value:
|
|
raise DuplicateJsonMemberError("duplicate JSON object member")
|
|
value[key] = item
|
|
return value
|
|
|
|
|
|
def reject_nonstandard_json_constant(_: str) -> None:
|
|
raise ValueError("non-standard JSON numeric constant")
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--root", type=Path, default=Path.cwd())
|
|
parser.add_argument("--inventory", required=True)
|
|
parser.add_argument("--schema", required=True)
|
|
parser.add_argument("--required-fr", action="append", default=[])
|
|
return parser.parse_args()
|
|
|
|
|
|
def path_has_symlink(root: Path, logical: PurePosixPath) -> bool:
|
|
current = root
|
|
for part in logical.parts:
|
|
current = current / part
|
|
if current.is_symlink():
|
|
return True
|
|
return False
|
|
|
|
|
|
def resolve_regular_file(root: Path, raw: str) -> Path | None:
|
|
if not raw or len(raw) > MAX_EVIDENCE_PATH_LENGTH or "\\" in raw:
|
|
return None
|
|
logical = PurePosixPath(raw)
|
|
if logical.is_absolute() or any(part in {"", ".", ".."} for part in logical.parts):
|
|
return None
|
|
if str(logical) != raw or path_has_symlink(root, logical):
|
|
return None
|
|
candidate = root.joinpath(*logical.parts)
|
|
try:
|
|
resolved = candidate.resolve(strict=True)
|
|
resolved.relative_to(root)
|
|
except (FileNotFoundError, OSError, RuntimeError, ValueError):
|
|
return None
|
|
if not resolved.is_file():
|
|
return None
|
|
return resolved
|
|
|
|
|
|
def read_json_document(
|
|
root: Path,
|
|
raw_path: str,
|
|
invalid_code: str,
|
|
) -> tuple[Any | None, Diagnostic | None]:
|
|
path = resolve_regular_file(root, raw_path)
|
|
pointer = "/schema" if invalid_code == "INVALID_SCHEMA" else "/inventory"
|
|
if path is None:
|
|
return None, Diagnostic(invalid_code, pointer, "document path is unavailable")
|
|
try:
|
|
if path.stat().st_size > MAX_DOCUMENT_BYTES:
|
|
return None, Diagnostic("INPUT_TOO_LARGE", pointer, "document exceeds limit")
|
|
data = path.read_bytes()
|
|
if len(data) > MAX_DOCUMENT_BYTES:
|
|
return None, Diagnostic("INPUT_TOO_LARGE", pointer, "document exceeds limit")
|
|
text = data.decode("utf-8")
|
|
return (
|
|
json.loads(
|
|
text,
|
|
object_pairs_hook=reject_duplicate_json_members,
|
|
parse_constant=reject_nonstandard_json_constant,
|
|
),
|
|
None,
|
|
)
|
|
except (OSError, UnicodeDecodeError, ValueError, RecursionError, json.JSONDecodeError):
|
|
return None, Diagnostic(invalid_code, pointer, "document is not valid UTF-8 JSON")
|
|
|
|
|
|
def nested(document: Any, *parts: str) -> Any:
|
|
current = document
|
|
for part in parts:
|
|
if not isinstance(current, dict) or part not in current:
|
|
return None
|
|
current = current[part]
|
|
return current
|
|
|
|
|
|
def contract_values_equal(actual: Any, expected: Any) -> bool:
|
|
if isinstance(expected, list):
|
|
return (
|
|
isinstance(actual, list)
|
|
and all(isinstance(item, str) for item in actual)
|
|
and len(actual) == len(set(actual))
|
|
and set(actual) == set(expected)
|
|
)
|
|
return type(actual) is type(expected) and actual == expected
|
|
|
|
|
|
def validate_required_keyword(
|
|
schema: dict[str, Any], parts: tuple[str, ...], expected: frozenset[str]
|
|
) -> list[Diagnostic]:
|
|
raw = nested(schema, *parts)
|
|
valid = (
|
|
isinstance(raw, list)
|
|
and all(isinstance(item, str) for item in raw)
|
|
and len(raw) == len(set(raw))
|
|
and set(raw) == expected
|
|
)
|
|
if valid:
|
|
return []
|
|
return [
|
|
Diagnostic(
|
|
"INVALID_SCHEMA_CONTRACT",
|
|
"/schema/" + "/".join(parts),
|
|
"required fields differ from validator",
|
|
)
|
|
]
|
|
|
|
|
|
def validate_schema_contract(schema: Any) -> list[Diagnostic]:
|
|
expected: list[tuple[tuple[str, ...], Any]] = [
|
|
(("$schema",), SCHEMA_DIALECT),
|
|
(("$id",), "https://crank.local/schemas/capability-inventory.schema.json"),
|
|
(("title",), "Crank Community Capability Inventory"),
|
|
(("type",), "object"),
|
|
(("additionalProperties",), False),
|
|
(("properties", "schema_version", "const"), 1),
|
|
(("properties", "product", "const"), "crank-community"),
|
|
(("properties", "flows", "type"), "array"),
|
|
(("properties", "flows", "maxItems"), MAX_FLOWS),
|
|
(("properties", "flows", "minItems"), 1),
|
|
(("properties", "flows", "items", "$ref"), "#/$defs/flow"),
|
|
(("$defs", "flow", "type"), "object"),
|
|
(("$defs", "flow", "additionalProperties"), False),
|
|
(("$defs", "flow", "properties", "id", "type"), "string"),
|
|
(("$defs", "flow", "properties", "id", "minLength"), 1),
|
|
(("$defs", "flow", "properties", "id", "maxLength"), MAX_ID_LENGTH),
|
|
(("$defs", "flow", "properties", "id", "pattern"), FLOW_ID.pattern),
|
|
(("$defs", "flow", "properties", "type", "enum"), list(FLOW_TYPES)),
|
|
(("$defs", "flow", "properties", "status", "enum"), list(FLOW_STATUSES)),
|
|
(
|
|
("$defs", "flow", "properties", "requirements", "maxItems"),
|
|
MAX_REQUIREMENTS_PER_FLOW,
|
|
),
|
|
(("$defs", "flow", "properties", "requirements", "type"), "array"),
|
|
(("$defs", "flow", "properties", "requirements", "minItems"), 1),
|
|
(("$defs", "flow", "properties", "requirements", "uniqueItems"), True),
|
|
(
|
|
("$defs", "flow", "properties", "requirements", "items", "type"),
|
|
"string",
|
|
),
|
|
(("$defs", "flow", "properties", "requirements", "items", "minLength"), 1),
|
|
(
|
|
("$defs", "flow", "properties", "requirements", "items", "maxLength"),
|
|
MAX_REQUIREMENT_LENGTH,
|
|
),
|
|
(
|
|
("$defs", "flow", "properties", "requirements", "items", "pattern"),
|
|
REQUIREMENT_ID.pattern,
|
|
),
|
|
(
|
|
("$defs", "flow", "properties", "capabilities", "maxItems"),
|
|
MAX_CAPABILITIES_PER_FLOW,
|
|
),
|
|
(("$defs", "flow", "properties", "capabilities", "type"), "array"),
|
|
(("$defs", "flow", "properties", "capabilities", "minItems"), 1),
|
|
(("$defs", "flow", "properties", "capabilities", "uniqueItems"), True),
|
|
(
|
|
("$defs", "flow", "properties", "capabilities", "items", "enum"),
|
|
list(ALLOWED_CAPABILITIES),
|
|
),
|
|
(("$defs", "flow", "properties", "owner", "type"), "string"),
|
|
(("$defs", "flow", "properties", "owner", "minLength"), 1),
|
|
(("$defs", "flow", "properties", "owner", "maxLength"), MAX_OWNER_LENGTH),
|
|
(("$defs", "flow", "properties", "owner", "pattern"), NON_WHITESPACE_PATTERN),
|
|
(("$defs", "flow", "properties", "user_outcome", "type"), "string"),
|
|
(("$defs", "flow", "properties", "user_outcome", "minLength"), 1),
|
|
(
|
|
("$defs", "flow", "properties", "user_outcome", "maxLength"),
|
|
MAX_OUTCOME_LENGTH,
|
|
),
|
|
(
|
|
("$defs", "flow", "properties", "user_outcome", "pattern"),
|
|
NON_WHITESPACE_PATTERN,
|
|
),
|
|
(("$defs", "flow", "properties", "evidence", "$ref"), "#/$defs/evidence"),
|
|
(("$defs", "flow", "properties", "notes", "type"), "string"),
|
|
(("$defs", "flow", "properties", "notes", "maxLength"), MAX_NOTES_LENGTH),
|
|
(("$defs", "evidence", "type"), "object"),
|
|
(("$defs", "evidence", "additionalProperties"), False),
|
|
(
|
|
("$defs", "evidence", "properties", "automated", "$ref"),
|
|
"#/$defs/evidencePaths",
|
|
),
|
|
(
|
|
("$defs", "evidence", "properties", "manual", "$ref"),
|
|
"#/$defs/evidencePaths",
|
|
),
|
|
(("$defs", "evidencePaths", "type"), "array"),
|
|
(("$defs", "evidencePaths", "minItems"), 1),
|
|
(("$defs", "evidencePaths", "maxItems"), MAX_EVIDENCE_LINKS_PER_KIND),
|
|
(("$defs", "evidencePaths", "uniqueItems"), True),
|
|
(("$defs", "evidencePaths", "items", "type"), "string"),
|
|
(("$defs", "evidencePaths", "items", "minLength"), 1),
|
|
(("$defs", "evidencePaths", "items", "maxLength"), MAX_EVIDENCE_PATH_LENGTH),
|
|
(("$defs", "evidencePaths", "items", "pattern"), EVIDENCE_PATH_PATTERN),
|
|
]
|
|
if not isinstance(schema, dict):
|
|
return [Diagnostic("INVALID_SCHEMA_CONTRACT", "/schema", "schema must be an object")]
|
|
diagnostics = []
|
|
for parts, allowed_keys in SCHEMA_ALLOWED_KEYS.items():
|
|
node = schema if not parts else nested(schema, *parts)
|
|
if not isinstance(node, dict):
|
|
continue
|
|
if any(key not in allowed_keys for key in node):
|
|
pointer = "/schema" + ("/" + "/".join(parts) if parts else "")
|
|
diagnostics.append(
|
|
Diagnostic(
|
|
"INVALID_SCHEMA_CONTRACT",
|
|
f"{pointer}/<unsupported>",
|
|
"schema contains an unsupported keyword",
|
|
)
|
|
)
|
|
for parts, value in expected:
|
|
if not contract_values_equal(nested(schema, *parts), value):
|
|
diagnostics.append(
|
|
Diagnostic(
|
|
"INVALID_SCHEMA_CONTRACT",
|
|
"/schema/" + "/".join(parts),
|
|
"schema contract differs from validator",
|
|
)
|
|
)
|
|
diagnostics.extend(
|
|
validate_required_keyword(
|
|
schema,
|
|
("required",),
|
|
frozenset({"schema_version", "product", "flows"}),
|
|
)
|
|
)
|
|
diagnostics.extend(
|
|
validate_required_keyword(
|
|
schema,
|
|
("$defs", "flow", "required"),
|
|
frozenset(FLOW_FIELDS - {"notes"}),
|
|
)
|
|
)
|
|
diagnostics.extend(
|
|
validate_required_keyword(
|
|
schema,
|
|
("$defs", "evidence", "required"),
|
|
EVIDENCE_FIELDS,
|
|
)
|
|
)
|
|
return diagnostics
|
|
|
|
|
|
def is_nonempty_string(value: Any, maximum: int) -> bool:
|
|
return isinstance(value, str) and bool(value.strip()) and len(value) <= maximum
|
|
|
|
|
|
def validate_string_list(
|
|
value: Any,
|
|
pointer: str,
|
|
maximum_items: int,
|
|
maximum_length: int,
|
|
pattern: re.Pattern[str] | None = None,
|
|
) -> list[Diagnostic]:
|
|
if not isinstance(value, list) or not value:
|
|
return [Diagnostic("MISSING_REQUIRED_FIELD", pointer, "non-empty list is required")]
|
|
diagnostics: list[Diagnostic] = []
|
|
if len(value) > maximum_items:
|
|
diagnostics.append(Diagnostic("LIMIT_EXCEEDED", pointer, "collection exceeds limit"))
|
|
seen: set[str] = set()
|
|
for index, item in enumerate(value[: maximum_items + 1]):
|
|
if not is_nonempty_string(item, maximum_length):
|
|
diagnostics.append(
|
|
Diagnostic(
|
|
"LIMIT_EXCEEDED",
|
|
f"{pointer}/{index}",
|
|
"string is empty, invalid, or exceeds limit",
|
|
)
|
|
)
|
|
continue
|
|
if pattern is not None and pattern.fullmatch(item) is None:
|
|
diagnostics.append(
|
|
Diagnostic("INVALID_FORMAT", f"{pointer}/{index}", "string format is invalid")
|
|
)
|
|
if item in seen:
|
|
diagnostics.append(
|
|
Diagnostic(
|
|
"DUPLICATE_LIST_ITEM",
|
|
f"{pointer}/{index}",
|
|
"collection item is duplicated",
|
|
)
|
|
)
|
|
seen.add(item)
|
|
return diagnostics
|
|
|
|
|
|
def reject_unknown_fields(
|
|
value: dict[str, Any], allowed: frozenset[str], pointer: str
|
|
) -> Iterator[Diagnostic]:
|
|
for field in value:
|
|
if field not in allowed:
|
|
yield Diagnostic("UNKNOWN_FIELD", f"{pointer}/<unknown>", "field is not allowed")
|
|
|
|
|
|
def validate_evidence(root: Path, value: Any, pointer: str) -> Iterator[Diagnostic]:
|
|
if not isinstance(value, dict):
|
|
yield Diagnostic("MISSING_REQUIRED_FIELD", pointer, "evidence object is required")
|
|
return
|
|
yield from reject_unknown_fields(value, EVIDENCE_FIELDS, pointer)
|
|
for kind in ("automated", "manual"):
|
|
paths = value.get(kind)
|
|
path_pointer = f"{pointer}/{kind}"
|
|
yield from validate_string_list(
|
|
paths,
|
|
path_pointer,
|
|
MAX_EVIDENCE_LINKS_PER_KIND,
|
|
MAX_EVIDENCE_PATH_LENGTH,
|
|
)
|
|
if not isinstance(paths, list):
|
|
continue
|
|
for index, raw_path in enumerate(paths[: MAX_EVIDENCE_LINKS_PER_KIND + 1]):
|
|
if not isinstance(raw_path, str) or resolve_regular_file(root, raw_path) is None:
|
|
yield Diagnostic(
|
|
"BROKEN_EVIDENCE_LINK",
|
|
f"{path_pointer}/{index}",
|
|
"evidence path is unavailable",
|
|
)
|
|
|
|
|
|
def validate_inventory(
|
|
root: Path,
|
|
inventory: Any,
|
|
required_frs: list[str],
|
|
) -> tuple[DiagnosticCollector, dict[str, int]]:
|
|
counts = {status: 0 for status in FLOW_STATUSES}
|
|
diagnostics = DiagnosticCollector()
|
|
if not isinstance(inventory, dict):
|
|
diagnostics.append(Diagnostic("MISSING_REQUIRED_FIELD", "/", "inventory must be an object"))
|
|
return diagnostics, counts
|
|
|
|
diagnostics.extend(reject_unknown_fields(inventory, TOP_LEVEL_FIELDS, ""))
|
|
schema_version = inventory.get("schema_version")
|
|
if type(schema_version) is not int or schema_version != 1:
|
|
diagnostics.append(
|
|
Diagnostic("MISSING_REQUIRED_FIELD", "/schema_version", "schema_version 1 is required")
|
|
)
|
|
if inventory.get("product") != "crank-community":
|
|
diagnostics.append(
|
|
Diagnostic("MISSING_REQUIRED_FIELD", "/product", "product identity is required")
|
|
)
|
|
flows = inventory.get("flows")
|
|
if not isinstance(flows, list) or not flows:
|
|
diagnostics.append(Diagnostic("MISSING_REQUIRED_FIELD", "/flows", "non-empty flows are required"))
|
|
return diagnostics, counts
|
|
if len(flows) > MAX_FLOWS:
|
|
diagnostics.append(Diagnostic("LIMIT_EXCEEDED", "/flows", "flow collection exceeds limit"))
|
|
|
|
seen_ids: set[str] = set()
|
|
observed_frs: set[str] = set()
|
|
required_fields = (
|
|
"id",
|
|
"type",
|
|
"requirements",
|
|
"user_outcome",
|
|
"owner",
|
|
"status",
|
|
"capabilities",
|
|
"evidence",
|
|
)
|
|
for index, flow in enumerate(flows[: MAX_FLOWS + 1]):
|
|
pointer = f"/flows/{index}"
|
|
if not isinstance(flow, dict):
|
|
diagnostics.append(Diagnostic("MISSING_REQUIRED_FIELD", pointer, "flow must be an object"))
|
|
continue
|
|
diagnostics.extend(reject_unknown_fields(flow, FLOW_FIELDS, pointer))
|
|
for field in required_fields:
|
|
if field not in flow:
|
|
code = "MISSING_OWNER" if field == "owner" else "MISSING_REQUIRED_FIELD"
|
|
diagnostics.append(Diagnostic(code, f"{pointer}/{field}", "field is required"))
|
|
|
|
flow_id = flow.get("id")
|
|
if not is_nonempty_string(flow_id, MAX_ID_LENGTH) or not FLOW_ID.fullmatch(flow_id):
|
|
diagnostics.append(Diagnostic("LIMIT_EXCEEDED", f"{pointer}/id", "flow id is invalid"))
|
|
elif flow_id in seen_ids:
|
|
diagnostics.append(Diagnostic("DUPLICATE_FLOW_ID", f"{pointer}/id", "flow id is duplicated"))
|
|
else:
|
|
seen_ids.add(flow_id)
|
|
|
|
flow_type = flow.get("type")
|
|
if flow_type not in FLOW_TYPES:
|
|
diagnostics.append(Diagnostic("UNKNOWN_FLOW_TYPE", f"{pointer}/type", "flow type is unknown"))
|
|
|
|
owner = flow.get("owner")
|
|
if "owner" in flow:
|
|
if not isinstance(owner, str) or not owner.strip():
|
|
diagnostics.append(
|
|
Diagnostic("MISSING_OWNER", f"{pointer}/owner", "owner is required")
|
|
)
|
|
elif len(owner) > MAX_OWNER_LENGTH:
|
|
diagnostics.append(
|
|
Diagnostic("LIMIT_EXCEEDED", f"{pointer}/owner", "owner exceeds limit")
|
|
)
|
|
|
|
outcome = flow.get("user_outcome")
|
|
if "user_outcome" in flow and not is_nonempty_string(outcome, MAX_OUTCOME_LENGTH):
|
|
diagnostics.append(
|
|
Diagnostic("LIMIT_EXCEEDED", f"{pointer}/user_outcome", "user outcome is invalid")
|
|
)
|
|
|
|
notes = flow.get("notes")
|
|
if notes is not None and (not isinstance(notes, str) or len(notes) > MAX_NOTES_LENGTH):
|
|
diagnostics.append(Diagnostic("LIMIT_EXCEEDED", f"{pointer}/notes", "notes exceed limit"))
|
|
|
|
status = flow.get("status")
|
|
if status not in FLOW_STATUSES:
|
|
diagnostics.append(Diagnostic("UNKNOWN_STATUS", f"{pointer}/status", "status is unknown"))
|
|
else:
|
|
counts[status] += 1
|
|
|
|
requirements = flow.get("requirements")
|
|
diagnostics.extend(
|
|
validate_string_list(
|
|
requirements,
|
|
f"{pointer}/requirements",
|
|
MAX_REQUIREMENTS_PER_FLOW,
|
|
MAX_REQUIREMENT_LENGTH,
|
|
REQUIREMENT_ID,
|
|
)
|
|
)
|
|
if isinstance(requirements, list):
|
|
for requirement in requirements[: MAX_REQUIREMENTS_PER_FLOW + 1]:
|
|
if isinstance(requirement, str) and REQUIREMENT_ID.fullmatch(requirement):
|
|
observed_frs.add(requirement)
|
|
|
|
capabilities = flow.get("capabilities")
|
|
diagnostics.extend(
|
|
validate_string_list(
|
|
capabilities,
|
|
f"{pointer}/capabilities",
|
|
MAX_CAPABILITIES_PER_FLOW,
|
|
MAX_CAPABILITY_LENGTH,
|
|
)
|
|
)
|
|
if isinstance(capabilities, list):
|
|
for cap_index, capability in enumerate(capabilities[: MAX_CAPABILITIES_PER_FLOW + 1]):
|
|
cap_pointer = f"{pointer}/capabilities/{cap_index}"
|
|
if capability in FORBIDDEN_CAPABILITIES:
|
|
diagnostics.append(
|
|
Diagnostic(
|
|
"FORBIDDEN_COMMUNITY_CAPABILITY",
|
|
cap_pointer,
|
|
f"forbidden capability: {capability}",
|
|
)
|
|
)
|
|
elif isinstance(capability, str) and capability not in ALLOWED_CAPABILITIES:
|
|
diagnostics.append(
|
|
Diagnostic("LIMIT_EXCEEDED", cap_pointer, "capability is not allowed")
|
|
)
|
|
|
|
diagnostics.extend(validate_evidence(root, flow.get("evidence"), f"{pointer}/evidence"))
|
|
|
|
for required_fr in sorted(set(required_frs)):
|
|
if not REQUIREMENT_ID.fullmatch(required_fr) or len(required_fr) > MAX_REQUIREMENT_LENGTH:
|
|
diagnostics.append(
|
|
Diagnostic("LIMIT_EXCEEDED", "/required-fr", "required FR identifier is invalid")
|
|
)
|
|
elif required_fr not in observed_frs:
|
|
diagnostics.append(
|
|
Diagnostic(
|
|
"MISSING_REQUIRED_FR",
|
|
"/requirements",
|
|
f"mandatory requirement is missing: {required_fr}",
|
|
)
|
|
)
|
|
return diagnostics, counts
|
|
|
|
|
|
def render_report(diagnostics: DiagnosticCollector) -> str:
|
|
selected = diagnostics.ordered()
|
|
lines = [f"error[{item.code}] {item.pointer}: {item.message}" for item in selected]
|
|
if diagnostics.omitted:
|
|
lines.append(f"error[TRUNCATED] /: diagnostics omitted={diagnostics.omitted}")
|
|
|
|
report = "\n".join(lines) + "\n"
|
|
encoded = report.encode("utf-8")
|
|
if len(encoded) <= MAX_REPORT_BYTES:
|
|
return report
|
|
|
|
marker = "error[TRUNCATED] /: report byte limit reached\n"
|
|
marker_bytes = marker.encode("utf-8")
|
|
budget = MAX_REPORT_BYTES - len(marker_bytes)
|
|
kept: list[str] = []
|
|
used = 0
|
|
for line in lines:
|
|
line_bytes = (line + "\n").encode("utf-8")
|
|
if used + len(line_bytes) > budget:
|
|
break
|
|
kept.append(line)
|
|
used += len(line_bytes)
|
|
return "\n".join(kept) + ("\n" if kept else "") + marker
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
root = args.root.resolve()
|
|
|
|
schema, schema_error = read_json_document(root, args.schema, "INVALID_SCHEMA")
|
|
inventory, inventory_error = read_json_document(root, args.inventory, "INVALID_JSON")
|
|
diagnostics = DiagnosticCollector()
|
|
diagnostics.extend(error for error in (schema_error, inventory_error) if error is not None)
|
|
if schema_error is None:
|
|
diagnostics.extend(validate_schema_contract(schema))
|
|
|
|
counts = {status: 0 for status in FLOW_STATUSES}
|
|
if inventory_error is None:
|
|
inventory_diagnostics, counts = validate_inventory(root, inventory, args.required_fr)
|
|
diagnostics.extend(inventory_diagnostics)
|
|
|
|
if diagnostics:
|
|
sys.stderr.write(render_report(diagnostics))
|
|
return 1
|
|
|
|
total = sum(counts.values())
|
|
print(
|
|
"Capability inventory validation passed "
|
|
f"total={total} implemented={counts['implemented']} "
|
|
f"planned={counts['planned']} gap={counts['gap']} "
|
|
f"blocked={counts['blocked']} pass={counts['implemented']}"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|