tegwick
c1b60f322e
All checks were successful
Build and publish policy-nexus image / build-and-push (push) Successful in 1m10s
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a058f3-8ba0-7692-a042-9a870fc3d663
339 lines
14 KiB
Python
339 lines
14 KiB
Python
#!/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
|
|
source_root = os.environ.get("POLICY_NEXUS_SOURCE_ROOT", "")
|
|
if source_root:
|
|
root = Path(source_root).resolve()
|
|
lock_path = root / "source-lock.json"
|
|
try:
|
|
lock = json.loads(lock_path.read_text(encoding="utf-8"))
|
|
locked_revision = lock["repositories"][repo.name]["revision"]
|
|
except (KeyError, OSError, TypeError, json.JSONDecodeError) as exc:
|
|
raise ValueError(
|
|
f"{repo.name}: source revision is missing from {lock_path}"
|
|
) from exc
|
|
if not isinstance(locked_revision, str) or not CLEAN_GIT_REVISION.fullmatch(
|
|
locked_revision
|
|
):
|
|
raise ValueError(
|
|
f"{repo.name}: locked source revision must be a clean 40-hex Git commit, "
|
|
f"got {locked_revision!r}"
|
|
)
|
|
return locked_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 repository_paths(manifest_path: Path, manifest: dict[str, Any]) -> dict[str, Path]:
|
|
"""Resolve local or fetched source checkouts for a publication manifest."""
|
|
source_root = os.environ.get("POLICY_NEXUS_SOURCE_ROOT")
|
|
paths: dict[str, Path] = {}
|
|
for name, config in manifest["repositories"].items():
|
|
fetched = (Path(source_root) / name).resolve() if source_root else None
|
|
paths[name] = (
|
|
fetched
|
|
if fetched is not None and fetched.is_dir()
|
|
else (manifest_path.parent / config["path"]).resolve()
|
|
)
|
|
return paths
|
|
|
|
|
|
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">Governing documents and public risk records 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()
|
|
resolved_repositories = repository_paths(manifest_path, manifest)
|
|
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 = resolved_repositories[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}"
|
|
)
|
|
if (
|
|
document["source_repo"] == "risk-nexus"
|
|
and meta.get("disclosure") != "public"
|
|
):
|
|
raise ValueError(
|
|
f"{source}: risk-nexus publications require disclosure: public"
|
|
)
|
|
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_repo": document["source_repo"],
|
|
"source_path": document["source_path"],
|
|
"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())
|