"""Deploy the tested main-branch website to its existing hidden-repository Space. GitHub decides which runs may publish; Hugging Face authenticates the workflow with a short-lived, single-Space OIDC token. Neither credential is saved in the bundle, receipt or logs. This module does not publish models or datasets. """ from __future__ import annotations import argparse import hashlib import json import os from pathlib import Path import re import subprocess import tempfile import time from typing import Callable from urllib.error import HTTPError, URLError from urllib.parse import urlencode, urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener from .hf_recipe_hub import _repository_hidden, build, publish GITHUB_REPOSITORY = "ChTauchmann/hessian-recipes" SPACE = "ctauchmann/hessian-recipes" RESOURCE = "spaces/" + SPACE CI_PATH = ".github/workflows/ci.yml" def check_eligibility(expected_sha: str, *, event_name: str, event: dict, ref: str, github_get: Callable[[str], dict]) -> dict: """Reject PRs, failed tests, other workflows and superseded main commits.""" if not re.fullmatch(r"[0-9a-f]{40}", expected_sha) or ref != "refs/heads/main": raise ValueError("Deployment requires an exact main-branch commit") workflow = github_get("actions/workflows/ci.yml") if workflow.get("path") != CI_PATH or workflow.get("name") != "checks": raise ValueError("The required checks workflow identity changed") def trusted(run: dict) -> bool: if not isinstance(run, dict) or not isinstance(run.get("head_repository"), dict): return False return (run.get("name") == "checks" and run.get("path") == CI_PATH and run.get("workflow_id") == workflow.get("id") and type(run.get("workflow_id")) is int and run.get("status") == "completed" and run.get("conclusion") == "success" and run.get("event") == "push" and run.get("head_branch") == "main" and run.get("head_sha") == expected_sha and run["head_repository"].get("full_name") == GITHUB_REPOSITORY) if event_name == "workflow_run": payload = event.get("workflow_run", {}) run_id = payload.get("id") if (event.get("action") != "completed" or type(run_id) is not int or not trusted(payload) or event.get("repository", {}).get("full_name") != GITHUB_REPOSITORY): raise ValueError("Expected a completed checks event for the exact commit") run = github_get(f"actions/runs/{run_id}") if run.get("id") != run_id: raise ValueError("Checks run identity does not match the event") elif event_name == "workflow_dispatch": runs = github_get("actions/workflows/ci.yml/runs?head_sha=" + expected_sha + "&branch=main&per_page=20").get("workflow_runs", []) if not runs: raise ValueError("No completed main-branch checks found for this commit") # Prefer the newest execution; a failed newer run cannot be bypassed by # selecting an earlier successful execution of the same source commit. run = max(runs, key=lambda item: item.get("id", 0)) else: raise ValueError("Only completed checks and main-branch retries may deploy") if not trusted(run): raise ValueError("Deployment requires successful push checks on this repository's main") if github_get("commits/main").get("sha") != expected_sha: raise ValueError("This commit was superseded on main; the newer commit will deploy") return {"run_id": run["id"], "source_commit": expected_sha, "workflow": CI_PATH} class _NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise ValueError("Unexpected redirect during deployment verification") _STATIC_APP_ORIGIN = "https://ctauchmann-hessian-recipes.static.hf.space" class _StaticRootRedirect(HTTPRedirectHandler): """Allow only HF Static's anonymous root-to-index normalization, once.""" def __init__(self, source: str): parsed = urlsplit(source) if (parsed.scheme + "://" + parsed.netloc != _STATIC_APP_ORIGIN or parsed.path != "/" or parsed.fragment or not re.fullmatch(r"protected-check=[0-9]+", parsed.query)): raise ValueError("Unexpected Protected root probe URL") self.source = source self.target = _STATIC_APP_ORIGIN + "/index.html?" + parsed.query self.followed = False def redirect_request(self, req, fp, code, msg, headers, newurl): credential_headers = {"authorization", "proxy-authorization", "cookie"} if (self.followed or code != 302 or req.full_url != self.source or req.get_method() != "GET" or req.data is not None or any(name.lower() in credential_headers for name, _ in req.header_items()) or newurl != self.target): raise ValueError("Unexpected redirect during deployment verification") self.followed = True return Request(self.target, headers={}, method="GET") def _read_protected_root(url: str) -> bytes: """Read the public Static root with its one permitted canonical redirect.""" request = Request(url, headers={}, method="GET") with build_opener(_StaticRootRedirect(url)).open(request, timeout=20) as response: raw = response.read(2_000_001) if len(raw) > 2_000_000: raise ValueError("Deployment verification response exceeds its size bound") return raw def _read(url: str, headers: dict[str, str]) -> bytes: request = Request(url, headers=headers) with build_opener(_NoRedirect()).open(request, timeout=20) as response: raw = response.read(2_000_001) if len(raw) > 2_000_000: raise ValueError("Deployment verification response exceeds its size bound") return raw def github_get(path: str) -> dict: token = os.environ.get("GITHUB_TOKEN") if not token or os.environ.get("GITHUB_REPOSITORY") != GITHUB_REPOSITORY: raise ValueError("Deployment requires this repository's GitHub Actions identity") raw = _read("https://api.github.com/repos/" + GITHUB_REPOSITORY + "/" + path, {"Authorization": "Bearer " + token, "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28"}) return json.loads(raw) def _gate(expected_sha: str) -> dict: event = json.loads(Path(os.environ["GITHUB_EVENT_PATH"]).read_text()) return check_eligibility(expected_sha, event_name=os.environ["GITHUB_EVENT_NAME"], event=event, ref=os.environ["GITHUB_REF"], github_get=github_get) def _source(repo: Path, expected_sha: str) -> None: head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=repo, text=True).strip() if head != expected_sha: raise ValueError("Checkout does not match the tested deployment commit") def _inventory(folder: Path) -> dict[str, str]: if folder.is_symlink() or any(path.is_symlink() for path in folder.rglob("*")): raise ValueError("Deployment bundles may not contain symlinks") return {str(path.relative_to(folder)): hashlib.sha256(path.read_bytes()).hexdigest() for path in folder.rglob("*") if path.is_file()} def verify_bundle(repo: Path, bundle: Path, expected_sha: str, visibility: str = "private") -> dict: """Rebuild independently before authenticating; artifacts cannot supply code.""" _source(repo, expected_sha) with tempfile.TemporaryDirectory(prefix="hessian-deploy-") as temporary: reference = Path(temporary) / "reference" manifest = build(repo, reference) if visibility == "private" else build(repo, reference, visibility=visibility) if manifest["uncommitted_source_paths"]: raise ValueError("Deployment requires a clean source snapshot") if _inventory(bundle) != _inventory(reference): raise ValueError("Deployment bundle differs from the tested repository source") return manifest def _app_host(host: str | None) -> str: if not host: raise ValueError("Private Space has no running App host") host = host if host.startswith("https://") else "https://" + host origin = urlsplit(host) if (origin.scheme != "https" or origin.netloc not in { "ctauchmann-hessian-recipes.static.hf.space", "ctauchmann-hessian-recipes.hf.space"} or origin.path not in {"", "/"} or origin.query or origin.fragment): raise ValueError("Unexpected private App host") return host.rstrip("/") def verify_privacy(host: str) -> None: """Probe without credentials before any upload, and again after delivery.""" host = _app_host(host) for path in ("/", "/index.html", "/assets/site.css", "/sources/docs/validation.md"): try: _read(host + path + "?private-check=" + str(time.time_ns()), {}) except HTTPError as error: if error.code in {401, 403, 404}: continue raise ValueError("Unexpected response during anonymous-access check") from None raise ValueError("Private website unexpectedly permits anonymous access") def _space_jwt(token: str) -> str: """Exchange Hub authorization for five minutes of private App access. Static App ingress does not accept a repository publisher token directly. The first-party Space JWT endpoint issues the signed App capability; neither it nor the signed request URL is written to logs or deployment receipts. """ try: response = json.loads(_read( "https://huggingface.co/api/spaces/" + SPACE + "/jwt?expires_in=300", {"Authorization": "Bearer " + token}, )) except HTTPError as error: raise ValueError("Private App token exchange failed (HTTP " + str(error.code) + ")") from None except URLError: raise ValueError("Private App token exchange failed (transport error)") from None signed = response.get("token") if isinstance(response, dict) else None if not isinstance(signed, str) or not re.fullmatch(r"[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+", signed): raise ValueError("Invalid private App token exchange response") return signed def _matching_space(info, visibility: str = "private") -> bool: return (_repository_hidden(info, visibility) and info.id == SPACE and getattr(info, "sdk", None) == "static") def _matches_entry(raw: bytes, entry) -> bool: if entry.lfs: return entry.lfs.sha256 == hashlib.sha256(raw).hexdigest() return entry.blob_id == hashlib.sha1( b"blob " + str(len(raw)).encode() + b"\0" + raw).hexdigest() def verify_protected(api, host: str) -> None: """Verify public App bytes against the current hidden repository tree. Preflight compares to the currently deployed revision, allowing a new build to change its branding. No account token or signed App URL is used. """ host = _app_host(host) info = api.repo_info(SPACE, repo_type="space", files_metadata=True) if not _matching_space(info, "protected"): raise ValueError("Protected publication requires the existing hidden Static Space") siblings = {entry.rfilename: entry for entry in info.siblings} for path, filename in (("/", "index.html"), ("/index.html", "index.html"), ("/assets/site.css", "assets/site.css"), ("/sources/docs/validation.md", "sources/docs/validation.md"), ("/hub-manifest.json", "hub-manifest.json")): try: url = host + path + "?protected-check=" + str(time.time_ns()) raw = _read_protected_root(url) if path == "/" else _read(url, {}) except HTTPError as error: raise ValueError("Protected App content is unavailable (HTTP " + str(error.code) + ")") from None if filename == "index.html": # HF Static ingress injects its own script into the HTML head. Check # the actual recipe document, while JSON/CSS/source remain exact. markers = (b"", b"Post-training and evaluation", b'<body class="home-page">', b"hub-manifest.json") if not all(marker in raw for marker in markers): raise ValueError("Protected App did not serve the expected recipe page") elif filename not in siblings or not _matches_entry(raw, siblings[filename]): raise ValueError("Protected App content differs from the current repository bytes") # A public App must not expose the Hub repository endpoints. Do not follow # redirects or send the publisher's Authorization header to these probes. for route in ("raw/main/README.md", "resolve/main/README.md"): try: _read("https://huggingface.co/spaces/" + SPACE + "/" + route + "?protected-check=" + str(time.time_ns()), {}) except HTTPError as error: if error.code in {401, 403, 404}: continue raise ValueError("Unexpected response during anonymous repository check") from None raise ValueError("Protected Hub repository unexpectedly permits anonymous access") def verify_repository(api, bundle: Path, receipt: dict, *, visibility: str = "private") -> dict: """Verify hidden repository head and every file without App authorization.""" info = api.repo_info(SPACE, repo_type="space", files_metadata=True) if not _matching_space(info, visibility) or info.sha != receipt["commit"]: raise ValueError("Published Space head, owner, SDK or private visibility changed") files = _inventory(bundle) siblings = {entry.rfilename: entry for entry in info.siblings} if set(siblings) - {".gitattributes"} != set(files): raise ValueError("Remote Space inventory does not match the generated website") for name in files: if not _matches_entry((bundle / name).read_bytes(), siblings[name]): raise ValueError("Remote website content differs from its deployment bundle") result = {"remote_files_verified": len(files), "private": True} if visibility == "protected": result |= {"visibility": "protected", "protected": True} return result def verify_delivery(api, bundle: Path, receipt: dict, token: str, *, attempts: int = 45, app_token: str | None = None, visibility: str = "private") -> dict: """Check HF head, every file identity and the exact served manifest.""" if type(attempts) is not int or attempts < 1: raise ValueError("At least one live manifest check is required") verified = verify_repository(api, bundle, receipt, visibility=visibility) host = _app_host(api.space_info(SPACE).host) expected = (bundle / "hub-manifest.json").read_bytes() if visibility == "protected": manifest_url = host + "/hub-manifest.json?protected-check=" + str(time.time_ns()) headers = {} else: signed = app_token if app_token is not None else _space_jwt(token) if not isinstance(signed, str) or not re.fullmatch(r"[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+", signed): raise ValueError("Invalid private App access token") manifest_url = host + "/hub-manifest.json?" + urlencode({"__sign": signed}) headers = {"Cache-Control": "no-cache"} last_response = "not checked" for attempt in range(attempts): if visibility == "protected": manifest_url = host + "/hub-manifest.json?protected-check=" + str(time.time_ns()) try: served = _read(manifest_url, headers) if served == expected: break last_response = "different manifest" except HTTPError as error: last_response = "HTTP " + str(error.code) except URLError: last_response = "transport error" if attempt + 1 == attempts: raise ValueError("App has not served the expected website manifest (" + last_response + ")") time.sleep(2) if visibility == "protected": verify_protected(api, host) return verified | {"served_manifest_matches": True, "anonymous_app_access": True, "anonymous_repository_access_denied": True} verify_privacy(host) return verified | {"served_manifest_matches": True, "anonymous_access_denied": True} def main() -> None: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) for name in ("build", "publish", "verify-app"): command = subparsers.add_parser(name) command.add_argument("--expected-sha", required=True) command.add_argument("--visibility", choices=("private", "protected"), default="private") command.add_argument("--repo", type=Path, default=Path.cwd()) if name == "build": command.add_argument("--output", type=Path, required=True) else: command.add_argument("--bundle", type=Path, required=True) command.add_argument("--receipt", type=Path, required=True) if name == "publish": command.add_argument("--verification", choices=("app", "repository"), default="app", help="CI verifies repository bytes/privacy; App access needs a private account") args = parser.parse_args() token = app_token = None try: eligibility = {} if args.command == "verify-app" else _gate(args.expected_sha) _source(args.repo, args.expected_sha) if args.command == "build": manifest = (build(args.repo, args.output) if args.visibility == "private" else build(args.repo, args.output, visibility=args.visibility)) if manifest["uncommitted_source_paths"]: raise ValueError("Deployment requires a clean source snapshot") print(json.dumps(eligibility | {"built": True, "private_target": SPACE, "visibility": args.visibility})) return verify_bundle(args.repo, args.bundle, args.expected_sha, args.visibility) if args.command == "publish" and os.environ.get("HF_OIDC_RESOURCE") != RESOURCE: raise ValueError("Deployment requires the approved single-Space OIDC resource") from huggingface_hub import HfApi, get_token if args.command == "verify-app": token = get_token() if not token: raise ValueError("App verification requires an authenticated private account") api = HfApi(token=token) receipt = {"commit": api.space_info(SPACE).sha, "source_commit": args.expected_sha, "verification": ("anonymous-protected-app" if args.visibility == "protected" else "authenticated-app")} receipt |= verify_delivery(api, args.bundle, receipt, token, visibility=args.visibility) else: token = get_token() if not token or not token.startswith("hf_jwt_"): raise ValueError("Deployment requires a short-lived repository publisher token") api = HfApi(token=token) destination = api.space_info(SPACE) if not _matching_space(destination, args.visibility): raise ValueError("Deployment target must remain the existing " + args.visibility + " Static Space") if args.visibility == "protected": verify_protected(api, destination.host) else: verify_privacy(destination.host) if args.verification == "app" and args.visibility == "private": app_token = _space_jwt(token) # Recheck after building/exchanging credentials, immediately before writing. _gate(args.expected_sha) options = {"visibility": "protected"} if args.visibility == "protected" else {} receipt = publish(args.bundle, SPACE, api=api, publisher_token=token, mirror=True, **options) if args.visibility == "protected": evidence = verify_delivery(api, args.bundle, receipt, token, visibility="protected") evidence |= {"verification": "repository-and-anonymous-protected-app", "app_verified_by_ci": True} elif args.verification == "app": evidence = verify_delivery(api, args.bundle, receipt, token, app_token=app_token) evidence["verification"] = "authenticated-app" else: evidence = verify_repository(api, args.bundle, receipt) verify_privacy(api.space_info(SPACE).host) evidence |= {"anonymous_access_denied": True, "served_manifest_matches": None, "verification": "repository-and-anonymous-access", "app_verified_by_ci": False} receipt |= eligibility | evidence args.receipt.parent.mkdir(parents=True, exist_ok=True) args.receipt.write_text(json.dumps(receipt, indent=2) + "\n") print(json.dumps(receipt)) except Exception as error: # HTTP/SDK exception strings can contain request details; do not print # them from the credential-bearing process or upload a traceback. message = str(error) if isinstance(error, ValueError) else type(error).__name__ for credential in (token, app_token): if credential: message = message.replace(credential, "[redacted]") for name in ("GITHUB_TOKEN", "HF_TOKEN", "HF_OIDC_ID_TOKEN", "ACTIONS_ID_TOKEN_REQUEST_TOKEN"): value = os.environ.get(name) if value: message = message.replace(value, "[redacted]") raise SystemExit("Website deployment failed: " + message) from None if __name__ == "__main__": main()