376 lines
15 KiB
Python
376 lines
15 KiB
Python
"""Pure validation and deployment-truth evaluation for runtime contracts.
|
|
|
|
Runtime contracts describe stable structure. A release-owned deployment target
|
|
is supplied separately so that changing a release does not rewrite structural
|
|
configuration or turn a source branch into runtime evidence.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from datetime import datetime, timezone
|
|
import re
|
|
from pathlib import PurePosixPath
|
|
from typing import Any, Mapping, Sequence
|
|
|
|
|
|
CONTRACT_SCHEMA = "aios.runtime-contract.v1"
|
|
LIFECYCLES = frozenset({"active", "candidate", "legacy", "example"})
|
|
_DYNAMIC_FIELDS = frozenset(
|
|
{"desired", "desired_commit", "desired_release", "desired_target", "desired_targets"}
|
|
)
|
|
_SECRET_FIELDS = frozenset(
|
|
{"secret", "secret_bytes", "token", "password", "private_key", "credential_value", "value"}
|
|
)
|
|
_DIGEST = re.compile(r"^(?:sha256:)?[0-9a-fA-F]{64}$")
|
|
_SAFE_SYMBOLIC_REF = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*:[/A-Za-z0-9_.:@+-]+$")
|
|
|
|
|
|
def _error(code: str, path: str = "") -> ValueError:
|
|
return ValueError(f"{code}:{path}" if path else code)
|
|
|
|
|
|
def _nonempty_string(value: object, path: str) -> str:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise _error("field_invalid", path)
|
|
return value.strip()
|
|
|
|
|
|
def _positive_int(value: object, path: str) -> int:
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
|
|
raise _error("field_invalid", path)
|
|
return value
|
|
|
|
|
|
def _reject_dynamic_fields(value: object, path: str = "contract") -> None:
|
|
if isinstance(value, Mapping):
|
|
for key, child in value.items():
|
|
if not isinstance(key, str):
|
|
raise _error("field_name_invalid", path)
|
|
child_path = f"{path}.{key}"
|
|
if key.lower() in _DYNAMIC_FIELDS:
|
|
raise _error("dynamic_field_forbidden", child_path)
|
|
_reject_dynamic_fields(child, child_path)
|
|
elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
|
|
for index, child in enumerate(value):
|
|
_reject_dynamic_fields(child, f"{path}[{index}]")
|
|
|
|
|
|
def _safe_path_ref(value: object, path: str) -> str:
|
|
text = _nonempty_string(value, path)
|
|
if "\x00" in text or "\\" in text:
|
|
raise _error("unsafe_path_ref", path)
|
|
candidate = PurePosixPath(text)
|
|
if candidate.is_absolute() or any(part in {"", ".", ".."} for part in candidate.parts):
|
|
raise _error("unsafe_path_ref", path)
|
|
return text
|
|
|
|
|
|
def _symbolic_ref(value: object, path: str) -> str:
|
|
text = _nonempty_string(value, path)
|
|
if "\x00" in text or ".." in text or not _SAFE_SYMBOLIC_REF.fullmatch(text):
|
|
raise _error("resolver_ref_invalid", path)
|
|
return text
|
|
|
|
|
|
def _credential_ref(value: object, path: str) -> str:
|
|
text = _nonempty_string(value, path)
|
|
allowed = (
|
|
text.startswith("credential://")
|
|
or text.startswith("secret://")
|
|
or text.startswith("systemd-credential:")
|
|
)
|
|
if not allowed or "\x00" in text or ".." in text or text.startswith("/run/credentials/"):
|
|
raise _error("credential_ref_invalid", path)
|
|
return text
|
|
|
|
|
|
def _validate_binding_group(
|
|
group: object,
|
|
group_name: str,
|
|
component_id: str,
|
|
environment: str,
|
|
tenant_scope: str,
|
|
) -> dict[str, Any]:
|
|
if not isinstance(group, Mapping):
|
|
raise _error("binding_group_invalid", group_name)
|
|
normalized: dict[str, Any] = {}
|
|
for name, raw in group.items():
|
|
item_path = f"{group_name}.{name}"
|
|
_nonempty_string(name, f"{group_name}.name")
|
|
if isinstance(raw, str):
|
|
reference_field = (
|
|
"credential_ref"
|
|
if group_name == "credential_refs"
|
|
else "path_ref"
|
|
if group_name == "resource_refs" and ":" not in raw
|
|
else "resolver_ref"
|
|
)
|
|
item = {
|
|
"component_id": component_id,
|
|
"environment": environment,
|
|
reference_field: raw,
|
|
}
|
|
elif isinstance(raw, Mapping):
|
|
item = deepcopy(dict(raw))
|
|
else:
|
|
raise _error("binding_invalid", item_path)
|
|
if _nonempty_string(item.get("component_id"), f"{item_path}.component_id") != component_id:
|
|
raise _error("binding_component_mismatch", item_path)
|
|
if _nonempty_string(item.get("environment"), f"{item_path}.environment") != environment:
|
|
raise _error("binding_environment_mismatch", item_path)
|
|
if "tenant_scope" in item and (
|
|
_nonempty_string(item["tenant_scope"], f"{item_path}.tenant_scope") != tenant_scope
|
|
):
|
|
raise _error("binding_tenant_mismatch", item_path)
|
|
|
|
reference_fields = [key for key in ("resolver_ref", "path_ref", "credential_ref") if key in item]
|
|
if len(reference_fields) != 1:
|
|
raise _error("binding_reference_invalid", item_path)
|
|
reference_field = reference_fields[0]
|
|
reference_path = f"{item_path}.{reference_field}"
|
|
if group_name == "credential_refs" or reference_field == "credential_ref":
|
|
for key in item:
|
|
if key.lower() in _SECRET_FIELDS:
|
|
raise _error("secret_material_forbidden", f"{item_path}.{key}")
|
|
item[reference_field] = _credential_ref(item[reference_field], reference_path)
|
|
elif reference_field == "path_ref":
|
|
item[reference_field] = _safe_path_ref(item[reference_field], reference_path)
|
|
else:
|
|
item[reference_field] = _symbolic_ref(item[reference_field], reference_path)
|
|
normalized[str(name)] = item
|
|
return normalized
|
|
|
|
|
|
def validate_contract(document: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""Validate and copy a stable ``aios.runtime-contract.v1`` document.
|
|
|
|
The returned mapping owns its nested containers, so callers cannot mutate the
|
|
source after validation. Invalid contracts raise ``ValueError`` with a stable
|
|
machine-readable code prefix.
|
|
"""
|
|
|
|
if not isinstance(document, Mapping):
|
|
raise _error("contract_invalid")
|
|
_reject_dynamic_fields(document)
|
|
result = deepcopy(dict(document))
|
|
if result.get("contract_schema") != CONTRACT_SCHEMA:
|
|
raise _error("contract_schema_invalid", "contract_schema")
|
|
result["contract_revision"] = _positive_int(result.get("contract_revision"), "contract_revision")
|
|
component_id = _nonempty_string(result.get("component_id"), "component_id")
|
|
environment = _nonempty_string(result.get("environment"), "environment")
|
|
result["component_id"] = component_id
|
|
result["environment"] = environment
|
|
tenant_scope = _nonempty_string(result.get("tenant_scope"), "tenant_scope")
|
|
result["tenant_scope"] = tenant_scope
|
|
lifecycle = _nonempty_string(result.get("lifecycle"), "lifecycle").lower()
|
|
if lifecycle not in LIFECYCLES:
|
|
raise _error("lifecycle_invalid", "lifecycle")
|
|
result["lifecycle"] = lifecycle
|
|
|
|
if "resources" in result and "resource_refs" in result:
|
|
raise _error("duplicate_resource_group")
|
|
if "resources" in result:
|
|
result["resource_refs"] = result.pop("resources")
|
|
for group_name in ("entrypoints", "bindings", "resource_refs", "credential_refs"):
|
|
result[group_name] = _validate_binding_group(
|
|
result.get(group_name), group_name, component_id, environment, tenant_scope
|
|
)
|
|
|
|
operations = result.get("operations")
|
|
if (
|
|
not isinstance(operations, Sequence)
|
|
or isinstance(operations, (str, bytes, bytearray))
|
|
or not operations
|
|
):
|
|
raise _error("operations_invalid", "operations")
|
|
normalized_operations = [
|
|
_nonempty_string(operation, f"operations[{index}]")
|
|
for index, operation in enumerate(operations)
|
|
]
|
|
if len(set(normalized_operations)) != len(normalized_operations):
|
|
raise _error("duplicate_operation", "operations")
|
|
result["operations"] = normalized_operations
|
|
return result
|
|
|
|
|
|
def _unknown(*reasons: str) -> dict[str, Any]:
|
|
return {
|
|
"status": "UNKNOWN",
|
|
"state": "UNKNOWN",
|
|
"verification": "UNKNOWN",
|
|
"converged": False,
|
|
"reasons": list(dict.fromkeys(reasons)),
|
|
}
|
|
|
|
|
|
def _drift(reasons: list[str]) -> dict[str, Any]:
|
|
return {
|
|
"status": "DRIFT",
|
|
"state": "DRIFT",
|
|
"verification": "FAILED",
|
|
"converged": False,
|
|
"reasons": list(dict.fromkeys(reasons)),
|
|
}
|
|
|
|
|
|
def _parse_time(value: object) -> datetime | None:
|
|
if isinstance(value, datetime):
|
|
parsed = value
|
|
elif isinstance(value, str):
|
|
try:
|
|
parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
else:
|
|
return None
|
|
if parsed.tzinfo is None or parsed.utcoffset() is None:
|
|
return None
|
|
return parsed.astimezone(timezone.utc)
|
|
|
|
|
|
def _digest_value(document: Mapping[str, Any], *names: str) -> object:
|
|
for name in names:
|
|
if name in document:
|
|
return document[name]
|
|
return None
|
|
|
|
|
|
def _valid_digest(value: object) -> bool:
|
|
return isinstance(value, str) and bool(_DIGEST.fullmatch(value))
|
|
|
|
|
|
def evaluate_deployment(
|
|
contract: Mapping[str, Any],
|
|
target: Mapping[str, Any] | None,
|
|
observation: Mapping[str, Any] | None,
|
|
provenance: Mapping[str, Any] | None,
|
|
now: datetime | str,
|
|
) -> dict[str, Any]:
|
|
"""Evaluate one target using fresh runtime and provenance evidence.
|
|
|
|
Missing or malformed evidence is ``UNKNOWN``. ``DRIFT`` is reserved for a
|
|
complete evidence set that proves an incompatible target or runtime fact.
|
|
Repository HEAD is intentionally not an input to this decision.
|
|
"""
|
|
|
|
stable = validate_contract(contract)
|
|
missing: list[str] = []
|
|
if target is None:
|
|
missing.append("target_missing")
|
|
elif not isinstance(target, Mapping):
|
|
missing.append("target_invalid")
|
|
if observation is None:
|
|
missing.append("observation_missing")
|
|
elif not isinstance(observation, Mapping):
|
|
missing.append("observation_invalid")
|
|
if provenance is None:
|
|
missing.append("provenance_missing")
|
|
elif not isinstance(provenance, Mapping):
|
|
missing.append("provenance_invalid")
|
|
current_time = _parse_time(now)
|
|
if current_time is None:
|
|
missing.append("now_invalid")
|
|
if missing:
|
|
return _unknown(*missing)
|
|
assert target is not None and observation is not None and provenance is not None
|
|
assert current_time is not None
|
|
|
|
target_errors: list[str] = []
|
|
for field in ("component_id", "environment", "release_id", "deployment_id"):
|
|
if not isinstance(target.get(field), str) or not str(target[field]).strip():
|
|
target_errors.append(f"target_{field}_invalid")
|
|
if "tenant_scope" in target and (
|
|
not isinstance(target["tenant_scope"], str) or not target["tenant_scope"].strip()
|
|
):
|
|
target_errors.append("target_tenant_scope_invalid")
|
|
target_artifact = target.get("artifact_digest")
|
|
if not _valid_digest(target_artifact):
|
|
target_errors.append("target_artifact_digest_invalid")
|
|
if isinstance(target.get("revision"), bool) or not isinstance(target.get("revision"), int) or target.get("revision", 0) < 1:
|
|
target_errors.append("target_revision_invalid")
|
|
|
|
observed_at = _parse_time(observation.get("observed_at"))
|
|
max_age = observation.get("max_age_seconds", observation.get("max_age"))
|
|
observed_artifact = _digest_value(
|
|
observation, "actual_artifact_digest", "actual_artifact", "artifact_digest"
|
|
)
|
|
observation_errors: list[str] = []
|
|
if observed_at is None:
|
|
observation_errors.append("observation_observed_at_invalid")
|
|
if isinstance(max_age, bool) or not isinstance(max_age, (int, float)) or max_age <= 0:
|
|
observation_errors.append("observation_max_age_invalid")
|
|
if not isinstance(observation.get("source"), str) or not observation.get("source", "").strip():
|
|
observation_errors.append("observation_source_invalid")
|
|
if not _valid_digest(observed_artifact):
|
|
observation_errors.append("observation_actual_artifact_invalid")
|
|
if not isinstance(observation.get("healthy"), bool):
|
|
observation_errors.append("observation_healthy_invalid")
|
|
for field in ("component_id", "environment", "tenant_scope", "deployment_id"):
|
|
if field in observation and (
|
|
not isinstance(observation[field], str) or not observation[field].strip()
|
|
):
|
|
observation_errors.append(f"observation_{field}_invalid")
|
|
if "revision" in observation and (
|
|
isinstance(observation["revision"], bool)
|
|
or not isinstance(observation["revision"], int)
|
|
or observation["revision"] < 1
|
|
):
|
|
observation_errors.append("observation_revision_invalid")
|
|
|
|
artifact_provenance = _digest_value(provenance, "artifact_digest", "artifact_hash")
|
|
source_provenance = _digest_value(provenance, "source_digest", "source_hash")
|
|
builder_provenance = _digest_value(provenance, "builder_digest", "builder_hash")
|
|
provenance_errors: list[str] = []
|
|
if not _valid_digest(artifact_provenance):
|
|
provenance_errors.append("provenance_artifact_digest_invalid")
|
|
if not _valid_digest(source_provenance):
|
|
provenance_errors.append("provenance_source_digest_invalid")
|
|
if not _valid_digest(builder_provenance):
|
|
provenance_errors.append("provenance_builder_digest_invalid")
|
|
|
|
evidence_errors = target_errors + observation_errors + provenance_errors
|
|
if evidence_errors:
|
|
return _unknown(*evidence_errors)
|
|
assert observed_at is not None and isinstance(max_age, (int, float))
|
|
age = (current_time - observed_at).total_seconds()
|
|
if age < 0:
|
|
return _unknown("observation_from_future")
|
|
if age > max_age:
|
|
return _unknown("observation_stale")
|
|
|
|
drift: list[str] = []
|
|
if target["component_id"] != stable["component_id"]:
|
|
drift.append("target_component_id_mismatch")
|
|
if target["environment"] != stable["environment"]:
|
|
drift.append("target_environment_mismatch")
|
|
if "tenant_scope" in target and target["tenant_scope"] != stable["tenant_scope"]:
|
|
drift.append("target_tenant_scope_mismatch")
|
|
if "tenant_scope" in observation and observation["tenant_scope"] != stable["tenant_scope"]:
|
|
drift.append("observation_tenant_scope_mismatch")
|
|
for field in ("component_id", "environment", "deployment_id", "revision"):
|
|
if field in observation and observation[field] != target[field]:
|
|
drift.append(f"observation_{field}_mismatch")
|
|
if observed_artifact != target_artifact:
|
|
drift.append("artifact_mismatch")
|
|
if artifact_provenance != target_artifact:
|
|
drift.append("provenance_artifact_mismatch")
|
|
if observation["healthy"] is False:
|
|
drift.append("observation_unhealthy")
|
|
if drift:
|
|
return _drift(drift)
|
|
return {
|
|
"status": "VERIFIED",
|
|
"state": "CONVERGED",
|
|
"verification": "VERIFIED",
|
|
"converged": True,
|
|
"reasons": [],
|
|
"component_id": target["component_id"],
|
|
"environment": target["environment"],
|
|
"release_id": target["release_id"],
|
|
"deployment_id": target["deployment_id"],
|
|
"artifact_digest": target_artifact,
|
|
"revision": target["revision"],
|
|
"observed_at": observation["observed_at"],
|
|
"observation_source": observation["source"],
|
|
}
|