"""Isolated, user-token-only Hugging Face Bucket operations for the API.""" from __future__ import annotations import base64 import csv from datetime import datetime, timezone import hashlib import io import json import os from pathlib import Path import re import subprocess import sys import tempfile import uuid from portable_bundle import ( BundleError, MAX_BUNDLE, MAX_NETWORK, MAX_ANALYSIS_ARCHIVE, MAX_ANALYSIS_FILE_BYTES, load_bundle, validate_freeze, ) from worker import network_input HF_ENDPOINT = "https://huggingface.co" MAX_REQUEST = 2 * MAX_BUNDLE + 64 * 1024 MAX_RECIPE_LIBRARY = 32 * 1024 MAX_CHILD_INPUT = 4 * ((MAX_BUNDLE + 2) // 3) + 64 * 1024 MAX_CHILD_OUTPUT = 128 * 1024 CHILD_TIMEOUT_SECONDS = 55 MAX_LIST_SCAN = 10_000 MAX_LIST_RESULTS = 50 LEGACY_APP_KEY = re.compile(r"^ald-exports/([0-9a-f]{32}|[0-9a-f]{64})/bundle\.json$") RECIPE_APP_KEY = re.compile(r"^ald-recipes/(?P[0-9]{8}T[0-9]{6}Z)-(?P[0-9a-f]{32})/library\.json$") AUTOSAVE_APP_KEY = re.compile( r"^ald-exports/(?P[0-9]{8}T[0-9]{6}Z)-" r"(?P[a-z0-9][a-z0-9-]{0,63})-" r"(?Pformula|rally|snowmobile|truck)-" r"g(?P0|[1-9][0-9]{0,3})-" r"lap(?Pnone|[0-9]{1,16}ms)-(?P[0-9a-f]{32})/bundle\.json$" ) VEHICLE_NAMES = ("Formula", "Rally", "Snowmobile", "Truck") MAX_SIDECAR_TOTAL = 320 * 1024 * 1024 BUCKET_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}/[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$") class BucketStorageError(Exception): """Safe error category; never carries upstream text or credentials.""" def __init__(self, category: str): self.category = category super().__init__(category) def valid_bucket_id(value: object) -> bool: return isinstance(value, str) and bool(BUCKET_ID.fullmatch(value)) def valid_bundle_key(value: object) -> bool: return isinstance(value, str) and bool(LEGACY_APP_KEY.fullmatch(value) or AUTOSAVE_APP_KEY.fullmatch(value)) def valid_recipe_key(value: object) -> bool: if not isinstance(value, str): return False match = RECIPE_APP_KEY.fullmatch(value) if match is None: return False try: datetime.strptime(match.group("timestamp"), "%Y%m%dT%H%M%SZ") except ValueError: return False return True RECIPE_DEFAULT_SETTINGS = { "population": 4, "generations": 1, "ticks": 60, "mutation": 0.01, "seed": 42, "mutation_end": 0.01, "schedule": "fixed", "selection": "best", "parents": 1, "elite": 1, "crossover": "none", "decay": 0, "adaptive_mutation": False, "control_hz": 60, "workers": 1, "collision": False, "idle": False, "rewards": [{"metric": "total_score", "type": "default", "weight": 100}], } RECIPE_METRICS = { "total_score", "total_speed", "total_abs_steering", "total_actual_distance", "total_drifting_amount", "total_momentum_change_frequency", "total_throttle_change_frequency", "total_braking_frequency", "total_distance_from_wall", "total_distance_from_center", "network_novelty", "collision_count", "best_lap_performance", "total_racing_line_efficiency", } def validate_recipe_library(raw: bytes) -> dict: """Validate the existing browser RecipeLibrary v1 format, without storing other app data.""" if not isinstance(raw, bytes) or len(raw) > MAX_RECIPE_LIBRARY: raise ValueError("Invalid recipe library") def unique_object(pairs): value = {} for key, item in pairs: if key in value: raise ValueError("Duplicate recipe field") value[key] = item return value try: book = json.loads(raw.decode("utf-8"), object_pairs_hook=unique_object, parse_constant=lambda _value: (_ for _ in ()).throw(ValueError("Invalid number"))) except Exception: raise ValueError("Invalid recipe library") from None if not isinstance(book, dict) or set(book) not in ({"version", "recipes"}, {"version", "recipes", "batches"}): raise ValueError("Invalid recipe library") if type(book.get("version")) is not int or book["version"] != 1: raise ValueError("Invalid recipe library") recipes = book.get("recipes") if not isinstance(recipes, list) or len(recipes) > 10: raise ValueError("Invalid recipe library") seen = set() for recipe in recipes: if not isinstance(recipe, dict) or set(recipe) != {"slot", "name", "settings"}: raise ValueError("Invalid recipe library") slot, name = recipe["slot"], recipe["name"] if not isinstance(slot, str) or not re.fullmatch(r"[A-J]", slot) or slot in seen: raise ValueError("Invalid recipe library") if not isinstance(name, str) or not name.strip() or len(name) > 80: raise ValueError("Invalid recipe library") seen.add(slot) settings = recipe["settings"] if (not isinstance(settings, dict) or set(settings) - (set(RECIPE_DEFAULT_SETTINGS) | {"freeze"})): raise ValueError("Invalid recipe library") if "freeze" in settings: validate_freeze(settings["freeze"]) clean = {**RECIPE_DEFAULT_SETTINGS, **settings} numeric = { "population": (1, 1024, True), "generations": (1, 2000, True), "ticks": (1, 18000, True), "mutation": (0, 4.2, False), "mutation_end": (0, 4.2, False), "seed": (0, 2147483647, True), "parents": (1, 1024, True), "elite": (0, 1024, True), "decay": (0, 1, False), "workers": (1, 4, True), } for key, (low, high, integer) in numeric.items(): value = clean[key] if (type(value) not in (int, float) or not low <= value <= high or (integer and not float(value).is_integer())): raise ValueError("Invalid recipe library") enums = {"schedule": {"fixed", "linear", "geometric"}, "selection": {"best", "tournament", "roulette"}, "crossover": {"none", "single_point", "uniform"}, "control_hz": {60, 30, 15}} for key, choices in enums.items(): if type(clean[key]) not in (str, int, float) or clean[key] not in choices: raise ValueError("Invalid recipe library") for key in ("adaptive_mutation", "collision", "idle"): if type(clean[key]) is not bool: raise ValueError("Invalid recipe library") if clean["parents"] > clean["population"] or clean["elite"] > clean["population"] or clean["mutation"] <= 0: raise ValueError("Invalid recipe library") if clean["schedule"] != "fixed" and clean["mutation_end"] <= 0: raise ValueError("Invalid recipe library") rewards = clean["rewards"] if not isinstance(rewards, list) or not 1 <= len(rewards) <= len(RECIPE_METRICS): raise ValueError("Invalid recipe library") reward_metrics = set() for reward in rewards: if not isinstance(reward, dict) or set(reward) != {"metric", "type", "weight"}: raise ValueError("Invalid recipe library") metric, reward_type, weight = reward["metric"], reward["type"], reward["weight"] if (not isinstance(metric, str) or metric not in RECIPE_METRICS or not isinstance(reward_type, str) or reward_type not in {"default", "average"} or metric in reward_metrics or type(weight) not in (int, float) or not -100 <= weight <= 100 or not float(weight).is_integer()): raise ValueError("Invalid recipe library") reward_metrics.add(metric) if not any(reward["weight"] for reward in rewards): raise ValueError("Invalid recipe library") batches = book.get("batches", []) if not isinstance(batches, list) or len(batches) > 10: raise ValueError("Invalid recipe library") batch_slots = set() for batch in batches: if not isinstance(batch, dict) or set(batch) != {"slot", "name", "carry", "runs"}: raise ValueError("Invalid recipe library") slot, name, runs = batch["slot"], batch["name"], batch["runs"] if (not isinstance(slot, str) or not re.fullmatch(r"[A-J]", slot) or slot in batch_slots or not isinstance(name, str) or not name.strip() or len(name) > 80 or type(batch["carry"]) is not bool or not isinstance(runs, list) or not 1 <= len(runs) <= 100 or any(not isinstance(run, str) or run not in seen for run in runs)): raise ValueError("Invalid recipe library") batch_slots.add(slot) return book def _recipe_list_metadata(key: str, uploaded: object) -> tuple[str, str | None]: match = RECIPE_APP_KEY.fullmatch(key) if match is None: raise BucketStorageError("invalid_request") try: path_when = datetime.strptime(match.group("timestamp"), "%Y%m%dT%H%M%SZ").replace(tzinfo=timezone.utc) except ValueError: path_when = None when = _valid_utc_datetime(uploaded) or path_when if when is None: return f"Saved recipe library · {match.group('uuid')[:8]}", None created_at = when.isoformat(timespec="microseconds").replace("+00:00", "Z") return f"Recipe library · {when.strftime('%Y-%m-%d %H:%M UTC')} · {match.group('uuid')[:8]}", created_at def _autosave_prefix(bundle: dict, files: dict[str, bytes], now: datetime | None = None) -> tuple[str, str, str]: artifacts = bundle["artifacts"] receipt = artifacts.get("run_receipt") track = artifacts.get("selected_track") track_id = _track_path_slug(track) if receipt is not None: report = receipt["report"] vehicle_type = report["vehicle_type"] generations = report["generations_completed"] lap = report["best_lap_seconds"] else: try: vehicle_type = network_input(files["selected_network.evolution"])["Type"] except Exception: raise BucketStorageError("invalid_bundle") from None generations = 0 lap = None vehicle = VEHICLE_NAMES[vehicle_type] timestamp = (now or datetime.now(timezone.utc)).astimezone(timezone.utc).strftime("%Y%m%dT%H%M%SZ") lap_part = "none" if lap is None else f"{int(round(lap * 1000))}ms" segment = f"{timestamp}-{track_id}-{vehicle.lower()}-g{generations}-lap{lap_part}-{uuid.uuid4().hex}" return f"ald-exports/{segment}", track_id, timestamp def _readable_track(track_id: str, track: dict | None) -> str: if isinstance(track, dict) and track.get("source") == "builtin": return _track_path_slug(track) if track_id == "custom": return "custom-track" return "custom-track" def winner_sidecar_filename(track_slug: str, vehicle: str, timestamp: str, unique: str, role: str) -> str: role_label = {"fitness": "fitness-winner", "fastest_lap": "fastest-lap-winner", "clean_laps": "clean-laps-winner"}.get(role) if (not role_label or not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,63}", track_slug) or vehicle not in {"formula", "rally", "snowmobile", "truck"} or not re.fullmatch(r"[0-9]{8}T[0-9]{6}Z", timestamp) or not re.fullmatch(r"[0-9a-f]{8}", unique)): raise BucketStorageError("invalid_bundle") return f"{track_slug}-{vehicle}-{timestamp}-{unique}-{role_label}.evolution" def _track_path_slug(track: dict | None) -> str: if isinstance(track, dict) and track.get("source") == "builtin": display_name = track.get("display_name") if isinstance(display_name, str): slug = re.sub(r"[^A-Za-z0-9]+", "-", display_name).strip("-").lower()[:64].strip("-") if slug and re.fullmatch(r"[a-z0-9][a-z0-9-]{0,63}", slug): return slug track_id = track.get("track_id") if isinstance(track_id, str): fragment = re.sub(r"[^A-Za-z0-9]+", "-", track_id).strip("-").lower()[:48].strip("-") if fragment: return "track-" + fragment return "custom-track" def _render_history_csv(report: dict) -> bytes: stream = io.StringIO(newline="") writer = csv.writer(stream, lineterminator="\n") fields = ("generation", "mutation", "best_score", "mean_score", "best_lap_seconds", "lapped_cars", "simulated_ticks") writer.writerow(fields) for row in report.get("history", []): writer.writerow(["" if row[field] is None else row[field] for field in fields]) return stream.getvalue().encode("utf-8") def _render_report(bundle: dict) -> bytes: artifacts = bundle["artifacts"] receipt = artifacts["run_receipt"] report = receipt["report"] track = artifacts.get("selected_track") summary = report.get("analysis_summary") evaluation = isinstance(summary, dict) and summary.get("mode") == "evaluate" track_label = track.get("display_name", "Built-in track") if isinstance(track, dict) and track.get("source") == "builtin" else "Custom or uploaded track" lap = report["best_lap_seconds"] lap_label = "No completed lap" if lap is None else f"{lap:.3f} s" lines = [ "# ALD frozen evaluation" if evaluation else "# ALD training run", "", f"- Track: {track_label}", f"- Vehicle: {report['vehicle_name']}", f"- Backend: {receipt['backend']} ({report['execution_backend']})", f"- Build: {receipt['build_id']}", f"- Generations: {report['generations_completed']} completed of {report['generations_requested']} requested", f"- Population: {report['population']}", f"- Final-generation best score: {report['best_score']}", f"- Final-generation mean score: {report['mean_score']}", f"- Best completed lap observed during this run: {lap_label}", f"- Lapped cars in the final generation: {report['lapped_cars']}", f"- Simulated ticks: {report['simulated_ticks']} of {report['requested_ticks']} requested per car", f"- Elapsed seconds: {report['elapsed_seconds']:.3f}", f"- Exported network SHA-256: {receipt['output_sha256']}", "", "## Validated run settings", "", ] for key in sorted(report["settings"]): lines.append(f"- {key}: `{json.dumps(report['settings'][key], ensure_ascii=True, allow_nan=False, separators=(',', ':'))}`") if isinstance(summary, dict): lines.extend(["", "## Candidate analysis", "", f"- Mode: {summary['mode']}"]) if evaluation: lines.append("- This is a frozen evaluation; the input network bytes and weights were preserved.") else: for role in ("fitness", "fastest_lap", "clean_laps"): winner = summary["winners"][role] if winner["status"] == "available": lines.append(f"- {role.replace('_', ' ').title()}: generation {winner['generation']}, car {winner['car']}, candidate `{winner['candidate_id']}`") else: lines.append(f"- {role.replace('_', ' ').title()}: unavailable ({winner['reason']})") lines.append("- A run-wide lap winner may be a different car from the exported fitness candidate.") elif not evaluation: lines.append("- The best completed lap may belong to a different car than the exported candidate.") return ("\n".join(lines) + "\n").encode("utf-8") def _save_files(bundle_bytes: bytes, bundle: dict, files: dict[str, bytes], *, now: datetime | None = None) -> tuple[str, dict[str, bytes]]: prefix, track, timestamp = _autosave_prefix(bundle, files, now) artifacts = bundle["artifacts"] receipt = artifacts.get("run_receipt") if receipt is not None: report = receipt["report"] vehicle = report["vehicle_name"].lower() else: vehicle = VEHICLE_NAMES[network_input(files["selected_network.evolution"])["Type"]].lower() uploads = {"bundle.json": bundle_bytes} network_suffix = prefix.rsplit("-", 1)[-1][:8] readable_track = _readable_track(track, artifacts.get('selected_track')) uploads[f"{readable_track}-{vehicle}-{timestamp}-{network_suffix}.evolution"] = files["selected_network.evolution"] if receipt is not None: uploads["report.md"] = _render_report(bundle) uploads["generation-history.csv"] = _render_history_csv(receipt["report"]) if bundle.get("version") == 2: if "analysis_archive.gz" not in files: raise BucketStorageError("invalid_bundle") uploads["analysis-archive.gz"] = files["analysis_archive.gz"] for name, content in files.items(): if name.startswith("analysis/"): uploads[name] = content if "analysis/stars.json" in files: stars=json.loads(files["analysis/stars.json"]) # Plain text avoids interpreting user-provided track labels as markup. lines=["Star challenge check for the run's primary network", stars['basis'], "This check grants no game stars and keeps no campaign progression.", "Policy: "+stars['network_sha256'], "Check duration: "+str(stars['horizon_seconds'])+" seconds", ""] for challenge in stars['challenges']: lines.extend([challenge['text'], "Result: "+challenge['status'].replace('_',' '), "Reason: "+(challenge['reason'] or 'Condition observed.'), ""]) uploads['star-check.txt']=("\n".join(lines)+"\n").encode('utf-8') for role in ("fitness", "fastest_lap", "clean_laps"): name = f"analysis/winners/{role}.evolution" if name in files: uploads[winner_sidecar_filename(readable_track, vehicle, timestamp, network_suffix, role)] = files[name] if sum(map(len, uploads.values())) > MAX_BUNDLE + MAX_NETWORK + MAX_SIDECAR_TOTAL: raise BucketStorageError("invalid_bundle") return prefix, uploads def _list_metadata(key: str, uploaded: datetime | None) -> tuple[str, str | None]: match = AUTOSAVE_APP_KEY.fullmatch(key) if match: try: path_when = datetime.strptime(match.group("timestamp"), "%Y%m%dT%H%M%SZ").replace(tzinfo=timezone.utc) except ValueError: path_when = None when = _valid_utc_datetime(uploaded) or path_when if when is None: created_at = None date_label = "Saved bundle" else: # Fixed fractional precision keeps lexicographic ordering chronological # when multiple uploads occur within the same generated-key second. created_at = when.isoformat(timespec="microseconds").replace("+00:00", "Z") date_label = when.strftime("%Y-%m-%d %H:%M UTC") lap = match.group("lap") lap_label = "no completed lap" if lap == "none" else f"best run lap {int(lap[:-2])} ms" track = "Custom track" if match.group("track") == "custom" else match.group("track").replace("-", " ").title() label = f"{date_label} · {track} · {match.group('vehicle').title()} · {match.group('generations')} generations · {lap_label}" return label[:192], created_at date_label = "Saved bundle" created_at = None uploaded = _valid_utc_datetime(uploaded) if uploaded is not None: date_label = uploaded.strftime("%Y-%m-%d %H:%M UTC") created_at = uploaded.isoformat(timespec="microseconds").replace("+00:00", "Z") legacy = LEGACY_APP_KEY.fullmatch(key) label = f"{date_label} · imported portable network {legacy.group(1)[:8] if legacy else ''}".strip() return label[:192], created_at def _valid_utc_datetime(value: object) -> datetime | None: if not isinstance(value, datetime): return None try: if value.tzinfo is None: value = value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) except (OverflowError, ValueError): return None def _child_environment(cache_root: Path) -> dict[str, str]: # Do not inherit host HF tokens, endpoints, cache locations, or login state. env = { "PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"), "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "TMPDIR": str(cache_root), "HF_HOME": str(cache_root / "hf"), "HF_HUB_CACHE": str(cache_root / "hub-cache"), "HF_XET_CACHE": str(cache_root / "xet-cache"), "HF_XET_LOG_DEST": str(cache_root / "xet.log"), "HF_HUB_DISABLE_TELEMETRY": "1", "HF_HUB_DISABLE_IMPLICIT_TOKEN": "1", "HF_HUB_DISABLE_PROGRESS_BARS": "1", "PYTHONNOUSERSITE": "1", "ALD_BUCKET_CHILD": "1", "ALD_BUCKET_TEMP": str(cache_root), } return env def run_bucket_operation(operation: str, token: str, bucket_id: str, **args): """Run one HF operation in an isolated process with a temporary Xet cache.""" if operation not in {"connect", "list", "save", "restore", "recipes_list", "recipes_save", "recipes_restore"}: raise BucketStorageError("invalid_request") if not isinstance(token, str) or not token or len(token) > 4096 or any(ch.isspace() or ord(ch) < 33 for ch in token): raise BucketStorageError("unauthorized") if not valid_bucket_id(bucket_id): raise BucketStorageError("invalid_request") try: with tempfile.TemporaryDirectory(prefix="ald-bucket-") as temp_name: temp = Path(temp_name) payload = {"operation": operation, "token": token, "bucket_id": bucket_id, **args} raw = json.dumps(payload, separators=(",", ":")).encode("utf-8") if len(raw) > MAX_CHILD_INPUT: raise BucketStorageError("invalid_request") proc = subprocess.run( [sys.executable, str(Path(__file__).resolve()), "--child"], input=raw, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=CHILD_TIMEOUT_SECONDS, check=False, shell=False, env=_child_environment(temp), ) if len(proc.stdout) > MAX_CHILD_OUTPUT: raise BucketStorageError("storage_unavailable") try: result = json.loads(proc.stdout.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): raise BucketStorageError("storage_unavailable") from None if proc.returncode != 0 or not isinstance(result, dict): category = result.get("error") if isinstance(result, dict) else None if category in {"unauthorized", "forbidden", "not_found", "invalid_request", "storage_unavailable"}: raise BucketStorageError(category) raise BucketStorageError("storage_unavailable") if operation in {"save", "restore", "recipes_save", "recipes_restore"}: is_recipe = operation.startswith("recipes_") is_save = operation in {"save", "recipes_save"} expected_name = "recipe-readback.json" if operation == "recipes_save" else ( "recipe-restore.json" if operation == "recipes_restore" else ( "readback.json" if is_save else "restore.json")) file_path = temp / expected_name if not file_path.is_file() or file_path.is_symlink(): raise BucketStorageError("storage_unavailable") max_bytes = MAX_RECIPE_LIBRARY if is_recipe else MAX_BUNDLE if file_path.stat().st_size > max_bytes: raise BucketStorageError("invalid_request") content = file_path.read_bytes() if is_save: if (hashlib.sha256(content).hexdigest() != result.get("sha256") or len(content) != result.get("bytes")): raise BucketStorageError("storage_unavailable") result["verified_bytes"] = content else: if is_recipe: try: validate_recipe_library(content) except Exception: raise BucketStorageError("invalid_request") from None result["recipe_bytes"] = content else: try: load_bundle(content) except BundleError: raise BucketStorageError("invalid_bundle") from None result["bundle_bytes"] = content return result except subprocess.TimeoutExpired: raise BucketStorageError("timeout") from None except BucketStorageError: raise except Exception: raise BucketStorageError("storage_unavailable") from None def _verify_bucket(api, token: str, bucket_id: str): try: user = api.whoami(token=token, cache=False) except Exception: raise BucketStorageError("unauthorized") from None username = user.get("name") if isinstance(user, dict) else None if not isinstance(username, str) or not username or bucket_id.split("/", 1)[0] != username: raise BucketStorageError("forbidden") try: info = api.bucket_info(bucket_id, token=token) except Exception: # Inaccessible and nonexistent buckets deliberately share one response. raise BucketStorageError("forbidden") from None if getattr(info, "id", None) != bucket_id or getattr(info, "private", None) is not True: raise BucketStorageError("forbidden") return username def _download_descriptor(api, token: str, bucket_id: str, key: str, destination: Path, *, max_bytes: int = MAX_BUNDLE): from huggingface_hub._buckets import BucketFile metadata = api.get_bucket_file_metadata(bucket_id, key, token=token) size = getattr(metadata, "size", None) xet_data = getattr(metadata, "xet_file_data", None) xet_hash = getattr(xet_data, "file_hash", None) if type(size) is not int or size < 0 or size > max_bytes or not isinstance(xet_hash, str) or not xet_hash: raise BucketStorageError("invalid_request") descriptor = BucketFile( type="file", path=key, size=size, xetHash=xet_hash, mtime=None, uploadedAt=None, ) try: api.download_bucket_files( bucket_id, [(descriptor, destination)], raise_on_missing_files=True, token=token, ) except Exception: raise BucketStorageError("storage_unavailable") from None if not destination.is_file() or destination.is_symlink() or destination.stat().st_size != size: raise BucketStorageError("storage_unavailable") return size, xet_hash def _perform_operation(request: dict, api, temp: Path) -> dict: operation = request["operation"] token = request["token"] bucket_id = request["bucket_id"] username = _verify_bucket(api, token, bucket_id) if operation == "connect": return {"status": "connected", "bucket_id": bucket_id, "username": username, "private": True} if operation == "recipes_list": items = [] scan_incomplete = False for index, item in enumerate(api.list_bucket_tree(bucket_id, prefix="ald-recipes", recursive=True, token=token)): if index >= MAX_LIST_SCAN: scan_incomplete = True break key = getattr(item, "path", None) size = getattr(item, "size", None) if valid_recipe_key(key) and type(size) is int and 0 <= size <= MAX_RECIPE_LIBRARY: uploaded = getattr(item, "uploaded_at", None) or getattr(item, "mtime", None) label, created_at = _recipe_list_metadata(key, uploaded) items.append({"key": key, "bytes": size, "label": label, "created_at": created_at}) items.sort(key=lambda item: item["created_at"] or "", reverse=True) truncated = len(items) > MAX_LIST_RESULTS return {"items": items[:MAX_LIST_RESULTS], "truncated": truncated, "scan_incomplete": scan_incomplete} if operation == "recipes_save": encoded = request.get("library_base64") if not isinstance(encoded, str) or len(encoded) > 4 * ((MAX_RECIPE_LIBRARY + 2) // 3): raise BucketStorageError("invalid_request") try: library = base64.b64decode(encoded, validate=True) validate_recipe_library(library) except Exception: raise BucketStorageError("invalid_request") from None digest = hashlib.sha256(library).hexdigest() timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") prefix = f"ald-recipes/{timestamp}-{uuid.uuid4().hex}" source = temp / "recipe-upload" source.mkdir(mode=0o700) (source / "library.json").write_bytes(library) try: api.sync_bucket(str(source), f"hf://buckets/{bucket_id}/{prefix}", delete=False, ignore_existing=True, quiet=True, token=token) except Exception: raise BucketStorageError("storage_unavailable") from None key = f"{prefix}/library.json" readback = temp / "recipe-readback.json" size, _ = _download_descriptor(api, token, bucket_id, key, readback, max_bytes=MAX_RECIPE_LIBRARY) content = readback.read_bytes() if size != len(library) or content != library or hashlib.sha256(content).hexdigest() != digest: raise BucketStorageError("storage_unavailable") return {"status": "saved_verified", "bucket_id": bucket_id, "key": key, "bytes": len(library), "sha256": digest} if operation == "recipes_restore": key = request.get("key") if not valid_recipe_key(key): raise BucketStorageError("invalid_request") restore = temp / "recipe-restore.json" size, _ = _download_descriptor(api, token, bucket_id, key, restore, max_bytes=MAX_RECIPE_LIBRARY) content = restore.read_bytes() try: validate_recipe_library(content) except Exception: raise BucketStorageError("invalid_request") from None return {"status": "restored", "key": key, "bytes": size, "sha256": hashlib.sha256(content).hexdigest()} if operation == "list": items = [] scan_incomplete = False for index, item in enumerate(api.list_bucket_tree(bucket_id, prefix="ald-exports", recursive=True, token=token)): if index >= MAX_LIST_SCAN: scan_incomplete = True break key = getattr(item, "path", None) size = getattr(item, "size", None) match = (LEGACY_APP_KEY.fullmatch(key) or AUTOSAVE_APP_KEY.fullmatch(key)) if isinstance(key, str) else None if match and type(size) is int and 0 <= size <= MAX_BUNDLE: uploaded = getattr(item, "uploaded_at", None) or getattr(item, "mtime", None) label, created_at = _list_metadata(key, uploaded) items.append({"key": key, "bytes": size, "label": label, "created_at": created_at}) # The primary ordering is newest first; unknown legacy timestamps remain at the end. items.sort(key=lambda item: item["created_at"] or "", reverse=True) truncated = len(items) > MAX_LIST_RESULTS if len(items) > MAX_LIST_RESULTS: items = items[:MAX_LIST_RESULTS] return {"items": items, "truncated": truncated, "scan_incomplete": scan_incomplete} if operation == "save": encoded = request.get("bundle_base64") if not isinstance(encoded, str) or len(encoded) > 4 * ((MAX_BUNDLE + 2) // 3): raise BucketStorageError("invalid_request") try: bundle = base64.b64decode(encoded, validate=True) bundle_obj, bundle_files = load_bundle(bundle) except Exception: raise BucketStorageError("invalid_bundle") from None digest = hashlib.sha256(bundle).hexdigest() prefix, uploads = _save_files(bundle, bundle_obj, bundle_files) key = f"{prefix}/bundle.json" source = temp / "upload" source.mkdir(mode=0o700) for filename, content in uploads.items(): if (not isinstance(filename, str) or filename.startswith("/") or any(part in {"", ".", ".."} for part in filename.split("/"))): raise BucketStorageError("invalid_bundle") path = source.joinpath(*filename.split("/")) path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) path.write_bytes(content) try: api.sync_bucket( str(source), f"hf://buckets/{bucket_id}/{prefix}", delete=False, ignore_existing=True, quiet=True, token=token, ) except Exception: raise BucketStorageError("storage_unavailable") from None verified_files = {} for index, (filename, expected) in enumerate(uploads.items()): remote_path = f"{prefix}/{filename}" readback = temp / f"readback-{index}.bin" size, _ = _download_descriptor(api, token, bucket_id, remote_path, readback, max_bytes=len(expected)) content = readback.read_bytes() expected_sha = hashlib.sha256(expected).hexdigest() if size != len(expected) or hashlib.sha256(content).hexdigest() != expected_sha: raise BucketStorageError("storage_unavailable") verified_files[filename] = {"bytes": size, "sha256": expected_sha} if verified_files.get("bundle.json", {}).get("sha256") != digest: raise BucketStorageError("storage_unavailable") (temp / "readback.json").write_bytes(bundle) return {"status": "saved_verified", "bucket_id": bucket_id, "key": key, "bytes": len(bundle), "sha256": digest, "verified_files": verified_files} key = request.get("key") if not valid_bundle_key(key): raise BucketStorageError("invalid_request") restore = temp / "restore.json" size, _ = _download_descriptor(api, token, bucket_id, key, restore) try: bundle = restore.read_bytes() load_bundle(bundle) except Exception: raise BucketStorageError("invalid_bundle") from None return {"status": "restored", "key": key, "bytes": size} def _child_main() -> int: if os.environ.get("ALD_BUCKET_CHILD") != "1": return 2 raw = sys.stdin.buffer.read(MAX_CHILD_INPUT + 1) if len(raw) > MAX_CHILD_INPUT: print(json.dumps({"error": "invalid_request"})) return 2 try: request = json.loads(raw.decode("utf-8")) if not isinstance(request, dict) or request.get("operation") not in { "connect", "list", "save", "restore", "recipes_list", "recipes_save", "recipes_restore"}: raise BucketStorageError("invalid_request") operation = request["operation"] token = request.get("token") bucket_id = request.get("bucket_id") if not isinstance(token, str) or not token or len(token) > 4096 or any(c.isspace() or ord(c) < 33 for c in token): raise BucketStorageError("unauthorized") if not valid_bucket_id(bucket_id): raise BucketStorageError("invalid_request") from huggingface_hub import HfApi api = HfApi(endpoint=HF_ENDPOINT, token=False) temp = Path(os.environ["ALD_BUCKET_TEMP"]) result = _perform_operation(request, api, temp) print(json.dumps(result, separators=(",", ":"))) return 0 except BucketStorageError as exc: print(json.dumps({"error": exc.category}, separators=(",", ":"))) return 1 except Exception: print(json.dumps({"error": "storage_unavailable"}, separators=(",", ":"))) return 1 if __name__ == "__main__" and len(sys.argv) == 2 and sys.argv[1] == "--child": raise SystemExit(_child_main())