Build immutable policy publication artifact
This commit is contained in:
parent
cac0301866
commit
e8035f3887
22 changed files with 2375 additions and 452 deletions
300
tools/build_site.py
Normal file
300
tools/build_site.py
Normal file
|
|
@ -0,0 +1,300 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Build the policy site atomically from an explicit source manifest."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import datetime as dt
|
||||
import hashlib
|
||||
import html
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path, PurePosixPath
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
from typing import Any
|
||||
|
||||
from render import STYLE, render_page, split_frontmatter
|
||||
|
||||
|
||||
SOURCE_REVISION = re.compile(
|
||||
r'<meta name="policy-source-revision" content="([^"]+)">'
|
||||
)
|
||||
SOURCE_DIGEST = re.compile(r'<meta name="policy-source-digest" content="([^"]+)">')
|
||||
CLEAN_GIT_REVISION = re.compile(r"^[a-f0-9]{40}$")
|
||||
|
||||
|
||||
def _safe_path(value: str) -> PurePosixPath:
|
||||
path = PurePosixPath(value)
|
||||
if path.is_absolute() or ".." in path.parts or not path.parts:
|
||||
raise ValueError(f"unsafe publication path: {value!r}")
|
||||
return path
|
||||
|
||||
|
||||
def _source_revision(repo: Path, source: Path) -> str:
|
||||
digest = hashlib.sha256(source.read_bytes()).hexdigest()
|
||||
revision_env = "POLICY_NEXUS_SOURCE_REVISION_" + re.sub(
|
||||
r"[^A-Z0-9]+", "_", repo.name.upper()
|
||||
)
|
||||
supplied_revision = os.environ.get(revision_env, "")
|
||||
if supplied_revision:
|
||||
if not CLEAN_GIT_REVISION.fullmatch(supplied_revision):
|
||||
raise ValueError(
|
||||
f"{revision_env} must be a clean 40-hex Git commit, got {supplied_revision!r}"
|
||||
)
|
||||
return supplied_revision
|
||||
try:
|
||||
head = subprocess.run(
|
||||
["git", "-C", str(repo), "rev-parse", "HEAD"],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
).stdout.strip()
|
||||
relative = source.relative_to(repo)
|
||||
dirty = subprocess.run(
|
||||
["git", "-C", str(repo), "status", "--porcelain", "--", str(relative)],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
).stdout.strip()
|
||||
return f"{head}+working-tree.{digest[:12]}" if dirty else head
|
||||
except (OSError, subprocess.CalledProcessError, ValueError):
|
||||
return f"sha256:{digest}"
|
||||
|
||||
|
||||
def _add_interval(reviewed: str, interval: str) -> dt.date:
|
||||
date = dt.date.fromisoformat(reviewed)
|
||||
match = re.fullmatch(r"([1-9][0-9]*)([dmy])", interval)
|
||||
if not match:
|
||||
raise ValueError(f"invalid review interval {interval!r}; expected Nd, Nm or Ny")
|
||||
amount, unit = int(match.group(1)), match.group(2)
|
||||
if unit == "d":
|
||||
return date + dt.timedelta(days=amount)
|
||||
months = amount * (12 if unit == "y" else 1)
|
||||
month_index = date.month - 1 + months
|
||||
year, month = date.year + month_index // 12, month_index % 12 + 1
|
||||
month_lengths = (31, 29 if year % 4 == 0 and (year % 100 != 0 or year % 400 == 0) else 28,
|
||||
31, 30, 31, 30, 31, 31, 30, 31, 30, 31)
|
||||
return date.replace(year=year, month=month, day=min(date.day, month_lengths[month - 1]))
|
||||
|
||||
|
||||
def load_manifest(path: Path) -> dict[str, Any]:
|
||||
manifest = json.loads(path.read_text(encoding="utf-8"))
|
||||
if manifest.get("schema_version") != 1:
|
||||
raise ValueError("publication manifest schema_version must be 1")
|
||||
if not manifest.get("documents"):
|
||||
raise ValueError("publication manifest has no documents")
|
||||
seen: set[PurePosixPath] = set()
|
||||
for document in manifest["documents"]:
|
||||
for raw in (
|
||||
document["canonical_path"],
|
||||
document["revision_path"].replace("{revision}", "revision"),
|
||||
*document.get("legacy_paths", []),
|
||||
):
|
||||
path_value = _safe_path(raw)
|
||||
if path_value in seen:
|
||||
raise ValueError(f"duplicate publication path: {path_value}")
|
||||
seen.add(path_value)
|
||||
return manifest
|
||||
|
||||
|
||||
def _redirect(target: str, title: str) -> str:
|
||||
escaped = html.escape(target, quote=True)
|
||||
return (
|
||||
"<!doctype html><html lang=\"en\"><meta charset=\"utf-8\">"
|
||||
f'<meta http-equiv="refresh" content="0; url={escaped}">'
|
||||
f"<title>{html.escape(title)}</title>"
|
||||
f'<p>Moved permanently to <a href="{escaped}">{escaped}</a>.</p></html>\n'
|
||||
)
|
||||
|
||||
|
||||
def _index_page(site: dict[str, Any], records: list[dict[str, str]]) -> str:
|
||||
rows = []
|
||||
for record in records:
|
||||
rows.append(
|
||||
"<tr>"
|
||||
f'<td><a href="/{html.escape(record["canonical_path"], quote=True)}">'
|
||||
f'{html.escape(record["title"])}</a></td>'
|
||||
f'<td>{html.escape(record["status"])}</td>'
|
||||
f'<td>{html.escape(record["lifecycle"])}</td>'
|
||||
f'<td>{html.escape(record["revision"])}</td>'
|
||||
f'<td>{html.escape(record["owner"])}</td>'
|
||||
f'<td>{html.escape(record["last_reviewed"])}</td>'
|
||||
f'<td>{html.escape(record["review_due"])}</td>'
|
||||
f'<td>{html.escape(record["currency"])}</td>'
|
||||
"</tr>"
|
||||
)
|
||||
return (
|
||||
"<!doctype html><html lang=\"en\"><meta charset=\"utf-8\">"
|
||||
f"<title>{html.escape(site['title'])}</title><style>{STYLE.read_text()}</style>"
|
||||
'<div class="wrap"><header><div class="eyebrow"><span>policy surface</span>'
|
||||
"<span>generated from canonical sources — do not edit</span></div>"
|
||||
f"<h1>{html.escape(site['title'])}</h1>"
|
||||
'<p class="sub">Canon and architecture decisions at stable addresses, with visible currency.</p>'
|
||||
"</header><main><table><thead><tr><th>Document</th><th>Status</th>"
|
||||
"<th>Lifecycle</th><th>Revision</th><th>Owner</th><th>Reviewed</th>"
|
||||
"<th>Review due</th><th>Currency</th>"
|
||||
f"</tr></thead><tbody>{''.join(rows)}</tbody></table></main></div></html>\n"
|
||||
)
|
||||
|
||||
|
||||
def build(
|
||||
manifest_path: Path,
|
||||
output: Path,
|
||||
*,
|
||||
as_of: dt.date | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
manifest_path = manifest_path.resolve()
|
||||
manifest = load_manifest(manifest_path)
|
||||
as_of = as_of or dt.date.today()
|
||||
repository_paths = {
|
||||
name: (manifest_path.parent / config["path"]).resolve()
|
||||
for name, config in manifest["repositories"].items()
|
||||
}
|
||||
output_parent = output.resolve().parent
|
||||
output_parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = Path(tempfile.mkdtemp(prefix=f".{output.name}-", dir=output_parent))
|
||||
if output.exists():
|
||||
shutil.copytree(output, temporary, dirs_exist_ok=True)
|
||||
|
||||
records: list[dict[str, str]] = []
|
||||
try:
|
||||
for document in manifest["documents"]:
|
||||
repo = repository_paths[document["source_repo"]]
|
||||
source = (repo / document["source_path"]).resolve()
|
||||
if not source.is_file() or repo not in source.parents:
|
||||
raise FileNotFoundError(f"canonical source unavailable: {source}")
|
||||
meta, _markdown = split_frontmatter(source.read_text(encoding="utf-8"))
|
||||
if meta.get("id") != document["id"]:
|
||||
raise ValueError(
|
||||
f"{source}: manifest id {document['id']!r} does not match {meta.get('id')!r}"
|
||||
)
|
||||
for required_field in ("title", "status", "owner"):
|
||||
if not meta.get(required_field):
|
||||
raise ValueError(f"{source}: {required_field} is required for publication")
|
||||
revision = meta.get("revision") or meta.get("version")
|
||||
if not revision:
|
||||
raise ValueError(f"{source}: revision or version is required")
|
||||
source_revision = _source_revision(repo, source)
|
||||
source_digest = hashlib.sha256(source.read_bytes()).hexdigest()
|
||||
reviewed = meta.get("last_reviewed") or meta.get("updated")
|
||||
interval = document.get("review_interval") or meta.get("review_interval")
|
||||
if not reviewed or not interval:
|
||||
raise ValueError(f"{source}: review date and interval are required")
|
||||
review_due = _add_interval(reviewed, interval)
|
||||
lifecycle = document.get("lifecycle", "active")
|
||||
if lifecycle not in {"active", "superseded", "withdrawn"}:
|
||||
raise ValueError(
|
||||
f"{document['id']}: lifecycle must be active, superseded or withdrawn"
|
||||
)
|
||||
successor = document.get("successor", "")
|
||||
if lifecycle == "superseded" and not successor:
|
||||
raise ValueError(f"{document['id']}: superseded documents require successor")
|
||||
publication = {
|
||||
"source_repo": document["source_repo"],
|
||||
"source_path": document["source_path"],
|
||||
"source_revision": source_revision,
|
||||
"source_digest": source_digest,
|
||||
"review_due": review_due.isoformat() if review_due else "",
|
||||
}
|
||||
revision_page, meta, _sections = render_page(
|
||||
source,
|
||||
subtitle=document.get("subtitle", ""),
|
||||
publication=publication,
|
||||
)
|
||||
current_publication = publication | {
|
||||
"lifecycle": lifecycle,
|
||||
"successor": successor,
|
||||
"stale": "true" if review_due and review_due < as_of else "false",
|
||||
}
|
||||
current_page, _current_meta, _current_sections = render_page(
|
||||
source,
|
||||
subtitle=document.get("subtitle", ""),
|
||||
publication=current_publication,
|
||||
)
|
||||
canonical = _safe_path(document["canonical_path"])
|
||||
revision_path = _safe_path(document["revision_path"].format(revision=revision))
|
||||
revision_target = temporary / revision_path
|
||||
if revision_target.exists():
|
||||
existing_revision = revision_target.read_text(encoding="utf-8")
|
||||
old_revision = SOURCE_REVISION.search(existing_revision)
|
||||
old_digest = SOURCE_DIGEST.search(existing_revision)
|
||||
if not old_revision or not old_digest:
|
||||
raise RuntimeError(
|
||||
f"immutable revision {revision_path} has incomplete source metadata"
|
||||
)
|
||||
if html.unescape(old_digest.group(1)) != source_digest:
|
||||
raise RuntimeError(
|
||||
f"immutable revision {revision_path} already records content digest "
|
||||
f"{old_digest.group(1)}; source is now {source_digest}. "
|
||||
"Publish a new revision id."
|
||||
)
|
||||
canonical_target = temporary / canonical
|
||||
canonical_target.parent.mkdir(parents=True, exist_ok=True)
|
||||
canonical_target.write_text(current_page, encoding="utf-8")
|
||||
if not revision_target.exists():
|
||||
revision_target.parent.mkdir(parents=True, exist_ok=True)
|
||||
revision_target.write_text(revision_page, encoding="utf-8")
|
||||
for legacy in document.get("legacy_paths", []):
|
||||
target = temporary / _safe_path(legacy)
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
canonical_url = "/" + canonical.as_posix()
|
||||
target.write_text(_redirect(canonical_url, meta["title"]), encoding="utf-8")
|
||||
records.append(
|
||||
{
|
||||
"id": document["id"],
|
||||
"title": meta["title"],
|
||||
"status": meta["status"],
|
||||
"revision": revision,
|
||||
"owner": meta["owner"],
|
||||
"last_reviewed": reviewed,
|
||||
"review_due": review_due.isoformat(),
|
||||
"currency": "stale" if review_due < as_of else "current",
|
||||
"lifecycle": lifecycle,
|
||||
"canonical_path": canonical.as_posix(),
|
||||
"revision_path": revision_path.as_posix(),
|
||||
"source_revision": source_revision,
|
||||
"source_digest": source_digest,
|
||||
}
|
||||
)
|
||||
|
||||
(temporary / "index.html").write_text(
|
||||
_index_page(manifest["site"], records), encoding="utf-8"
|
||||
)
|
||||
(temporary / "publication-manifest.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"schema_version": 1,
|
||||
"generated_as_of": as_of.isoformat(),
|
||||
"documents": records,
|
||||
},
|
||||
indent=2,
|
||||
sort_keys=True,
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
if output.exists():
|
||||
shutil.rmtree(output)
|
||||
os.replace(temporary, output)
|
||||
except BaseException:
|
||||
shutil.rmtree(temporary, ignore_errors=True)
|
||||
raise
|
||||
return records
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("manifest", type=Path)
|
||||
parser.add_argument("--output", type=Path, default=Path("build"))
|
||||
parser.add_argument("--as-of", type=dt.date.fromisoformat)
|
||||
args = parser.parse_args()
|
||||
records = build(args.manifest, args.output, as_of=args.as_of)
|
||||
print(f"{args.output}: published {len(records)} document(s)")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
38
tools/check_currency.py
Normal file
38
tools/check_currency.py
Normal file
|
|
@ -0,0 +1,38 @@
|
|||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import datetime as dt
|
||||
from pathlib import Path
|
||||
|
||||
from build_site import _add_interval, load_manifest
|
||||
from render import split_frontmatter
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("manifest", type=Path)
|
||||
parser.add_argument("--as-of", type=dt.date.fromisoformat, default=dt.date.today())
|
||||
args = parser.parse_args()
|
||||
manifest_path = args.manifest.resolve()
|
||||
manifest = load_manifest(manifest_path)
|
||||
stale = 0
|
||||
for document in manifest["documents"]:
|
||||
repo = manifest_path.parent / manifest["repositories"][document["source_repo"]]["path"]
|
||||
source = (repo / document["source_path"]).resolve()
|
||||
meta, _markdown = split_frontmatter(source.read_text(encoding="utf-8"))
|
||||
reviewed = meta.get("last_reviewed") or meta.get("updated")
|
||||
interval = document.get("review_interval") or meta.get("review_interval")
|
||||
if not reviewed or not interval:
|
||||
print(f"UNDECLARED {document['id']}: review date/interval missing")
|
||||
stale += 1
|
||||
continue
|
||||
due = _add_interval(reviewed, interval)
|
||||
state = "STALE" if due < args.as_of else "current"
|
||||
print(f"{state} {document['id']}: reviewed {reviewed}, due {due}")
|
||||
stale += state == "STALE"
|
||||
return 1 if stale else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
139
tools/render.py
139
tools/render.py
|
|
@ -19,7 +19,7 @@ Stdlib only, per the estate's structure-not-tooling stance: a publishing step
|
|||
that needs its own toolchain is a publishing step that stops being run.
|
||||
|
||||
Usage:
|
||||
python3 tools/render-artifact.py <source.md> --output <page.html> \\
|
||||
python3 tools/render.py <source.md> --output <page.html> \\
|
||||
[--title "Name"] [--subtitle "..."]
|
||||
"""
|
||||
|
||||
|
|
@ -39,7 +39,7 @@ EM = re.compile(r"(?<![*\w])\*([^*]+)\*(?!\*)")
|
|||
LINK = re.compile(r"\[([^\]]+)\]\(([^)]+)\)")
|
||||
HEADING = re.compile(r"^(#{1,4})\s+(.*)$")
|
||||
SECTION_NO = re.compile(r"^(\d+)\.\s+(.*)$")
|
||||
LEVEL_CELL = re.compile(r"^\*\*([IAEPR])(\d)\*\*$")
|
||||
LEVEL_CELL = re.compile(r"^\*\*([IAEPRV])(\d)\*\*$")
|
||||
DECISION = re.compile(r"^\*\*(Decision [\d.]+[^*]*)\*\*(.*)$", re.S)
|
||||
|
||||
|
||||
|
|
@ -91,10 +91,10 @@ def parse_table(lines: list[str], start: int) -> tuple[list[list[str]], int]:
|
|||
def render_ladder(rows: list[list[str]]) -> str:
|
||||
"""A level table becomes a stepped scale. Colour depth encodes strength."""
|
||||
body = rows[1:]
|
||||
plane = LEVEL_CELL.match(body[0][0]).group(1)
|
||||
axis = LEVEL_CELL.match(body[0][0]).group(1)
|
||||
names = {
|
||||
"I": "Identity", "A": "Authorization", "E": "Enforcement",
|
||||
"P": "Placement", "R": "Retention",
|
||||
"P": "Placement", "R": "Retention", "V": "Availability",
|
||||
}
|
||||
rungs = []
|
||||
for cells in body:
|
||||
|
|
@ -104,15 +104,15 @@ def render_ladder(rows: list[list[str]]) -> str:
|
|||
n = int(match.group(2))
|
||||
text = cells[1] if len(cells) > 1 else ""
|
||||
rungs.append(
|
||||
f'<div class="rung r{n}"><span class="code">{plane}{n}</span>'
|
||||
f'<div class="rung r{n}"><span class="code">{axis}{n}</span>'
|
||||
f'<span class="txt">{inline(text)}</span></div>'
|
||||
)
|
||||
while len(rungs) < 5:
|
||||
rungs.append('<div class="rung na"><span class="code">—</span>'
|
||||
f'<span class="txt">Ladder ends at {plane}{len(rungs) - 1}.</span></div>')
|
||||
f'<span class="txt">Ladder ends at {axis}{len(rungs) - 1}.</span></div>')
|
||||
return (
|
||||
'<div class="ladder"><div class="pname">'
|
||||
f'{names.get(plane, plane)}<span>plane {plane}</span></div>'
|
||||
f'{names.get(axis, axis)}<span>axis {axis}</span></div>'
|
||||
f'<div class="rungs">{"".join(rungs)}</div></div>'
|
||||
)
|
||||
|
||||
|
|
@ -326,25 +326,30 @@ def render_body(markdown: str) -> tuple[str, list[tuple[str, str, str]]]:
|
|||
return "\n".join(out), rail
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("source", type=pathlib.Path)
|
||||
ap.add_argument("--output", required=True, type=pathlib.Path)
|
||||
ap.add_argument("--title", default=None)
|
||||
ap.add_argument("--subtitle", default="")
|
||||
args = ap.parse_args()
|
||||
|
||||
meta, markdown = split_frontmatter(args.source.read_text())
|
||||
def render_page(
|
||||
source: pathlib.Path,
|
||||
*,
|
||||
title: str | None = None,
|
||||
subtitle: str = "",
|
||||
publication: dict[str, str] | None = None,
|
||||
) -> tuple[str, dict, int]:
|
||||
meta, markdown = split_frontmatter(source.read_text(encoding="utf-8"))
|
||||
body, rail = render_body(markdown)
|
||||
|
||||
title = args.title or meta.get("title", args.source.stem)
|
||||
display = title.split(":")[0].strip()
|
||||
resolved_title = title or meta.get("title", source.stem)
|
||||
display = resolved_title.split(":")[0].strip()
|
||||
eyebrow = " ".join(
|
||||
f"<span{' class=\"stat\"' if k == 'status' else ''}>{html.escape(v)}</span>"
|
||||
for k, v in (
|
||||
("id", meta.get("id", "")),
|
||||
("status", f"{meta.get('status', '')} · {meta.get('revision', '')}".strip(" ·")),
|
||||
("date", meta.get("date", "")),
|
||||
("owner", meta.get("owner", "")),
|
||||
(
|
||||
"review",
|
||||
f"reviewed {meta.get('last_reviewed', '')}"
|
||||
if meta.get("last_reviewed")
|
||||
else "",
|
||||
),
|
||||
)
|
||||
if v
|
||||
)
|
||||
|
|
@ -353,23 +358,103 @@ def main() -> int:
|
|||
for a, n, t in rail
|
||||
)
|
||||
|
||||
publication = publication or {}
|
||||
source_note = " · ".join(
|
||||
item
|
||||
for item in (
|
||||
publication.get("source_repo", ""),
|
||||
publication.get("source_path", ""),
|
||||
publication.get("source_revision", ""),
|
||||
)
|
||||
if item
|
||||
)
|
||||
source_line = (
|
||||
f'<p class="sub">Source: <code>{html.escape(source_note)}</code></p>'
|
||||
if source_note
|
||||
else ""
|
||||
)
|
||||
review_due = publication.get("review_due", "")
|
||||
review_line = (
|
||||
f'<p class="sub">Review due: {html.escape(review_due)}</p>'
|
||||
if review_due
|
||||
else ""
|
||||
)
|
||||
lifecycle = publication.get("lifecycle", "active")
|
||||
successor = publication.get("successor", "")
|
||||
lifecycle_notice = ""
|
||||
if lifecycle == "superseded":
|
||||
successor_link = (
|
||||
f' <a href="{html.escape(successor, quote=True)}">Read its successor.</a>'
|
||||
if successor
|
||||
else ""
|
||||
)
|
||||
lifecycle_notice = (
|
||||
'<div class="rule-quote"><p><strong>Superseded.</strong>'
|
||||
f" This address is retained as part of the policy record.{successor_link}</p></div>"
|
||||
)
|
||||
elif lifecycle == "withdrawn":
|
||||
lifecycle_notice = (
|
||||
'<div class="rule-quote"><p><strong>Withdrawn.</strong> '
|
||||
"This document is retained for historical reference and is not current policy."
|
||||
"</p></div>"
|
||||
)
|
||||
currency_notice = (
|
||||
'<div class="rule-quote"><p><strong>Review overdue.</strong> '
|
||||
f"This document was due for review on {html.escape(review_due)}.</p></div>"
|
||||
if review_due and publication.get("stale") == "true"
|
||||
else ""
|
||||
)
|
||||
source_revision_meta = (
|
||||
f'<meta name="policy-source-revision" content="'
|
||||
f'{html.escape(publication.get("source_revision", ""), quote=True)}">\n'
|
||||
if publication.get("source_revision")
|
||||
else ""
|
||||
)
|
||||
source_digest_meta = (
|
||||
f'<meta name="policy-source-digest" content="'
|
||||
f'{html.escape(publication.get("source_digest", ""), quote=True)}">\n'
|
||||
if publication.get("source_digest")
|
||||
else ""
|
||||
)
|
||||
page = (
|
||||
f"<title>{html.escape(display)}</title>\n"
|
||||
"<!doctype html>\n<html lang=\"en\"><meta charset=\"utf-8\">\n"
|
||||
+ source_revision_meta
|
||||
+ source_digest_meta
|
||||
+ f"<title>{html.escape(display)}</title>\n"
|
||||
f"<style>\n{STYLE.read_text()}\n</style>\n"
|
||||
'<div class="wrap"><header>'
|
||||
f'<div class="eyebrow">{eyebrow}<span>generated from canon — do not edit</span></div>'
|
||||
f'<div class="eyebrow">{eyebrow}<span>generated from canonical source — do not edit</span></div>'
|
||||
f"<h1>{html.escape(display)}</h1>"
|
||||
+ (f'<p class="sub">{html.escape(args.subtitle)}</p>' if args.subtitle else "")
|
||||
+ (f'<p class="sub">{html.escape(subtitle)}</p>' if subtitle else "")
|
||||
+ source_line
|
||||
+ review_line
|
||||
+ '</header><div class="layout">'
|
||||
f'<nav class="rail" aria-label="Sections"><ol>{rail_html}</ol></nav>'
|
||||
f"<main>{body}"
|
||||
f"<main>{lifecycle_notice}{currency_notice}{body}"
|
||||
f'<footer><span>{html.escape(meta.get("id", ""))} · '
|
||||
f'{html.escape(meta.get("revision", ""))} · {html.escape(meta.get("status", ""))}</span>'
|
||||
"<span>generated from the-custodian/canon</span></footer>"
|
||||
"</main></div></div>\n"
|
||||
f"<span>{html.escape(source_note or 'generated from canonical source')}</span></footer>"
|
||||
"</main></div></div></html>\n"
|
||||
)
|
||||
args.output.write_text(page)
|
||||
print(f"{args.output}: {len(rail)} sections, {len(page)} bytes")
|
||||
return page, meta, len(rail)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("source", type=pathlib.Path)
|
||||
ap.add_argument("--output", required=True, type=pathlib.Path)
|
||||
ap.add_argument("--title", default=None)
|
||||
ap.add_argument("--subtitle", default="")
|
||||
args = ap.parse_args()
|
||||
|
||||
page, _meta, section_count = render_page(
|
||||
args.source,
|
||||
title=args.title,
|
||||
subtitle=args.subtitle,
|
||||
)
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(page, encoding="utf-8")
|
||||
print(f"{args.output}: {section_count} sections, {len(page)} bytes")
|
||||
return 0
|
||||
|
||||
|
||||
|
|
|
|||
126
tools/verify_release.py
Normal file
126
tools/verify_release.py
Normal file
|
|
@ -0,0 +1,126 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Fail closed unless a built policy site is safe to publish as an OCI release."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import html
|
||||
import json
|
||||
from pathlib import Path, PurePosixPath
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
|
||||
HEX_DIGEST = re.compile(r"^[a-f0-9]{64}$")
|
||||
CLEAN_GIT_REVISION = re.compile(r"^[a-f0-9]{40}$")
|
||||
SOURCE_REVISION = re.compile(
|
||||
r'<meta name="policy-source-revision" content="([^"]+)">'
|
||||
)
|
||||
SOURCE_DIGEST = re.compile(r'<meta name="policy-source-digest" content="([^"]+)">')
|
||||
|
||||
|
||||
def _safe_relative(value: str) -> PurePosixPath:
|
||||
path = PurePosixPath(value)
|
||||
if path.is_absolute() or ".." in path.parts or not path.parts:
|
||||
raise ValueError(f"unsafe release path: {value!r}")
|
||||
return path
|
||||
|
||||
|
||||
def _page_provenance(path: Path) -> tuple[str, str]:
|
||||
page = path.read_text(encoding="utf-8")
|
||||
revision = SOURCE_REVISION.search(page)
|
||||
digest = SOURCE_DIGEST.search(page)
|
||||
if not revision or not digest:
|
||||
raise ValueError(f"{path}: missing source provenance metadata")
|
||||
return html.unescape(revision.group(1)), html.unescape(digest.group(1))
|
||||
|
||||
|
||||
def verify(build: Path) -> dict[str, Any]:
|
||||
build = build.resolve()
|
||||
if not build.is_dir():
|
||||
raise ValueError(f"release directory does not exist: {build}")
|
||||
for path in build.rglob("*"):
|
||||
if path.is_symlink():
|
||||
raise ValueError(f"release tree contains a symlink: {path.relative_to(build)}")
|
||||
|
||||
index = build / "index.html"
|
||||
manifest_path = build / "publication-manifest.json"
|
||||
if not index.is_file() or not manifest_path.is_file():
|
||||
raise ValueError("release requires index.html and publication-manifest.json")
|
||||
|
||||
manifest_bytes = manifest_path.read_bytes()
|
||||
manifest = json.loads(manifest_bytes)
|
||||
if manifest.get("schema_version") != 1:
|
||||
raise ValueError("publication manifest schema_version must be 1")
|
||||
if not manifest.get("generated_as_of"):
|
||||
raise ValueError("publication manifest generated_as_of is required")
|
||||
documents = manifest.get("documents")
|
||||
if not isinstance(documents, list) or not documents:
|
||||
raise ValueError("publication manifest must contain at least one document")
|
||||
|
||||
verified: list[str] = []
|
||||
for document in documents:
|
||||
document_id = document.get("id", "<unknown>")
|
||||
for field in (
|
||||
"id",
|
||||
"title",
|
||||
"status",
|
||||
"revision",
|
||||
"owner",
|
||||
"last_reviewed",
|
||||
"review_due",
|
||||
"canonical_path",
|
||||
"revision_path",
|
||||
):
|
||||
if not document.get(field) or document.get(field) == "unknown":
|
||||
raise ValueError(f"{document_id}: release metadata field {field} is required")
|
||||
source_revision = document.get("source_revision", "")
|
||||
source_digest = document.get("source_digest", "")
|
||||
if not CLEAN_GIT_REVISION.fullmatch(source_revision):
|
||||
raise ValueError(
|
||||
f"{document_id}: production source_revision must be a clean 40-hex Git commit; "
|
||||
f"got {source_revision!r}"
|
||||
)
|
||||
if not HEX_DIGEST.fullmatch(source_digest):
|
||||
raise ValueError(f"{document_id}: invalid source_digest {source_digest!r}")
|
||||
|
||||
canonical = build / _safe_relative(document["canonical_path"])
|
||||
revision = build / _safe_relative(document["revision_path"])
|
||||
if not canonical.is_file() or not revision.is_file():
|
||||
raise ValueError(f"{document_id}: canonical or immutable revision page is missing")
|
||||
|
||||
current_revision, current_digest = _page_provenance(canonical)
|
||||
immutable_revision, immutable_digest = _page_provenance(revision)
|
||||
if (current_revision, current_digest) != (source_revision, source_digest):
|
||||
raise ValueError(f"{document_id}: canonical page provenance differs from manifest")
|
||||
if immutable_digest != source_digest:
|
||||
raise ValueError(f"{document_id}: immutable revision digest differs from manifest")
|
||||
if not CLEAN_GIT_REVISION.fullmatch(immutable_revision):
|
||||
raise ValueError(
|
||||
f"{document_id}: immutable revision page was not built from a clean Git commit"
|
||||
)
|
||||
verified.append(document_id)
|
||||
|
||||
return {
|
||||
"schema_version": "policy-nexus-release/v1",
|
||||
"publication_manifest_digest": hashlib.sha256(manifest_bytes).hexdigest(),
|
||||
"generated_as_of": manifest["generated_as_of"],
|
||||
"documents": verified,
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("build", nargs="?", type=Path, default=Path("build"))
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
evidence = verify(args.build)
|
||||
except (KeyError, OSError, ValueError, json.JSONDecodeError) as exc:
|
||||
parser.error(str(exc))
|
||||
print(json.dumps(evidence, sort_keys=True))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Loading…
Add table
Add a link
Reference in a new issue