#!/usr/bin/env python3 """ HermesFace Full Directory Persistence for Hugging Face Spaces ============================================================= Tar-gz snapshot of /opt/data with atomic upload, rotation, and integrity check. Usage: python3 hermes_persist.py save # backup python3 hermes_persist.py load # restore latest python3 hermes_persist.py status # show current backups Env vars: HF_TOKEN HF access token with write permission HERMES_DATASET_REPO Target dataset repo (e.g. username/HermesFace-data) HERMES_HOME Source directory (default: /opt/data) MAX_BACKUPS Rotation size (default: 5) """ import hashlib import json import os import shutil import sys import tarfile def _safe_extractall(tf, dest: str) -> None: """P9-T06: reject path traversal during archive extraction.""" import os dest_path = os.path.abspath(dest) for member in tf.getmembers(): member_path = os.path.abspath(os.path.join(dest_path, member.name)) if not member_path.startswith(dest_path + os.sep) and member_path != dest_path: raise ValueError(f"unsafe archive member path: {member.name!r}") tf.extractall(dest) import tempfile from datetime import datetime from pathlib import Path from typing import Optional from huggingface_hub import HfApi, hf_hub_download from huggingface_hub.utils import RepositoryNotFoundError try: from secure_backup import build_backup, restore_backup except ImportError: # pragma: no cover from scripts.secure_backup import build_backup, restore_backup class Config: HERMES_HOME = Path(os.environ.get("HERMES_HOME", "/opt/data")) DATASET_REPO = os.environ.get("HERMES_DATASET_REPO", "") HF_TOKEN = os.environ.get("HF_TOKEN", "") MAX_BACKUPS = int(os.environ.get("MAX_BACKUPS", "5")) BACKUP_PREFIX = "backup-" EXCLUDE_SUFFIXES = (".lock", ".tmp", ".pyc", ".socket", ".pid") EXCLUDE_NAMES = {"__pycache__", "node_modules", ".DS_Store", ".git"} SKIP_DIRS = {".cache", "logs/sys_logs"} def _log(level: str, msg: str, **kv: object) -> None: payload = {"timestamp": datetime.now().isoformat(), "level": level, "module": "hermes-persist", "message": msg, **kv} print(json.dumps(payload)) def _tar_filter(info: tarfile.TarInfo) -> Optional[tarfile.TarInfo]: name = info.name if any(name.endswith(suf) for suf in Config.EXCLUDE_SUFFIXES): return None parts = set(Path(name).parts) if parts & Config.EXCLUDE_NAMES: return None rel = name.lstrip("./") for skip in Config.SKIP_DIRS: if rel == skip or rel.startswith(skip + "/"): return None return info def _api() -> HfApi: if not Config.HF_TOKEN: raise RuntimeError("HF_TOKEN not set") if not Config.DATASET_REPO: raise RuntimeError("HERMES_DATASET_REPO not set") return HfApi(token=Config.HF_TOKEN) def _ensure_repo(api: HfApi) -> None: try: api.repo_info(repo_id=Config.DATASET_REPO, repo_type="dataset") except RepositoryNotFoundError: _log("INFO", "creating_dataset", repo=Config.DATASET_REPO) api.create_repo(repo_id=Config.DATASET_REPO, repo_type="dataset", private=True) def save() -> int: if not Config.HERMES_HOME.exists(): _log("ERROR", "no_source_dir", path=str(Config.HERMES_HOME)) return 1 api = _api() _ensure_repo(api) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") archive_name = f"state/{Config.BACKUP_PREFIX}{timestamp}.tar.gz" with tempfile.TemporaryDirectory() as tmp: tar_path = Path(tmp) / "hermes.tar.gz" backup_manifest = build_backup(Config.HERMES_HOME, tar_path) size = tar_path.stat().st_size sha256 = str(backup_manifest["archive_sha256"]) _log("INFO", "uploading_backup", file=archive_name, bytes=size, sha256=sha256[:12]) api.upload_file( path_or_fileobj=str(tar_path), path_in_repo=archive_name, repo_id=Config.DATASET_REPO, repo_type="dataset", commit_message=f"HermesFace backup {timestamp}", ) try: files = api.list_repo_files(repo_id=Config.DATASET_REPO, repo_type="dataset") backups = sorted( f for f in files if f.startswith(f"state/{Config.BACKUP_PREFIX}") and f.endswith(".tar.gz") ) if len(backups) > Config.MAX_BACKUPS: for old in backups[: -Config.MAX_BACKUPS]: _log("INFO", "rotating_backup", delete=old) api.delete_file( path_in_repo=old, repo_id=Config.DATASET_REPO, repo_type="dataset", token=Config.HF_TOKEN, ) except Exception as e: _log("WARNING", "rotation_failed", error=str(e)) _log("INFO", "save_completed", file=archive_name) return 0 def load() -> int: api = _api() try: files = api.list_repo_files(repo_id=Config.DATASET_REPO, repo_type="dataset") except RepositoryNotFoundError: _log("ERROR", "repo_not_found", repo=Config.DATASET_REPO) return 1 backups = sorted( (f for f in files if f.startswith(f"state/{Config.BACKUP_PREFIX}") and f.endswith(".tar.gz")), reverse=True, ) if not backups: _log("WARNING", "no_backups_found") return 1 Config.HERMES_HOME.mkdir(parents=True, exist_ok=True) for backup in backups: _log("INFO", "attempting_restore", file=backup) try: local = hf_hub_download( repo_id=Config.DATASET_REPO, repo_type="dataset", filename=backup, token=Config.HF_TOKEN, ) result = restore_backup(Path(local), Config.HERMES_HOME) _log("INFO", "restore_completed", file=backup, restored=result.get("restored", [])) return 0 except Exception as e: _log("ERROR", "restore_failed", file=backup, error=str(e)) continue _log("ERROR", "all_restore_attempts_failed") return 1 def status() -> int: api = _api() try: files = api.list_repo_files(repo_id=Config.DATASET_REPO, repo_type="dataset") except RepositoryNotFoundError: _log("WARNING", "repo_not_found", repo=Config.DATASET_REPO) return 0 backups = sorted( (f for f in files if f.startswith(f"state/{Config.BACKUP_PREFIX}") and f.endswith(".tar.gz")), reverse=True, ) local_files = 0 if Config.HERMES_HOME.exists(): local_files = sum(1 for _ in Config.HERMES_HOME.rglob("*") if _.is_file()) print(json.dumps({ "repo": Config.DATASET_REPO, "local_dir": str(Config.HERMES_HOME), "local_files": local_files, "remote_backups": backups, "max_backups": Config.MAX_BACKUPS, }, indent=2)) return 0 COMMANDS = {"save": save, "load": load, "status": status} def main() -> None: if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS: print(f"Usage: {sys.argv[0]} {{{'|'.join(COMMANDS)}}}", file=sys.stderr) sys.exit(2) sys.exit(COMMANDS[sys.argv[1]]()) if __name__ == "__main__": main()