toki
8f00606c03
제품 결과와 harness·process·artifact 실패가 하나의 성공 값으로 덮이지 않도록 durable evidence와 모든 소비자 계약을 함께 마이그레이션한다.
1104 lines
42 KiB
Python
1104 lines
42 KiB
Python
"""Fail-closed S12 web evidence and immutable artifact validation.
|
|
|
|
The record written by this module is deliberately self-contained: it binds the
|
|
fixture and generated workspace files, browser observations, both manifest
|
|
viewports, screenshot bytes, ordered gates, and the attempt measurement. The
|
|
loader revalidates the referenced bytes instead of trusting a summary flag.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import stat
|
|
from dataclasses import dataclass
|
|
from pathlib import Path, PurePosixPath
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
from scripts.agent_benchmark.browser_cdp import (
|
|
BrowserError,
|
|
BrowserRenderer,
|
|
RenderObservation,
|
|
)
|
|
from scripts.agent_benchmark.lifecycle import publish_bytes_no_replace
|
|
from scripts.agent_benchmark.manifest import VIEWPORT_ID_RE
|
|
from scripts.agent_benchmark.measurement import (
|
|
AttemptMeasurement,
|
|
MEASUREMENT_FILENAME,
|
|
)
|
|
|
|
WEB_VALIDATION_FILENAME = "web-validation.json"
|
|
WEB_VALIDATION_VERSION = 2
|
|
WEB_STATUSES = ("passed", "failed", "blocked", "not_run")
|
|
WEB_GATES = (
|
|
"generated_files",
|
|
"static_safety",
|
|
"images",
|
|
"network",
|
|
"console",
|
|
"responsive",
|
|
"accessibility",
|
|
)
|
|
GENERATED_FILES = ("index.html", "script.js", "styles.css")
|
|
MAX_WORKSPACE_FILES = 256
|
|
MAX_WORKSPACE_BYTES = 32 * 1024 * 1024
|
|
MAX_EVIDENCE_BYTES = 64 * 1024 * 1024
|
|
DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
|
|
TOKEN_RE = re.compile(r"^[a-z0-9][a-z0-9_]{0,95}$")
|
|
UNRECOVERABLE_BROWSER_REASONS = {
|
|
"browser_process_cleanup_failed",
|
|
"output_unavailable",
|
|
"screenshot_cleanup_failed",
|
|
"screenshot_collision",
|
|
}
|
|
|
|
|
|
class WebValidationError(Exception):
|
|
"""The S12 record or one of its referenced artifacts is untrusted."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WebValidation:
|
|
status: str
|
|
record: dict[str, Any]
|
|
|
|
|
|
def _digest(data: bytes) -> str:
|
|
return "sha256:" + hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def _bytes(value: dict[str, Any]) -> bytes:
|
|
return (
|
|
json.dumps(
|
|
value,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
ensure_ascii=True,
|
|
).encode("ascii")
|
|
+ b"\n"
|
|
)
|
|
|
|
|
|
def _regular(path: Path) -> bytes:
|
|
flags = os.O_RDONLY | os.O_CLOEXEC | os.O_NONBLOCK
|
|
if hasattr(os, "O_NOFOLLOW"):
|
|
flags |= os.O_NOFOLLOW
|
|
try:
|
|
fd = os.open(path, flags)
|
|
except OSError as exc:
|
|
raise WebValidationError("web validation is unavailable") from exc
|
|
try:
|
|
info = os.fstat(fd)
|
|
if not stat.S_ISREG(info.st_mode) or info.st_size > MAX_EVIDENCE_BYTES:
|
|
raise WebValidationError("web validation must be a bounded regular file")
|
|
data = bytearray()
|
|
while len(data) < info.st_size:
|
|
chunk = os.read(fd, min(1024 * 1024, info.st_size - len(data)))
|
|
if not chunk:
|
|
raise WebValidationError("web validation changed while reading")
|
|
data.extend(chunk)
|
|
if os.read(fd, 1):
|
|
raise WebValidationError("web validation changed while reading")
|
|
return bytes(data)
|
|
except OSError as exc:
|
|
raise WebValidationError("web validation is unavailable") from exc
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def _canonical_path(value: Any) -> str:
|
|
if not isinstance(value, str) or not value or "\\" in value or ":" in value:
|
|
raise WebValidationError("web validation path is invalid")
|
|
path = PurePosixPath(value)
|
|
if path.is_absolute() or str(path) != value or any(
|
|
part in ("", ".", "..") for part in path.parts
|
|
):
|
|
raise WebValidationError("web validation path is invalid")
|
|
return value
|
|
|
|
|
|
def _safe_read(root: Path, relative: str) -> bytes:
|
|
"""Read a bounded regular file without following any path component."""
|
|
relative = _canonical_path(relative)
|
|
directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC
|
|
file_flags = os.O_RDONLY | os.O_CLOEXEC | os.O_NONBLOCK
|
|
if hasattr(os, "O_NOFOLLOW"):
|
|
directory_flags |= os.O_NOFOLLOW
|
|
file_flags |= os.O_NOFOLLOW
|
|
descriptors: list[int] = []
|
|
try:
|
|
current = os.open(root, directory_flags)
|
|
descriptors.append(current)
|
|
parts = PurePosixPath(relative).parts
|
|
for component in parts[:-1]:
|
|
current = os.open(component, directory_flags, dir_fd=current)
|
|
descriptors.append(current)
|
|
fd = os.open(parts[-1], file_flags, dir_fd=current)
|
|
descriptors.append(fd)
|
|
info = os.fstat(fd)
|
|
if not stat.S_ISREG(info.st_mode) or info.st_size > MAX_WORKSPACE_BYTES:
|
|
raise OSError("not a bounded regular file")
|
|
data = bytearray()
|
|
while len(data) < info.st_size:
|
|
chunk = os.read(fd, min(1024 * 1024, info.st_size - len(data)))
|
|
if not chunk:
|
|
raise OSError("short read")
|
|
data.extend(chunk)
|
|
if os.read(fd, 1):
|
|
raise OSError("file grew during read")
|
|
return bytes(data)
|
|
finally:
|
|
for descriptor in reversed(descriptors):
|
|
try:
|
|
os.close(descriptor)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _observed_file(root: Path, relative: str) -> tuple[str, str, int]:
|
|
try:
|
|
data = _safe_read(root, relative)
|
|
except FileNotFoundError:
|
|
return "missing", "", 0
|
|
except (OSError, ValueError, WebValidationError):
|
|
return (
|
|
("non_regular", "", 0)
|
|
if os.path.lexists(root / relative)
|
|
else ("missing", "", 0)
|
|
)
|
|
return "regular", _digest(data), len(data)
|
|
|
|
|
|
def _file_fact(
|
|
root: Path,
|
|
relative: str,
|
|
kind: str,
|
|
*,
|
|
expected: bytes | None = None,
|
|
) -> dict[str, Any]:
|
|
state, digest, size = _observed_file(root, relative)
|
|
expected_digest = "" if expected is None else _digest(expected)
|
|
if state == "regular" and expected is not None and digest != expected_digest:
|
|
state = "mismatch"
|
|
return {
|
|
"path": relative,
|
|
"kind": kind,
|
|
"state": state,
|
|
"digest": digest,
|
|
"size": size,
|
|
"expected_digest": expected_digest,
|
|
}
|
|
|
|
|
|
def _workspace_entries(root: Path) -> tuple[str, ...]:
|
|
entries: list[str] = []
|
|
for current, directories, files in os.walk(root, followlinks=False):
|
|
current_path = Path(current)
|
|
for name in sorted([*directories, *files]):
|
|
path = current_path / name
|
|
relative = path.relative_to(root).as_posix()
|
|
entries.append(relative)
|
|
if len(entries) > MAX_WORKSPACE_FILES:
|
|
raise WebValidationError("workspace evidence exceeds file budget")
|
|
return tuple(sorted(entries))
|
|
|
|
|
|
def _expected_directories(paths: set[str]) -> set[str]:
|
|
directories: set[str] = set()
|
|
for raw in paths:
|
|
parent = PurePosixPath(raw).parent
|
|
while str(parent) != ".":
|
|
directories.add(str(parent))
|
|
parent = parent.parent
|
|
return directories
|
|
|
|
|
|
def _workspace_snapshot(workspace: Path, manifest) -> dict[str, Any]:
|
|
inputs = [
|
|
_file_fact(
|
|
workspace,
|
|
asset.workspace_path,
|
|
"fixture",
|
|
expected=asset.content,
|
|
)
|
|
for asset in sorted(manifest.fixture.assets, key=lambda item: item.workspace_path)
|
|
]
|
|
generated = [
|
|
_file_fact(workspace, name, "generated") for name in GENERATED_FILES
|
|
]
|
|
expected_paths = {item["path"] for item in [*inputs, *generated]}
|
|
allowed_entries = expected_paths | _expected_directories(expected_paths)
|
|
extras = [
|
|
path for path in _workspace_entries(workspace) if path not in allowed_entries
|
|
]
|
|
return {"inputs": inputs, "generated": generated, "extra_paths": extras}
|
|
|
|
|
|
def _gate(
|
|
ident: str,
|
|
passed: bool,
|
|
reason: str,
|
|
source: str,
|
|
evidence: list[str],
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"id": ident,
|
|
"passed": passed,
|
|
"reason": "" if passed else reason,
|
|
"source": source,
|
|
"evidence": evidence,
|
|
}
|
|
|
|
|
|
def _generated_gate(snapshot: dict[str, Any]) -> dict[str, Any]:
|
|
facts = [*snapshot["inputs"], *snapshot["generated"]]
|
|
passed = all(item["state"] == "regular" for item in facts) and not snapshot[
|
|
"extra_paths"
|
|
]
|
|
if snapshot["extra_paths"]:
|
|
reason = "unexpected_content"
|
|
else:
|
|
failed = next((item for item in facts if item["state"] != "regular"), None)
|
|
reason = "" if failed is None else f"{failed['kind']}_{failed['state']}"
|
|
evidence = [
|
|
f"{item['kind']}:{item['path']}:{item['state']}" for item in facts
|
|
] + [f"extra:{item}" for item in snapshot["extra_paths"]]
|
|
return _gate("generated_files", passed, reason, "workspace", evidence)
|
|
|
|
|
|
def _static_gate(workspace: Path, generated_gate: dict[str, Any], manifest) -> dict[str, Any]:
|
|
if not generated_gate["passed"]:
|
|
return _gate(
|
|
"static_safety",
|
|
False,
|
|
"workspace_not_closed",
|
|
"source_scan",
|
|
["generated_files=false"],
|
|
)
|
|
try:
|
|
text = b"\n".join(_safe_read(workspace, name) for name in GENERATED_FILES).decode(
|
|
"utf-8", "strict"
|
|
)
|
|
except (OSError, UnicodeDecodeError, WebValidationError):
|
|
return _gate(
|
|
"static_safety", False, "source_unavailable", "source_scan", []
|
|
)
|
|
reason = ""
|
|
if re.search(r"(?:@import|\bimport\s*(?:\(|[\"']))", text, re.I):
|
|
reason = "external_or_module_reference"
|
|
elif re.search(r"\b(?:react|vue|angular|bootstrap|tailwind)\b", text, re.I):
|
|
reason = "framework_reference"
|
|
else:
|
|
declared = {asset.workspace_path for asset in manifest.fixture.assets}
|
|
for reference in re.findall(r"(?:src|href)\s*=\s*[\"']([^\"']+)", text, re.I):
|
|
path = reference.split("?", 1)[0].split("#", 1)[0]
|
|
if not path or path in GENERATED_FILES or reference.startswith("#"):
|
|
continue
|
|
if re.match(r"^(?:https?:|//|data:)", reference, re.I) or path not in declared:
|
|
reason = "undeclared_reference"
|
|
break
|
|
return _gate(
|
|
"static_safety",
|
|
not reason,
|
|
reason,
|
|
"source_scan",
|
|
[f"generated={','.join(GENERATED_FILES)}"],
|
|
)
|
|
|
|
|
|
def _viewport_record(view) -> dict[str, Any]:
|
|
return {
|
|
"id": view.id,
|
|
"width": view.width,
|
|
"height": view.height,
|
|
"screenshot": {
|
|
"file": view.screenshot,
|
|
"digest": view.screenshot_digest,
|
|
"size": view.screenshot_size,
|
|
},
|
|
"images": [dict(item) for item in view.image_facts],
|
|
"layout": dict(view.layout),
|
|
"accessibility": dict(view.accessibility),
|
|
}
|
|
|
|
|
|
def _runtime_gates(manifest, render: RenderObservation) -> dict[str, dict[str, Any]]:
|
|
viewports = tuple(render.viewports)
|
|
expected_assets = {
|
|
asset.workspace_path
|
|
for asset in manifest.fixture.assets
|
|
if asset.workspace_path.startswith("assets/")
|
|
}
|
|
image_failures: list[str] = []
|
|
responsive_failures: list[str] = []
|
|
accessibility_failures: list[str] = []
|
|
for view in viewports:
|
|
images = {
|
|
item.get("src"): item
|
|
for item in view.image_facts
|
|
if isinstance(item, dict) and isinstance(item.get("src"), str)
|
|
}
|
|
for path in sorted(expected_assets):
|
|
item = images.get(path)
|
|
if not item or not (
|
|
item.get("complete") is True
|
|
and isinstance(item.get("natural_width"), int)
|
|
and item["natural_width"] > 0
|
|
and isinstance(item.get("natural_height"), int)
|
|
and item["natural_height"] > 0
|
|
and item.get("visible") is True
|
|
and isinstance(item.get("alt"), str)
|
|
and bool(item["alt"].strip())
|
|
):
|
|
image_failures.append(f"{view.id}:{path}")
|
|
layout = view.layout
|
|
if (
|
|
layout.get("scroll_width") != layout.get("client_width")
|
|
or layout.get("clipped") != 0
|
|
or layout.get("overlaps") != 0
|
|
):
|
|
responsive_failures.append(view.id)
|
|
accessibility = view.accessibility
|
|
controls = accessibility.get("controls")
|
|
ax = accessibility.get("ax")
|
|
if not (
|
|
accessibility.get("h1_count") == 1
|
|
and accessibility.get("heading_progression") is True
|
|
and isinstance(accessibility.get("main_count"), int)
|
|
and accessibility["main_count"] >= 1
|
|
and isinstance(accessibility.get("landmarks"), int)
|
|
and accessibility["landmarks"] >= 1
|
|
and isinstance(controls, list)
|
|
and controls
|
|
and all(
|
|
item.get("name") is True
|
|
and item.get("focused") is True
|
|
and item.get("focus_visible") is True
|
|
and isinstance(item.get("tab_index"), int)
|
|
and item["tab_index"] >= 0
|
|
and isinstance(item.get("contrast"), (int, float))
|
|
and not isinstance(item.get("contrast"), bool)
|
|
and math.isfinite(item["contrast"])
|
|
and item["contrast"] >= 4.5
|
|
for item in controls
|
|
if isinstance(item, dict)
|
|
)
|
|
and isinstance(ax, dict)
|
|
and isinstance(ax.get("non_ignored"), int)
|
|
and ax["non_ignored"] > 0
|
|
and isinstance(ax.get("named"), int)
|
|
and ax["named"] > 0
|
|
):
|
|
accessibility_failures.append(view.id)
|
|
expected_viewports = [item.id for item in manifest.viewports]
|
|
observed_viewports = [item.id for item in viewports]
|
|
if observed_viewports != expected_viewports:
|
|
responsive_failures.append("viewport_set")
|
|
|
|
request_failures = [
|
|
item
|
|
for item in render.requests
|
|
if not item.get("allowed") or int(item.get("status", 0)) >= 400
|
|
]
|
|
console_failures = [
|
|
item
|
|
for item in render.console
|
|
if item.get("kind") == "exception"
|
|
or str(item.get("level", "")).lower()
|
|
in {"error", "warning", "warn", "assert"}
|
|
]
|
|
return {
|
|
"images": _gate(
|
|
"images",
|
|
not image_failures and bool(expected_assets),
|
|
"image_evidence_failed" if image_failures else "image_fixture_missing",
|
|
"browser_dom",
|
|
image_failures or [f"asset:{path}" for path in sorted(expected_assets)],
|
|
),
|
|
"network": _gate(
|
|
"network",
|
|
not request_failures,
|
|
"request_failed",
|
|
"browser_fetch",
|
|
[f"requests={len(render.requests)}"],
|
|
),
|
|
"console": _gate(
|
|
"console",
|
|
not console_failures,
|
|
"console_error",
|
|
"browser_console",
|
|
[f"events={len(render.console)}"],
|
|
),
|
|
"responsive": _gate(
|
|
"responsive",
|
|
not responsive_failures,
|
|
"layout_failed",
|
|
"browser_layout",
|
|
responsive_failures or [f"viewport:{item}" for item in observed_viewports],
|
|
),
|
|
"accessibility": _gate(
|
|
"accessibility",
|
|
not accessibility_failures,
|
|
"accessibility_failed",
|
|
"browser_dom_ax",
|
|
accessibility_failures
|
|
or [f"viewport:{item}" for item in observed_viewports],
|
|
),
|
|
}
|
|
|
|
|
|
def _not_observed_gates(reason: str, source: str) -> dict[str, dict[str, Any]]:
|
|
return {
|
|
ident: _gate(ident, False, reason, source, [f"reason={reason}"])
|
|
for ident in WEB_GATES
|
|
}
|
|
|
|
|
|
def _reason_token(value: Any, fallback: str) -> str:
|
|
text = str(value or "").strip().lower().replace("-", "_").replace(" ", "_")
|
|
return text if TOKEN_RE.fullmatch(text) else fallback
|
|
|
|
|
|
def build_web_validation(
|
|
manifest,
|
|
attempt,
|
|
measurement: AttemptMeasurement,
|
|
render: RenderObservation | None,
|
|
*,
|
|
blocked: str = "",
|
|
) -> WebValidation:
|
|
workspace = Path(
|
|
attempt.workspace_dir if hasattr(attempt, "workspace_dir") else attempt
|
|
)
|
|
attempt_root = Path(
|
|
attempt.attempt_root
|
|
if hasattr(attempt, "attempt_root")
|
|
else workspace.parent
|
|
)
|
|
try:
|
|
workspace_mode = os.lstat(workspace).st_mode
|
|
attempt_mode = os.lstat(attempt_root).st_mode
|
|
except OSError as exc:
|
|
raise WebValidationError("web validation workspace is unavailable") from exc
|
|
if not stat.S_ISDIR(workspace_mode) or not stat.S_ISDIR(attempt_mode):
|
|
raise WebValidationError("web validation workspace is invalid")
|
|
snapshot = _workspace_snapshot(workspace, manifest)
|
|
generated_gate = _generated_gate(snapshot)
|
|
static_gate = _static_gate(workspace, generated_gate, manifest)
|
|
browser = {"status": "not_observed", "product": "", "origin": ""}
|
|
requests: list[dict[str, Any]] = []
|
|
console: list[dict[str, Any]] = []
|
|
viewports: list[dict[str, Any]] = []
|
|
reason = ""
|
|
if blocked:
|
|
status = "blocked"
|
|
reason = _reason_token(blocked, "browser_failure")
|
|
gates = _not_observed_gates(reason, "browser")
|
|
gates["generated_files"] = generated_gate
|
|
gates["static_safety"] = static_gate
|
|
elif render is None:
|
|
status = "failed"
|
|
reason = generated_gate["reason"] or "render_not_run"
|
|
gates = _not_observed_gates("render_not_run", "pipeline")
|
|
gates["generated_files"] = generated_gate
|
|
gates["static_safety"] = static_gate
|
|
else:
|
|
browser = {
|
|
"status": "observed",
|
|
"product": render.browser,
|
|
"origin": render.origin,
|
|
}
|
|
requests = [dict(item) for item in render.requests]
|
|
console = [dict(item) for item in render.console]
|
|
viewports = [_viewport_record(item) for item in render.viewports]
|
|
gates = _runtime_gates(manifest, render)
|
|
gates["generated_files"] = generated_gate
|
|
gates["static_safety"] = static_gate
|
|
passed = all(gates[ident]["passed"] for ident in WEB_GATES)
|
|
status = "passed" if passed else "failed"
|
|
if not passed:
|
|
reason = next(gates[ident]["reason"] for ident in WEB_GATES if not gates[ident]["passed"])
|
|
|
|
screenshots = [
|
|
{"id": item["id"], **item["screenshot"]} for item in viewports
|
|
]
|
|
measurement_bytes = _regular(attempt_root / MEASUREMENT_FILENAME)
|
|
record = {
|
|
"record": "web-validation",
|
|
"web_validation_version": WEB_VALIDATION_VERSION,
|
|
"status": status,
|
|
"reason": reason,
|
|
"attempt": {
|
|
"run_id": measurement.run_id,
|
|
"cell_id": measurement.cell_id,
|
|
"repetition": measurement.repetition,
|
|
"attempt": measurement.attempt,
|
|
},
|
|
"manifest_digest": manifest.digest,
|
|
"fixture_checksum": manifest.fixture.checksum,
|
|
"measurement_digest": _digest(measurement_bytes),
|
|
"workspace": snapshot,
|
|
"browser": browser,
|
|
"requests": requests,
|
|
"console": console,
|
|
"viewports": viewports,
|
|
"screenshots": screenshots,
|
|
"gates": [gates[ident] for ident in WEB_GATES],
|
|
}
|
|
web = WebValidation(status, record)
|
|
_validate_record(record, attempt_root, manifest=manifest)
|
|
return web
|
|
|
|
|
|
def validate_web_attempt(
|
|
manifest,
|
|
attempt_root: str | Path,
|
|
prepared,
|
|
measurement: AttemptMeasurement,
|
|
result=None,
|
|
*,
|
|
browser_binary: str = "chromium",
|
|
) -> WebValidation:
|
|
workspace = Path(prepared.workspace_dir)
|
|
render = None
|
|
blocked = ""
|
|
generated_ready = all(
|
|
_observed_file(workspace, name)[0] == "regular" for name in GENERATED_FILES
|
|
)
|
|
if generated_ready:
|
|
try:
|
|
render = BrowserRenderer(browser_binary).render(
|
|
workspace_root=workspace,
|
|
output_root=attempt_root,
|
|
viewports=manifest.viewports,
|
|
timeout_seconds=manifest.timeout.run_seconds,
|
|
)
|
|
except BrowserError as exc:
|
|
blocked = _reason_token(exc, "browser_failure")
|
|
if blocked in UNRECOVERABLE_BROWSER_REASONS:
|
|
raise WebValidationError(
|
|
"browser evidence cleanup or collision is invalid"
|
|
) from exc
|
|
except (FileNotFoundError, OSError):
|
|
blocked = "browser_start_failed"
|
|
return build_web_validation(
|
|
manifest, prepared, measurement, render, blocked=blocked
|
|
)
|
|
|
|
|
|
def _require_fields(value: Any, fields: set[str], message: str) -> dict[str, Any]:
|
|
if not isinstance(value, dict) or set(value) != fields:
|
|
raise WebValidationError(message)
|
|
return value
|
|
|
|
|
|
def _is_int(value: Any, *, minimum: int = 0) -> bool:
|
|
return not isinstance(value, bool) and isinstance(value, int) and value >= minimum
|
|
|
|
|
|
def _is_number(value: Any, *, minimum: float | None = None) -> bool:
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
return False
|
|
if not math.isfinite(value):
|
|
return False
|
|
return minimum is None or value >= minimum
|
|
|
|
|
|
def _validate_fact(item: Any, expected_kind: str) -> dict[str, Any]:
|
|
fact = _require_fields(
|
|
item,
|
|
{"path", "kind", "state", "digest", "size", "expected_digest"},
|
|
"web validation workspace fact is invalid",
|
|
)
|
|
_canonical_path(fact["path"])
|
|
if fact["kind"] != expected_kind or fact["state"] not in {
|
|
"regular",
|
|
"missing",
|
|
"non_regular",
|
|
"mismatch",
|
|
}:
|
|
raise WebValidationError("web validation workspace fact is invalid")
|
|
if not _is_int(fact["size"]):
|
|
raise WebValidationError("web validation workspace fact is invalid")
|
|
if fact["state"] in {"regular", "mismatch"}:
|
|
if not isinstance(fact["digest"], str) or not DIGEST_RE.fullmatch(fact["digest"]):
|
|
raise WebValidationError("web validation workspace fact is invalid")
|
|
elif fact["digest"] != "" or fact["size"] != 0:
|
|
raise WebValidationError("web validation workspace fact is invalid")
|
|
if expected_kind == "fixture":
|
|
if not isinstance(fact["expected_digest"], str) or not DIGEST_RE.fullmatch(
|
|
fact["expected_digest"]
|
|
):
|
|
raise WebValidationError("web validation workspace fact is invalid")
|
|
if fact["state"] == "regular" and fact["digest"] != fact["expected_digest"]:
|
|
raise WebValidationError("web validation workspace fact is inconsistent")
|
|
if fact["state"] == "mismatch" and fact["digest"] == fact["expected_digest"]:
|
|
raise WebValidationError("web validation workspace fact is inconsistent")
|
|
elif fact["expected_digest"] != "":
|
|
raise WebValidationError("web validation workspace fact is invalid")
|
|
return fact
|
|
|
|
|
|
def _validate_rect(value: Any) -> None:
|
|
rect = _require_fields(
|
|
value,
|
|
{"x", "y", "width", "height", "right", "bottom"},
|
|
"web validation image rectangle is invalid",
|
|
)
|
|
if not all(_is_number(item) for item in rect.values()) or not all(
|
|
_is_number(rect[key], minimum=0) for key in ("width", "height")
|
|
):
|
|
raise WebValidationError("web validation image rectangle is invalid")
|
|
if not math.isclose(rect["right"], rect["x"] + rect["width"], abs_tol=0.01) or not math.isclose(
|
|
rect["bottom"], rect["y"] + rect["height"], abs_tol=0.01
|
|
):
|
|
raise WebValidationError("web validation image rectangle is inconsistent")
|
|
|
|
|
|
def _validate_viewport(value: Any) -> dict[str, Any]:
|
|
view = _require_fields(
|
|
value,
|
|
{"id", "width", "height", "screenshot", "images", "layout", "accessibility"},
|
|
"web validation viewport is invalid",
|
|
)
|
|
if not isinstance(view["id"], str) or not VIEWPORT_ID_RE.fullmatch(view["id"]):
|
|
raise WebValidationError("web validation viewport is invalid")
|
|
if not _is_int(view["width"], minimum=1) or not _is_int(view["height"], minimum=1):
|
|
raise WebValidationError("web validation viewport is invalid")
|
|
screenshot = _require_fields(
|
|
view["screenshot"],
|
|
{"file", "digest", "size"},
|
|
"web validation screenshot is invalid",
|
|
)
|
|
_canonical_path(screenshot["file"])
|
|
if (
|
|
"/" in screenshot["file"]
|
|
or screenshot["file"] != f"screenshot-{view['id']}.png"
|
|
or not DIGEST_RE.fullmatch(str(screenshot["digest"]))
|
|
or not _is_int(screenshot["size"], minimum=1)
|
|
):
|
|
raise WebValidationError("web validation screenshot is invalid")
|
|
if not isinstance(view["images"], list):
|
|
raise WebValidationError("web validation image evidence is invalid")
|
|
for raw in view["images"]:
|
|
image = _require_fields(
|
|
raw,
|
|
{
|
|
"src",
|
|
"alt",
|
|
"complete",
|
|
"natural_width",
|
|
"natural_height",
|
|
"visible",
|
|
"rect",
|
|
},
|
|
"web validation image evidence is invalid",
|
|
)
|
|
if (
|
|
not isinstance(image["src"], str)
|
|
or not isinstance(image["alt"], str)
|
|
or not isinstance(image["complete"], bool)
|
|
or not _is_int(image["natural_width"])
|
|
or not _is_int(image["natural_height"])
|
|
or not isinstance(image["visible"], bool)
|
|
):
|
|
raise WebValidationError("web validation image evidence is invalid")
|
|
_validate_rect(image["rect"])
|
|
layout = _require_fields(
|
|
view["layout"],
|
|
{"scroll_width", "client_width", "clipped", "overlaps"},
|
|
"web validation layout evidence is invalid",
|
|
)
|
|
if not all(_is_int(item) for item in layout.values()):
|
|
raise WebValidationError("web validation layout evidence is invalid")
|
|
accessibility = _require_fields(
|
|
view["accessibility"],
|
|
{
|
|
"h1_count",
|
|
"headings",
|
|
"heading_progression",
|
|
"main_count",
|
|
"landmarks",
|
|
"controls",
|
|
"ax",
|
|
},
|
|
"web validation accessibility evidence is invalid",
|
|
)
|
|
if (
|
|
not all(
|
|
_is_int(accessibility[key])
|
|
for key in ("h1_count", "main_count", "landmarks")
|
|
)
|
|
or not isinstance(accessibility["heading_progression"], bool)
|
|
or not isinstance(accessibility["headings"], list)
|
|
or not all(_is_int(item, minimum=1) and item <= 6 for item in accessibility["headings"])
|
|
or not isinstance(accessibility["controls"], list)
|
|
):
|
|
raise WebValidationError("web validation accessibility evidence is invalid")
|
|
expected_progression = all(
|
|
accessibility["headings"][index]
|
|
<= accessibility["headings"][index - 1] + 1
|
|
for index in range(1, len(accessibility["headings"]))
|
|
)
|
|
if (
|
|
accessibility["heading_progression"] != expected_progression
|
|
or accessibility["h1_count"]
|
|
!= sum(1 for item in accessibility["headings"] if item == 1)
|
|
):
|
|
raise WebValidationError(
|
|
"web validation accessibility evidence is inconsistent"
|
|
)
|
|
for raw in accessibility["controls"]:
|
|
control = _require_fields(
|
|
raw,
|
|
{"name", "tab_index", "focused", "focus_visible", "contrast"},
|
|
"web validation control evidence is invalid",
|
|
)
|
|
if (
|
|
not isinstance(control["name"], bool)
|
|
or not _is_int(control["tab_index"])
|
|
or not isinstance(control["focused"], bool)
|
|
or not isinstance(control["focus_visible"], bool)
|
|
or not _is_number(control["contrast"], minimum=0)
|
|
):
|
|
raise WebValidationError("web validation control evidence is invalid")
|
|
ax = _require_fields(
|
|
accessibility["ax"],
|
|
{"nodes", "non_ignored", "named"},
|
|
"web validation accessibility tree is invalid",
|
|
)
|
|
if not all(_is_int(item) for item in ax.values()):
|
|
raise WebValidationError("web validation accessibility tree is invalid")
|
|
if not 0 <= ax["named"] <= ax["non_ignored"] <= ax["nodes"]:
|
|
raise WebValidationError("web validation accessibility tree is inconsistent")
|
|
if (
|
|
view["layout"]["client_width"] > view["width"]
|
|
or view["layout"]["scroll_width"] < view["layout"]["client_width"]
|
|
):
|
|
raise WebValidationError("web validation layout evidence is inconsistent")
|
|
return view
|
|
|
|
|
|
def _validate_record(record: Any, attempt_root: Path, *, manifest=None) -> None:
|
|
fields = {
|
|
"record",
|
|
"web_validation_version",
|
|
"status",
|
|
"reason",
|
|
"attempt",
|
|
"manifest_digest",
|
|
"fixture_checksum",
|
|
"measurement_digest",
|
|
"workspace",
|
|
"browser",
|
|
"requests",
|
|
"console",
|
|
"viewports",
|
|
"screenshots",
|
|
"gates",
|
|
}
|
|
record = _require_fields(record, fields, "web validation schema is invalid")
|
|
if (
|
|
record["record"] != "web-validation"
|
|
or record["web_validation_version"] != WEB_VALIDATION_VERSION
|
|
or record["status"] not in WEB_STATUSES
|
|
or not isinstance(record["reason"], str)
|
|
or (record["reason"] and not TOKEN_RE.fullmatch(record["reason"]))
|
|
):
|
|
raise WebValidationError("web validation schema is invalid")
|
|
identity = _require_fields(
|
|
record["attempt"],
|
|
{"run_id", "cell_id", "repetition", "attempt"},
|
|
"web validation identity is invalid",
|
|
)
|
|
if (
|
|
not all(isinstance(identity[key], str) and identity[key] for key in ("run_id", "cell_id"))
|
|
or not _is_int(identity["repetition"], minimum=1)
|
|
or not _is_int(identity["attempt"], minimum=1)
|
|
):
|
|
raise WebValidationError("web validation identity is invalid")
|
|
for field in ("manifest_digest", "fixture_checksum", "measurement_digest"):
|
|
if not isinstance(record[field], str) or not DIGEST_RE.fullmatch(record[field]):
|
|
raise WebValidationError("web validation digest is invalid")
|
|
|
|
workspace = _require_fields(
|
|
record["workspace"],
|
|
{"inputs", "generated", "extra_paths"},
|
|
"web validation workspace evidence is invalid",
|
|
)
|
|
if not all(isinstance(workspace[field], list) for field in workspace):
|
|
raise WebValidationError("web validation workspace evidence is invalid")
|
|
inputs = [_validate_fact(item, "fixture") for item in workspace["inputs"]]
|
|
generated = [_validate_fact(item, "generated") for item in workspace["generated"]]
|
|
if (
|
|
[item["path"] for item in inputs] != sorted(item["path"] for item in inputs)
|
|
or [item["path"] for item in generated] != list(GENERATED_FILES)
|
|
or len({item["path"] for item in [*inputs, *generated]}) != len(inputs) + len(generated)
|
|
or workspace["extra_paths"] != sorted(workspace["extra_paths"])
|
|
):
|
|
raise WebValidationError("web validation workspace evidence is invalid")
|
|
for path in workspace["extra_paths"]:
|
|
_canonical_path(path)
|
|
|
|
workspace_root = attempt_root / "workspace"
|
|
for fact in [*inputs, *generated]:
|
|
state, digest, size = _observed_file(workspace_root, fact["path"])
|
|
expected_state = "regular" if fact["state"] == "mismatch" else fact["state"]
|
|
if state != expected_state or digest != fact["digest"] or size != fact["size"]:
|
|
raise WebValidationError("web validation workspace artifact changed")
|
|
expected_paths = {item["path"] for item in [*inputs, *generated]}
|
|
allowed_entries = expected_paths | _expected_directories(expected_paths)
|
|
extras = [
|
|
path for path in _workspace_entries(workspace_root) if path not in allowed_entries
|
|
]
|
|
if extras != workspace["extra_paths"]:
|
|
raise WebValidationError("web validation workspace artifact changed")
|
|
|
|
browser = _require_fields(
|
|
record["browser"],
|
|
{"status", "product", "origin"},
|
|
"web validation browser evidence is invalid",
|
|
)
|
|
if browser["status"] not in {"observed", "not_observed"} or not all(
|
|
isinstance(browser[key], str) for key in ("product", "origin")
|
|
):
|
|
raise WebValidationError("web validation browser evidence is invalid")
|
|
if browser["status"] == "observed":
|
|
if not browser["product"] or not re.fullmatch(r"http://127\.0\.0\.1:[0-9]+", browser["origin"]):
|
|
raise WebValidationError("web validation browser evidence is invalid")
|
|
elif browser["product"] or browser["origin"]:
|
|
raise WebValidationError("web validation browser evidence is invalid")
|
|
|
|
if not isinstance(record["requests"], list):
|
|
raise WebValidationError("web validation request evidence is invalid")
|
|
for raw in record["requests"]:
|
|
if not isinstance(raw, dict) or raw.get("kind") not in {"local", "external"}:
|
|
raise WebValidationError("web validation request evidence is invalid")
|
|
if raw["kind"] == "local":
|
|
request = _require_fields(
|
|
raw,
|
|
{"kind", "path", "allowed", "status"},
|
|
"web validation request evidence is invalid",
|
|
)
|
|
if not isinstance(request["path"], str) or not request["path"].startswith("/"):
|
|
raise WebValidationError("web validation request evidence is invalid")
|
|
else:
|
|
request = _require_fields(
|
|
raw,
|
|
{"kind", "url_digest", "allowed", "status"},
|
|
"web validation request evidence is invalid",
|
|
)
|
|
if not isinstance(request["url_digest"], str) or not DIGEST_RE.fullmatch(request["url_digest"]):
|
|
raise WebValidationError("web validation request evidence is invalid")
|
|
if not isinstance(request["allowed"], bool) or not _is_int(request["status"]):
|
|
raise WebValidationError("web validation request evidence is invalid")
|
|
if request["kind"] == "external" and (
|
|
request["allowed"] or request["status"] != 0
|
|
):
|
|
raise WebValidationError("web validation request evidence is inconsistent")
|
|
if request["kind"] == "local" and request["allowed"] != (
|
|
request["status"] < 400
|
|
):
|
|
raise WebValidationError("web validation request evidence is inconsistent")
|
|
if not isinstance(record["console"], list):
|
|
raise WebValidationError("web validation console evidence is invalid")
|
|
for raw in record["console"]:
|
|
item = _require_fields(
|
|
raw,
|
|
{"kind", "level"},
|
|
"web validation console evidence is invalid",
|
|
)
|
|
if item["kind"] not in {"console", "exception", "log"} or not isinstance(item["level"], str):
|
|
raise WebValidationError("web validation console evidence is invalid")
|
|
|
|
if not isinstance(record["viewports"], list):
|
|
raise WebValidationError("web validation viewport evidence is invalid")
|
|
viewports = [_validate_viewport(item) for item in record["viewports"]]
|
|
if len({item["id"] for item in viewports}) != len(viewports):
|
|
raise WebValidationError("web validation viewport evidence is invalid")
|
|
if not isinstance(record["screenshots"], list):
|
|
raise WebValidationError("web validation screenshot evidence is invalid")
|
|
expected_screenshots = [
|
|
{"id": item["id"], **item["screenshot"]} for item in viewports
|
|
]
|
|
if record["screenshots"] != expected_screenshots:
|
|
raise WebValidationError("web validation screenshot evidence is inconsistent")
|
|
for screenshot in record["screenshots"]:
|
|
data = _regular(attempt_root / screenshot["file"])
|
|
if (
|
|
not data.startswith(b"\x89PNG\r\n\x1a\n")
|
|
or len(data) != screenshot["size"]
|
|
or _digest(data) != screenshot["digest"]
|
|
):
|
|
raise WebValidationError("web validation screenshot artifact is invalid")
|
|
referenced_screenshots = {item["file"] for item in record["screenshots"]}
|
|
present_screenshots = {
|
|
item.name
|
|
for item in attempt_root.iterdir()
|
|
if item.name.startswith("screenshot-") and item.name.endswith(".png")
|
|
}
|
|
if present_screenshots != referenced_screenshots:
|
|
raise WebValidationError("web validation screenshot set is invalid")
|
|
|
|
if not isinstance(record["gates"], list) or [
|
|
item.get("id") if isinstance(item, dict) else None for item in record["gates"]
|
|
] != list(WEB_GATES):
|
|
raise WebValidationError("web validation gates are invalid")
|
|
for raw in record["gates"]:
|
|
gate = _require_fields(
|
|
raw,
|
|
{"id", "passed", "reason", "source", "evidence"},
|
|
"web validation gates are invalid",
|
|
)
|
|
if (
|
|
not isinstance(gate["passed"], bool)
|
|
or not isinstance(gate["reason"], str)
|
|
or not isinstance(gate["source"], str)
|
|
or not gate["source"]
|
|
or not isinstance(gate["evidence"], list)
|
|
or not all(isinstance(item, str) and item for item in gate["evidence"])
|
|
or (gate["passed"] and gate["reason"])
|
|
or (not gate["passed"] and not TOKEN_RE.fullmatch(gate["reason"]))
|
|
):
|
|
raise WebValidationError("web validation gates are invalid")
|
|
|
|
gate_manifest = manifest
|
|
if gate_manifest is None:
|
|
gate_manifest = SimpleNamespace(
|
|
fixture=SimpleNamespace(
|
|
assets=tuple(
|
|
SimpleNamespace(workspace_path=item["path"])
|
|
for item in inputs
|
|
)
|
|
),
|
|
viewports=tuple(
|
|
SimpleNamespace(
|
|
id=item["id"], width=item["width"], height=item["height"]
|
|
)
|
|
for item in viewports
|
|
),
|
|
)
|
|
generated_gate = _generated_gate(workspace)
|
|
static_gate = _static_gate(workspace_root, generated_gate, gate_manifest)
|
|
if browser["status"] == "observed":
|
|
projected_render = SimpleNamespace(
|
|
requests=tuple(record["requests"]),
|
|
console=tuple(record["console"]),
|
|
viewports=tuple(
|
|
SimpleNamespace(
|
|
id=item["id"],
|
|
width=item["width"],
|
|
height=item["height"],
|
|
image_facts=tuple(item["images"]),
|
|
layout=item["layout"],
|
|
accessibility=item["accessibility"],
|
|
)
|
|
for item in viewports
|
|
),
|
|
)
|
|
expected_gates = _runtime_gates(gate_manifest, projected_render)
|
|
expected_gates["generated_files"] = generated_gate
|
|
expected_gates["static_safety"] = static_gate
|
|
elif record["status"] == "blocked":
|
|
expected_gates = _not_observed_gates(record["reason"], "browser")
|
|
expected_gates["generated_files"] = generated_gate
|
|
expected_gates["static_safety"] = static_gate
|
|
elif record["status"] == "not_run":
|
|
expected_gates = _not_observed_gates(record["reason"], "lifecycle")
|
|
else:
|
|
expected_gates = _not_observed_gates("render_not_run", "pipeline")
|
|
expected_gates["generated_files"] = generated_gate
|
|
expected_gates["static_safety"] = static_gate
|
|
if record["gates"] != [expected_gates[ident] for ident in WEB_GATES]:
|
|
raise WebValidationError("web validation gate evidence is inconsistent")
|
|
|
|
all_passed = all(item["passed"] for item in record["gates"])
|
|
observed = browser["status"] == "observed"
|
|
if record["status"] == "passed":
|
|
if not all_passed or not observed or not viewports or record["reason"]:
|
|
raise WebValidationError("web validation status is inconsistent")
|
|
elif record["status"] == "failed":
|
|
expected_reason = next(
|
|
item["reason"] for item in record["gates"] if not item["passed"]
|
|
)
|
|
if all_passed or record["reason"] != expected_reason:
|
|
raise WebValidationError("web validation status is inconsistent")
|
|
elif record["status"] == "blocked":
|
|
if observed or viewports or record["screenshots"] or not record["reason"]:
|
|
raise WebValidationError("web validation status is inconsistent")
|
|
elif record["status"] == "not_run":
|
|
if observed or viewports or record["screenshots"] or not record["reason"].startswith("lifecycle_"):
|
|
raise WebValidationError("web validation status is inconsistent")
|
|
|
|
measurement = _regular(attempt_root / MEASUREMENT_FILENAME)
|
|
if _digest(measurement) != record["measurement_digest"]:
|
|
raise WebValidationError("web validation measurement artifact changed")
|
|
if manifest is not None:
|
|
_validate_manifest_binding(record, manifest)
|
|
|
|
|
|
def _validate_manifest_binding(record: dict[str, Any], manifest) -> None:
|
|
if (
|
|
record["manifest_digest"] != manifest.digest
|
|
or record["fixture_checksum"] != manifest.fixture.checksum
|
|
):
|
|
raise WebValidationError("web validation manifest binding is invalid")
|
|
inputs = record["workspace"]["inputs"]
|
|
expected_inputs = [
|
|
(asset.workspace_path, _digest(asset.content))
|
|
for asset in sorted(manifest.fixture.assets, key=lambda item: item.workspace_path)
|
|
]
|
|
if [(item["path"], item["expected_digest"]) for item in inputs] != expected_inputs:
|
|
raise WebValidationError("web validation fixture binding is invalid")
|
|
if record["browser"]["status"] == "observed":
|
|
expected_viewports = [
|
|
(item.id, item.width, item.height) for item in manifest.viewports
|
|
]
|
|
observed_viewports = [
|
|
(item["id"], item["width"], item["height"])
|
|
for item in record["viewports"]
|
|
]
|
|
if observed_viewports != expected_viewports:
|
|
raise WebValidationError("web validation viewport binding is invalid")
|
|
if record["status"] == "passed" and any(
|
|
item["state"] != "regular" for item in record["workspace"]["inputs"]
|
|
):
|
|
raise WebValidationError("web validation fixture status is inconsistent")
|
|
|
|
|
|
def validate_web_validation_manifest(record: WebValidation, manifest) -> None:
|
|
"""Rebind an already loaded record to the immutable run manifest."""
|
|
_validate_manifest_binding(record.record, manifest)
|
|
|
|
|
|
def publish_web_validation(attempt_root: str | Path, record: WebValidation) -> Path:
|
|
root = Path(attempt_root)
|
|
if not isinstance(record, WebValidation) or record.status != record.record.get("status"):
|
|
raise WebValidationError("web validation object is invalid")
|
|
_validate_record(record.record, root)
|
|
path = root / WEB_VALIDATION_FILENAME
|
|
try:
|
|
publish_bytes_no_replace(path, _bytes(record.record))
|
|
except Exception as exc:
|
|
raise WebValidationError(
|
|
"web validation publication refused an existing target"
|
|
) from exc
|
|
return path
|
|
|
|
|
|
def load_web_validation(
|
|
attempt_root: str | Path, *, manifest=None
|
|
) -> WebValidation:
|
|
root = Path(attempt_root)
|
|
raw = _regular(root / WEB_VALIDATION_FILENAME)
|
|
try:
|
|
record = json.loads(raw.decode("ascii"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise WebValidationError("web validation is invalid") from exc
|
|
if not isinstance(record, dict) or _bytes(record) != raw:
|
|
raise WebValidationError("web validation is not canonical")
|
|
_validate_record(record, root, manifest=manifest)
|
|
return WebValidation(record["status"], record)
|