code-be-01andorch-01
9014a510a9
ci/woodpecker/push/publish Pipeline was successful
Co-authored-by: code-be-01 <[email protected]>
387 lines
18 KiB
Bash
Executable File
387 lines
18 KiB
Bash
Executable File
#!/usr/bin/env bash
|
||
# validate-repo-json.sh — declaration validator for T51 repo structure declarations.
|
||
#
|
||
# Spec of record: docs/plans/2026-08-23_repo-structure-declaration.md @ 1896adc1
|
||
# (R3). Implements the spec's validation surface: schema v1/v2 (§1.2), host:/
|
||
# path grammar with canonical segment normalization — empty/./.. rejected
|
||
# BEFORE resolution — and tilde rejection (§1.2a), mirror path component
|
||
# validation (§3.1), cross-field rules (§5.2), remote normalization (§5.3).
|
||
#
|
||
# PROVENANCE (T51 WP5c vendoring): ported verbatim from the mosaic-brain tree —
|
||
# tools/repo-structure-decl/validate-repo-json.sh @ brain main merge 515bcbab
|
||
# (PR 28, wave-1 R9 PASS, 101-arm suite green). This file is now the framework
|
||
# home per spec §5.1 ("shipped in the framework package"); the brain copy is the
|
||
# development origin. Re-sync rule: changes land here via reviewed PR and are
|
||
# back-ported to the brain tree (or the brain copy retires) — never fork silently.
|
||
# Validator version at port: 1.1.0+t51spec-r3+t51p2rw1 (R2-R8 rework included).
|
||
# No operator literal appears in this file; the host root is read from
|
||
# MOSAIC_HOST_ROOT configuration only.
|
||
#
|
||
# Unset-root semantics (spec §1.2a, fail-closed; T51P2R1 F1): v2 declarations
|
||
# always consume a path (canonical_clone is required), so in --mode managed an
|
||
# unset OR EMPTY MOSAIC_HOST_ROOT is a VALIDATION_ERROR for every v2 file —
|
||
# not only tool-managed. In --mode display it warns and omits root-dependent
|
||
# resolution; grammar checks still run.
|
||
#
|
||
# Error contract (T51P2R1 F3): every malformed input — bad UTF-8, non-RFC JSON
|
||
# constants (NaN/Infinity), wrong-typed enums, anything unexpected — yields
|
||
# exactly one stable VALIDATION_ERROR line and exit 1. No traceback ever
|
||
# escapes. Branch names are validated by delegating to `git check-ref-format
|
||
# --branch` (F2), translated into this contract.
|
||
#
|
||
# Usage:
|
||
# validate-repo-json.sh <repo.json> [--mode managed|display] [--require-v2]
|
||
# validate-repo-json.sh --mirror-path <host> <owner> <repo> # §3.1 component check
|
||
# validate-repo-json.sh --normalize-remote <url> # §5.3, prints normalized
|
||
# validate-repo-json.sh --version
|
||
#
|
||
# Output: OK (exit 0) | VALIDATION_ERROR <key>: <reason> (exit 1) | warnings on stderr.
|
||
set -euo pipefail
|
||
|
||
VERSION="1.1.0+t51spec-r3+t51p2rw1"
|
||
|
||
if [ "${1:-}" = "--version" ]; then echo "validate-repo-json $VERSION"; exit 0; fi
|
||
|
||
exec python3 - "$@" <<'PYEOF'
|
||
import json, os, re, subprocess, sys, urllib.parse
|
||
|
||
def err(key, reason):
|
||
print(f"VALIDATION_ERROR {key}: {reason}")
|
||
sys.exit(1)
|
||
|
||
def warn(msg):
|
||
print(f"warning: {msg}", file=sys.stderr)
|
||
|
||
def main():
|
||
ARGS = sys.argv[1:]
|
||
MODE = "managed"
|
||
REQUIRE_V2 = False
|
||
|
||
# ---- subcommands first (they take no file argument) ----
|
||
def check_mirror_components(host, owner, repo):
|
||
# fullmatch: '$' must bind at true end (F5 — a trailing newline must
|
||
# NOT pass); the charset excludes control bytes outright.
|
||
comp_re = re.compile(r"[a-z0-9][a-z0-9.-]*")
|
||
for label, value in (("host", host), ("owner", owner), ("repo", repo)):
|
||
if not isinstance(value, str) or not value or not comp_re.fullmatch(value):
|
||
err("mirror-component", f"{label} {value!r} fails §3.1 charset ^[a-z0-9][a-z0-9.-]*$ (fullmatch, no '/', no delimiter, no control bytes)")
|
||
return f"projects/{host}/{owner}/{repo}/repo.json"
|
||
|
||
def normalize_remote(url):
|
||
# R5-B1 part 1: reject raw control bytes BEFORE urlsplit — urlsplit
|
||
# silently strips TAB/LF/CR, so different input bytes would normalize
|
||
# to a different host. The raw bytes ARE the input; nothing may rewrite them.
|
||
for ch in url:
|
||
if ord(ch) < 0x20 or ord(ch) == 0x7F:
|
||
err("canonical_remote", "control byte in URL rejected before parsing (urlsplit would strip it and change the host)")
|
||
try:
|
||
p = urllib.parse.urlsplit(url)
|
||
except ValueError as e:
|
||
# py3.12 urlsplit itself validates bracketed hosts (ipaddress) and
|
||
# raises for garbage authorities — translate, never traceback.
|
||
err("canonical_remote", f"invalid URL authority: {e}")
|
||
if not p.scheme or not p.netloc:
|
||
err("canonical_remote", f"not a URL with scheme+host: {url!r}")
|
||
if p.username or p.password or "@" in (p.netloc or ""):
|
||
err("canonical_remote", "userinfo in URL is rejected (§5.3)")
|
||
scheme = p.scheme.lower()
|
||
# T51P2R4 B1: bracketing is detected from the RAW netloc ('[' prefix),
|
||
# not from a ':' in the parsed hostname — IPvFuture literals ([v1.fe80])
|
||
# contain no colon and must not bypass the raw-authority proof.
|
||
bracketed = p.netloc.startswith("[")
|
||
if bracketed:
|
||
# Prove the RAW authority is exactly '[host]' + optional ':port'
|
||
# (case-normalized); any text after ']' is hostile/truncated input,
|
||
# rejected — never silently rewritten.
|
||
import re as _re
|
||
import ipaddress as _ip
|
||
m = _re.fullmatch(r"\[([^\]]*)\](?::([0-9]+))?", p.netloc)
|
||
if not m:
|
||
err("canonical_remote", f"malformed bracketed authority {p.netloc!r}: text after ']' is rejected (no silent truncation)")
|
||
payload = m.group(1)
|
||
# R5-B1 part 2: the bracket payload must be a REAL RFC literal —
|
||
# an IPv6 address (ipaddress parse) or an IPvFuture literal
|
||
# ("v" + HEXDIG+ + "." + unreserved / sub-delims / ":" only).
|
||
# Anything else inside brackets is rejected, closing the payload
|
||
# grammar as a class.
|
||
if _re.fullmatch(r"v[0-9A-Fa-f]+\.[A-Za-z0-9._~!$&'()*+,;=:-]*", payload):
|
||
pass # IPvFuture (case-normalized below)
|
||
elif _re.fullmatch(r"[0-9A-Fa-f:.]+", payload):
|
||
# strict IPv6 lexical form (hex/colon/dot only — ipaddress alone
|
||
# would also accept scoped zone-ids like fe80::1%eth0, which are
|
||
# not valid URI host grammar unless %25-encoded)
|
||
try:
|
||
_ip.IPv6Address(payload)
|
||
except ValueError:
|
||
err("canonical_remote",
|
||
f"bracket payload {payload!r} is not a valid IPv6 address")
|
||
else:
|
||
err("canonical_remote",
|
||
f"bracket payload {payload!r} is neither a valid IPv6 address nor an IPvFuture literal (v+HEXDIG+.+unreserved/sub-delims/colon)")
|
||
host = f"[{payload.lower()}]" # brackets preserved (IPv6 + IPvFuture)
|
||
else:
|
||
# R6-B1: the non-bracketed branch — urlsplit PARSES but does not
|
||
# VALIDATE reg-name, and netloc-nonempty is not host presence.
|
||
# Split the raw authority ourselves (host[:port]) and validate the
|
||
# raw host against real grammar: unreserved / sub-delims / complete
|
||
# %HH octets (reg-name), or IPv4 dotted-quad (reg-name's numeric
|
||
# case). Port must be all digits. No branch trusts urlsplit alone.
|
||
import re as _re
|
||
raw_host, sep, raw_port = p.netloc.rpartition(":")
|
||
if sep and _re.fullmatch(r"[0-9]+", raw_port):
|
||
pass # host:port split
|
||
elif sep:
|
||
err("canonical_remote", f"invalid port {raw_port!r} in authority {p.netloc!r} (ASCII digits only)")
|
||
else:
|
||
raw_host, raw_port = p.netloc, None
|
||
if not raw_host:
|
||
err("canonical_remote", f"empty host in authority {p.netloc!r}")
|
||
# strict reg-name / IPv4 scan: unreserved + sub-delims, with '%'
|
||
# only inside complete %HH octets (IPv4 dotted-quad is a subset of
|
||
# this charset — digits and dots — so one scan covers both).
|
||
import re as _re
|
||
i = 0
|
||
ok_host = True
|
||
while i < len(raw_host):
|
||
c = raw_host[i]
|
||
if c == "%":
|
||
if i + 2 >= len(raw_host) or not _re.fullmatch(r"[0-9A-Fa-f]{2}", raw_host[i+1:i+3]):
|
||
ok_host = False; break
|
||
i += 3
|
||
elif c in "!$&'()*+,;=-._~" or ("a" <= c <= "z") or ("A" <= c <= "Z") or ("0" <= c <= "9"):
|
||
# R7-B1: EXPLICIT ASCII only — str.isalnum() is Unicode-aware
|
||
# and admits non-ASCII letters/digits (é, full-width ０). Policy
|
||
# is ASCII-only reg-name; punycode xn-- is the sanctioned
|
||
# Unicode spelling and remains legal under this charset.
|
||
i += 1
|
||
else:
|
||
ok_host = False; break
|
||
if not ok_host:
|
||
err("canonical_remote", f"host {raw_host!r} is not valid reg-name/IPv4 grammar (unreserved/sub-delims/complete %HH only)")
|
||
host = raw_host.lower()
|
||
port = p.port # None when absent; preserved whenever explicitly present (B2: incl. 0)
|
||
authority = host + (f":{port}" if port is not None else "")
|
||
path = p.path or "/"
|
||
# canonical trailing-slash + .git strip as ONE operation (F4): slash
|
||
# first, then .git, then any slash exposed by that strip.
|
||
path = path.rstrip("/")
|
||
if path.endswith(".git"):
|
||
path = path[:-4].rstrip("/")
|
||
return f"{scheme}://{authority}{path or ''}"
|
||
|
||
if "--mirror-path" in ARGS:
|
||
idx = ARGS.index("--mirror-path")
|
||
parts = ARGS[idx + 1:]
|
||
if len(parts) != 3:
|
||
err("usage", "--mirror-path takes <host> <owner> <repo>")
|
||
print(check_mirror_components(*parts))
|
||
sys.exit(0)
|
||
if "--normalize-remote" in ARGS:
|
||
idx = ARGS.index("--normalize-remote")
|
||
vals = ARGS[idx + 1:]
|
||
if len(vals) != 1:
|
||
err("usage", "--normalize-remote takes <url>")
|
||
print(normalize_remote(vals[0]))
|
||
sys.exit(0)
|
||
|
||
# ---- arg parsing ----
|
||
if not ARGS:
|
||
err("usage", "a repo.json path is required")
|
||
path = None
|
||
i = 0
|
||
while i < len(ARGS):
|
||
a = ARGS[i]
|
||
if a == "--mode":
|
||
i += 1
|
||
if i >= len(ARGS) or ARGS[i] not in ("managed", "display"):
|
||
err("usage", "--mode takes managed|display")
|
||
MODE = ARGS[i]
|
||
elif a == "--require-v2":
|
||
REQUIRE_V2 = True
|
||
elif a.startswith("--"):
|
||
err("usage", f"unknown option {a}")
|
||
else:
|
||
if path is not None:
|
||
err("usage", "multiple file arguments")
|
||
path = a
|
||
i += 1
|
||
if path is None:
|
||
err("usage", "a repo.json path is required")
|
||
|
||
# ---- load: strict UTF-8, strict RFC JSON (F3) ----
|
||
try:
|
||
with open(path, "rb") as fh:
|
||
raw_bytes = fh.read()
|
||
except OSError as e:
|
||
err("file", str(e))
|
||
try:
|
||
raw = raw_bytes.decode("utf-8")
|
||
except UnicodeDecodeError as e:
|
||
err("json", f"invalid UTF-8: {e}")
|
||
def _reject_constant(name):
|
||
raise ValueError(f"non-RFC JSON constant {name}")
|
||
try:
|
||
doc = json.loads(raw, parse_constant=_reject_constant)
|
||
except (json.JSONDecodeError, ValueError) as e:
|
||
err("json", f"malformed JSON: {e}")
|
||
if not isinstance(doc, dict):
|
||
err("json", "top level must be an object")
|
||
|
||
HOST_ROOT = os.environ.get("MOSAIC_HOST_ROOT", "")
|
||
|
||
V1_KEYS = {"integration_trunk", "release_branch"}
|
||
V2_REQUIRED = ["schema_version", "integration_trunk", "release_branch", "flow",
|
||
"canonical_remote", "canonical_clone"]
|
||
V2_OPTIONAL = {"worktree_root", "worktree_policy", "notes", "x_extensions"}
|
||
ENUM_FLOW = {"direct", "trunk-release"}
|
||
ENUM_POLICY = {"tool-managed", "orchestrator-precreated"}
|
||
|
||
def check_branch(key, value):
|
||
# Delegate the full git branch grammar to git itself (F2). B1: reject
|
||
# reflog shorthand BEFORE delegation — `git check-ref-format --branch
|
||
# '@{-n}'` expands from the CALLER repo's checkout history, making
|
||
# validation cwd-dependent; a persistent declaration must never bind
|
||
# to ambient reflog state.
|
||
if not isinstance(value, str) or not value:
|
||
err(key, "must be a non-empty string")
|
||
if "@{" in value:
|
||
err(key, f"{value!r} contains '@{{' reflog/namespace shorthand — declarations must be literal branch names (B1)")
|
||
if value.startswith("refs/heads/"): # check-ref-format --branch strips this; we do not allow it
|
||
err(key, "bare branch name expected, not a full ref")
|
||
try:
|
||
r = subprocess.run(["git", "check-ref-format", "--branch", value],
|
||
capture_output=True)
|
||
except OSError as e:
|
||
err(key, f"cannot invoke git check-ref-format: {e}")
|
||
if r.returncode != 0:
|
||
err(key, f"{value!r} is not a valid git branch name (git check-ref-format, §5.2)")
|
||
|
||
def check_host_path(key, value):
|
||
# §1.2a: host:/-anchored; canonical segment normalization; empty/./.. rejected
|
||
# BEFORE resolution; tilde rejected outright.
|
||
if not isinstance(value, str) or not value:
|
||
err(key, "must be a non-empty string")
|
||
if "~" in value:
|
||
err(key, "tilde-anchored path rejected (§1.2a: ~ binds to caller HOME)")
|
||
if not value.startswith("host:/"):
|
||
err(key, "must be host:/-anchored (§1.2a)")
|
||
rest = value[len("host:/"):]
|
||
if rest == "":
|
||
err(key, "no segments after host:/")
|
||
segments = rest.split("/")
|
||
for seg in segments:
|
||
if seg == "":
|
||
err(key, f"empty segment in {value!r} (canonical normalization, §1.2a)")
|
||
if seg in (".", ".."):
|
||
err(key, f"dot segment {seg!r} rejected before resolution (§1.2a)")
|
||
return segments
|
||
|
||
# ---- version ----
|
||
sv = doc.get("schema_version")
|
||
if "schema_version" in doc:
|
||
if not isinstance(sv, int) or isinstance(sv, bool):
|
||
err("schema_version", "must be an integer")
|
||
if sv not in (1, 2):
|
||
err("schema_version", f"unknown schema_version {sv} — treated as ABSENT per keep-list K3; update tooling")
|
||
version = sv
|
||
else:
|
||
version = 1
|
||
warn("schema_version absent → v1 compatibility mode (two keys only)")
|
||
|
||
if REQUIRE_V2 and version != 2:
|
||
err("schema_version", "CI authoring rule: new or edited declarations must declare schema_version 2")
|
||
|
||
# ---- v1 ----
|
||
if version == 1:
|
||
for k in V1_KEYS:
|
||
check_branch(k, doc.get(k))
|
||
extra = set(doc) - V1_KEYS
|
||
if extra:
|
||
err("x_extensions", f"unknown top-level keys in v1: {sorted(extra)}")
|
||
print("OK (v1)")
|
||
sys.exit(0)
|
||
|
||
# ---- v2 required ----
|
||
for k in V2_REQUIRED:
|
||
if k not in doc:
|
||
err(k, "required for v2 (§1.2)")
|
||
|
||
check_branch("integration_trunk", doc["integration_trunk"])
|
||
check_branch("release_branch", doc["release_branch"])
|
||
|
||
# type-check BEFORE membership (F3: list-typed enums must not traceback)
|
||
if not isinstance(doc["flow"], str) or doc["flow"] not in ENUM_FLOW:
|
||
err("flow", f"must be one of {sorted(ENUM_FLOW)} (§1.2, required — no defaulting, R7)")
|
||
|
||
unknown = set(doc) - set(V2_REQUIRED) - V2_OPTIONAL
|
||
if unknown:
|
||
err("x_extensions", f"unknown top-level keys {sorted(unknown)} — place extensions inside x_extensions")
|
||
|
||
if "worktree_policy" in doc and (not isinstance(doc["worktree_policy"], str)
|
||
or doc["worktree_policy"] not in ENUM_POLICY):
|
||
err("worktree_policy", f"must be one of {sorted(ENUM_POLICY)}")
|
||
if "notes" in doc and not isinstance(doc["notes"], str):
|
||
err("notes", "must be a string")
|
||
if "x_extensions" in doc and not isinstance(doc["x_extensions"], dict):
|
||
err("x_extensions", "must be an object")
|
||
for strkey in ("canonical_remote", "canonical_clone", "worktree_root"):
|
||
if strkey in doc and not isinstance(doc[strkey], str):
|
||
err(strkey, "must be a string")
|
||
|
||
# ---- remote (§5.3) ----
|
||
if not isinstance(doc["canonical_remote"], str):
|
||
err("canonical_remote", "must be a string")
|
||
else:
|
||
normalize_remote(doc["canonical_remote"])
|
||
|
||
# ---- paths (§1.2a) ----
|
||
canonical_segments = check_host_path("canonical_clone", doc["canonical_clone"])
|
||
wt_segments = None
|
||
if "worktree_root" in doc:
|
||
wt_segments = check_host_path("worktree_root", doc["worktree_root"])
|
||
|
||
# ---- cross-field (§5.2) ----
|
||
trunk, rel, flow = doc["integration_trunk"], doc["release_branch"], doc["flow"]
|
||
if flow == "direct" and trunk != rel:
|
||
err("flow", "direct requires integration_trunk == release_branch (§5.2)")
|
||
if flow == "trunk-release" and trunk == rel:
|
||
err("flow", "trunk-release requires integration_trunk != release_branch (§5.2)")
|
||
|
||
# ---- root gate (§1.2a fail-closed; T51P2R1 F1) ----
|
||
# Every v2 declaration consumes a path (canonical_clone is required), so
|
||
# managed mode cannot proceed without a provable host anchor. Display mode
|
||
# warns and omits root-dependent resolution only.
|
||
if not HOST_ROOT:
|
||
if MODE == "managed":
|
||
err("MOSAIC_HOST_ROOT",
|
||
"unset or empty — managed validation of a v2 declaration consumes paths "
|
||
"(canonical_clone required); fail closed (§1.2a, DR3 X2b)")
|
||
else:
|
||
warn("host root unset; root-dependent resolution omitted (display mode, §1.2a)")
|
||
|
||
if doc.get("worktree_policy") == "tool-managed":
|
||
if "worktree_root" not in doc:
|
||
err("worktree_policy", "tool-managed requires worktree_root (containment provable, §4.5)")
|
||
if HOST_ROOT:
|
||
root_real = os.path.realpath(HOST_ROOT)
|
||
# B3 fix: containment is tested on the LEXICAL normalized path, not
|
||
# on realpath of the joined result — a child symlink under the host
|
||
# root can no longer fake outside-ness. host:/ segments are always
|
||
# lexically under the root, so tool-managed fails universally until
|
||
# the anchor scheme grows a real outside-root form (J3 charter).
|
||
resolved = os.path.normpath(os.path.join(root_real, *wt_segments))
|
||
if resolved == root_real or resolved.startswith(root_real + os.sep):
|
||
err("worktree_policy",
|
||
f"tool-managed worktree_root resolves inside MOSAIC_HOST_ROOT (§4.5: outside-root requirement)")
|
||
# no-root case: managed mode already failed at the gate above; display warned
|
||
|
||
print("OK")
|
||
|
||
try:
|
||
main()
|
||
except SystemExit:
|
||
raise
|
||
except Exception as e: # F3: no traceback may ever escape the contract
|
||
err("internal", f"input rejected (unexpected condition: {type(e).__name__})")
|
||
PYEOF
|