File size: 11,209 Bytes
9d6c005 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 | """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)
|