"""Publish the verified AUREOLE-R release. Default: offline integrity check. The token stays in process memory; no login, credential file, shell command, or token argument is used. SDK calls permit an explicit parent-commit guard and verification of the public, immutable result. No existing file is deleted. """ from __future__ import annotations import argparse import getpass import hashlib import json import os from pathlib import Path, PurePosixPath import re import shutil import sys import tempfile import warnings from datetime import datetime, timezone ROOT = Path(__file__).resolve().parents[1] ENDPOINT = "https://huggingface.co" DEFAULT_REPO = "PureOne/AUREOLE-R-v3" EDITION = "3.0.0-hf.1" class ReleaseError(Exception): """Safe-to-display error containing no credentials or HTTP payloads.""" def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def safe_name(name: str) -> bool: parts = PurePosixPath(name).parts return (bool(parts) and not name.startswith("/") and not any(part in (".", "..", "") for part in name.split("/")) and not any(c in name for c in "\\:\x00\r\n") and not any(part.endswith((" ", ".")) for part in parts) and not any(part.casefold() in {".env", ".git", ".venv", ".venv-publish"} or part.casefold().endswith(".token") for part in parts) and name not in {"publication_receipt.json", "CHECKSUMS.sha256"}) def release_files(root: Path = ROOT) -> dict[str, Path]: root = root.resolve() manifest = root / "CHECKSUMS.sha256" if manifest.is_symlink() or not manifest.is_file(): raise ReleaseError("CHECKSUMS.sha256 is missing or is a symbolic link.") files: dict[str, Path] = {} seen = set() for line in manifest.read_text(encoding="utf-8").splitlines(): match = re.fullmatch(r"([0-9a-f]{64}) (.+)", line) if not match: raise ReleaseError("Malformed release checksum entry.") expected, name = match.groups() if not safe_name(name) or name.casefold() in seen: raise ReleaseError("Unsafe or duplicate release path.") seen.add(name.casefold()) path = root / name if (not path.is_file() or not path.resolve().is_relative_to(root) or any(p.is_symlink() for p in [path, *path.parents] if p != root)): raise ReleaseError("A release file is missing or uses a symbolic link.") if sha256(path) != expected: raise ReleaseError(f"Checksum mismatch: {name}. Extract a fresh release ZIP.") files[name] = path if not files: raise ReleaseError("The release checksum manifest is empty.") files["CHECKSUMS.sha256"] = manifest return files def copy_release(files: dict[str, Path], stage: Path) -> dict[str, Path]: for name, source in files.items(): dest = stage / name dest.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(source, dest) return release_files(stage) def remote_matches(sibling, path: Path) -> bool: if sibling.size != path.stat().st_size: return False lfs = getattr(sibling, "lfs", None) if lfs is not None: digest = lfs.get("sha256") if isinstance(lfs, dict) else lfs.sha256 return digest == sha256(path) payload = path.read_bytes() git_blob = hashlib.sha1(b"blob " + str(len(payload)).encode() + b"\0" + payload).hexdigest() return getattr(sibling, "blob_id", None) == git_blob def inspect_remote(info, files: dict[str, Path], *, complete: bool) -> set[str]: if info.private or getattr(info, "gated", False) not in (False, None): raise ReleaseError("The destination is private or gated; its visibility was not changed.") present = set() for sibling in info.siblings or []: name = sibling.rfilename if name == ".gitattributes": # Server-managed storage configuration. continue if name not in files: raise ReleaseError("The destination contains unrelated files. Choose a new repository with --repo.") if not remote_matches(sibling, files[name]): raise ReleaseError("The destination contains different release bytes. No overwrite was attempted; choose a new repository with --repo.") present.add(name) if complete and present != set(files): raise ReleaseError("The public commit is missing release files. Rerun this same package to resume.") return present def check_owner(identity: dict, repo: str) -> None: if identity.get("name", "").casefold() != repo.split("/", 1)[0].casefold(): raise ReleaseError("This token belongs to a different account than the repository owner. Use your own account token or --repo YOUR_ACCOUNT/NAME.") def publish(files, repo, api, public_api, not_found, add_operation, download, work: Path): """Explicit dependencies allow testing the complete control flow offline.""" identity = api.whoami() check_owner(identity, repo) try: before = api.repo_info(repo, repo_type="model", files_metadata=True) except not_found: api.create_repo(repo_id=repo, repo_type="model", private=False, exist_ok=False) before = api.repo_info(repo, repo_type="model", files_metadata=True) present = inspect_remote(before, files, complete=False) api.auth_check(repo_id=repo, repo_type="model", write=True) if present != set(files): operations = [add_operation(path_in_repo=name, path_or_fileobj=str(path)) for name, path in sorted(files.items()) if name not in present] result = api.create_commit( repo_id=repo, repo_type="model", operations=operations, commit_message=f"AUREOLE-R {EDITION}: standalone public research release", commit_description="Verified paper, source, prior, raw evidence, protocols and machine-readable research indexes. Finite CPU reference; full unified SR/RR/FG remains unvalidated.", parent_commit=before.sha, ) revision = result.oid else: revision = before.sha public = public_api.repo_info(repo, repo_type="model", revision=revision, files_metadata=True) if public.sha != revision: raise ReleaseError("The public server did not return the uploaded commit.") inspect_remote(public, files, complete=True) remote_manifest = Path(download( repo_id=repo, filename="CHECKSUMS.sha256", repo_type="model", revision=revision, token=False, endpoint=ENDPOINT, cache_dir=str(work / "public-verification"), force_download=True, )) if sha256(remote_manifest) != sha256(files["CHECKSUMS.sha256"]): raise ReleaseError("The anonymously downloaded manifest did not match the release.") return { "schema_version": "1.0", "publication_verified": True, "verified_utc": datetime.now(timezone.utc).isoformat(), "repo_id": repo, "repo_type": "model", "revision": revision, "repository_url": f"{ENDPOINT}/{repo}", "immutable_url": f"{ENDPOINT}/{repo}/tree/{revision}", "paper_url": f"{ENDPOINT}/{repo}/resolve/{revision}/AUREOLE_R_v3.0.0_Certified_Innovation_Rendering.pdf", "scientific_version": "3.0.0", "publication_edition": EDITION, "verified_release_files": len(files), "checksum_manifest_sha256": sha256(files["CHECKSUMS.sha256"]), "verification": "Anonymous commit read, size and Git-blob/LFS-hash verification for every release file, and anonymous SHA-256 manifest download.", "scientific_scope": "Finite direct-light CPU reference. No GPU or full SR/RR/FG validation.", } def main(argv=None, root: Path = ROOT) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo", default=DEFAULT_REPO, help="Personal Hugging Face account/repository") parser.add_argument("--publish", action="store_true", help="Create or safely resume this PUBLIC research release") args = parser.parse_args(argv) if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9][A-Za-z0-9._-]*", args.repo): raise ReleaseError("Repository must be ACCOUNT/NAME.") files = release_files(root) print(f"PUBLIC destination: {ENDPOINT}/{args.repo}") print(f"Verified {len(files)} release files ({sum(p.stat().st_size for p in files.values()):,} bytes).") if not args.publish: print("Offline check complete. No authentication or network calls. Add --publish to upload.") return 0 os.environ["HF_DEBUG"] = "0" os.environ["HF_HUB_VERBOSITY"] = "warning" os.environ["HF_HUB_DISABLE_TELEMETRY"] = "1" try: from huggingface_hub import HfApi, CommitOperationAdd, hf_hub_download from huggingface_hub.errors import RepositoryNotFoundError except ImportError: raise ReleaseError("Install requirements-publish.txt, or use PUBLISH_TO_HUGGINGFACE.bat.") from None with tempfile.TemporaryDirectory(prefix="aureole-publication-") as temp: work = Path(temp) staged = copy_release(files, work / "release") with warnings.catch_warnings(): warnings.simplefilter("error", getpass.GetPassWarning) token = os.environ.get("HF_TOKEN") or getpass.getpass("Hugging Face write token (paste; input is hidden): ") token = token.strip() if not token: raise ReleaseError("No token supplied. Nothing was uploaded.") print("Checking account and public destination, then uploading the verified release...") receipt = publish( staged, args.repo, HfApi(endpoint=ENDPOINT, token=token), HfApi(endpoint=ENDPOINT, token=False), RepositoryNotFoundError, CommitOperationAdd, hf_hub_download, work, ) del token receipt_path = root / "publication_receipt.json" receipt_path.write_text(json.dumps(receipt, indent=2) + "\n", encoding="utf-8") print("PUBLICATION VERIFIED: " + receipt["repository_url"]) print("Immutable research release: " + receipt["immutable_url"]) print("Receipt saved to publication_receipt.json (contains no token).") return 0 if __name__ == "__main__": try: raise SystemExit(main()) except (KeyboardInterrupt, EOFError, getpass.GetPassWarning): print("Cancelled, or secure token input unavailable. Use the Windows launcher in a terminal.", file=sys.stderr) raise SystemExit(1) except ReleaseError as exc: print(str(exc), file=sys.stderr) raise SystemExit(1) except Exception as exc: status = getattr(getattr(exc, "response", None), "status_code", None) suffix = f" HTTP status: {status}." if isinstance(status, int) else "" print("Publication was not verified." + suffix + " Check your connection and write-token permissions, then rerun the same package. A remote commit may already exist; matching bytes are safe to resume. No exception payload or token was logged.", file=sys.stderr) raise SystemExit(1)