scripts/backup_allowlist.py

#!/usr/bin/env python3
"""P0-T05: strict backup allowlist and secret-pattern scanning.

Used by save_to_dataset.py / save_to_dataset_atomic.py / sync_hf persistence
paths so that .env, exchange keys, tokens, cookies, sessions with secrets,
and private configuration never enter a Dataset backup.
"""
from __future__ import annotations

import os
import re
from pathlib import Path
from typing import Iterable, List, Optional, Sequence, Tuple

# Directories under HERMES_HOME that may be included in a backup.
# Anything not listed is excluded by default (fail-closed).
ALLOWED_DIR_NAMES = frozenset({
    "memories",
    "skills",
    "plans",
    "skins",
    "cron",
    "hooks",
})

# Explicit file names (basename) that may be included when they sit at the
# HERMES_HOME root or under an allowed directory.
ALLOWED_FILE_NAMES = frozenset({
    "telegram_state.json",
    "futures_symbols_cache.json",
    "SOUL.md",
    "soul.md",
    "AGENTS.md",
    "agents.md",
    "preferences.json",
    "hermes_domain.sqlite3",
})

# Basename patterns that are always excluded.
DENIED_BASENAME_PATTERNS = (
    re.compile(r"^\.env($|\.)", re.I),
    re.compile(r".*\.pem$", re.I),
    re.compile(r".*\.key$", re.I),
    re.compile(r".*credential.*", re.I),
    re.compile(r".*secret.*", re.I),
    re.compile(r".*token.*", re.I),
    re.compile(r".*cookie.*", re.I),
    re.compile(r".*password.*", re.I),
    re.compile(r"^id_rsa", re.I),
    re.compile(r"^id_ed25519", re.I),
    re.compile(r".*\.p12$", re.I),
    re.compile(r".*\.pfx$", re.I),
)

# Content patterns that cause a file to be rejected (secret-like).
SECRET_CONTENT_PATTERNS = (
    re.compile(r'''(?i)['"]?(api[_-]?key|api[_-]?secret|access[_-]?token)['"]?\s*[:=]\s*['"]?[A-Za-z0-9_\-]{16,}'''),
    re.compile(r'''(?i)['"]?(password|passwd|secret)['"]?\s*[:=]\s*['"]?[^\s'"]{8,}'''),
    re.compile(r"(?i)-----BEGIN (RSA |EC |OPENSSH )?PRIVATE KEY-----"),
    re.compile(r"(?i)(hf_|sk-|xox[baprs]-)[A-Za-z0-9]{20,}"),
    re.compile(r"(?i)bearer\s+[A-Za-z0-9\-_\.]{20,}"),
)

# Max bytes scanned per file for secret content (avoid huge binary reads).
_SECRET_SCAN_LIMIT = 64 * 1024


class BackupAllowlistError(RuntimeError):
    """Raised when a backup must abort because secret-like content was found."""


def is_denied_basename(name: str) -> bool:
    base = os.path.basename(name)
    return any(p.search(base) for p in DENIED_BASENAME_PATTERNS)


def is_path_allowed(rel_path: str, *, root_allowed_dirs: Optional[Sequence[str]] = None) -> bool:
    """Return True if relative path is within the strict allowlist."""
    rel = rel_path.replace("\\", "/").lstrip("./")
    if not rel or rel in (".",):
        return True  # directory root marker
    if is_denied_basename(rel):
        return False
    parts = [p for p in rel.split("/") if p and p != "."]
    if not parts:
        return True
    allowed_dirs = frozenset(root_allowed_dirs) if root_allowed_dirs is not None else ALLOWED_DIR_NAMES
    # Single file at root
    if len(parts) == 1:
        return parts[0] in ALLOWED_FILE_NAMES or parts[0] in allowed_dirs
    # Nested: first component must be an allowed directory
    if parts[0] not in allowed_dirs:
        return False
    # Still deny secret-named files deeper in the tree
    if is_denied_basename(parts[-1]):
        return False
    return True


def scan_file_for_secrets(path: Path) -> Optional[str]:
    """Return a short reason if the file appears to contain secrets, else None."""
    try:
        if not path.is_file():
            return None
        if path.stat().st_size == 0:
            return None
        # Skip obvious binaries by extension
        if path.suffix.lower() in {".png", ".jpg", ".jpeg", ".gif", ".webp", ".zip", ".gz", ".tar", ".pyc", ".so"}:
            return None
        with open(path, "rb") as f:
            raw = f.read(_SECRET_SCAN_LIMIT)
        try:
            text = raw.decode("utf-8", errors="ignore")
        except Exception:
            return None
        for pat in SECRET_CONTENT_PATTERNS:
            if pat.search(text):
                return f"secret_pattern:{pat.pattern[:40]}"
    except OSError:
        return None
    return None


def filter_tarinfo(info, *, root: Optional[str] = None):
    """tarfile filter: return None to exclude, or the TarInfo to keep.

    Compatible with tarfile.TarFile.add(..., filter=...).
    """
    name = info.name
    # Always drop locks/tmp/pycache
    if name.endswith((".lock", ".tmp", ".pid", ".socket")):
        return None
    if "__pycache__" in name or name.endswith(".pyc"):
        return None
    if not is_path_allowed(name):
        return None
    return info


def collect_allowed_paths(state_dir: str) -> Tuple[List[Path], List[str]]:
    """Walk state_dir and return (allowed_files, rejection_reasons).

    If any allowed candidate fails secret content scan, raise BackupAllowlistError
    so the backup aborts rather than uploading partial secret material.
    """
    root = Path(state_dir)
    allowed: List[Path] = []
    rejections: List[str] = []
    if not root.is_dir():
        return allowed, rejections

    for dirpath, dirnames, filenames in os.walk(root):
        # Prune denied / non-allowlisted directories early
        rel_dir = os.path.relpath(dirpath, root).replace("\\", "/")
        if rel_dir == ".":
            # Only descend into allowlisted top-level dirs
            dirnames[:] = [d for d in dirnames if d in ALLOWED_DIR_NAMES and not is_denied_basename(d)]
        else:
            if not is_path_allowed(rel_dir):
                dirnames[:] = []
                continue
            dirnames[:] = [d for d in dirnames if not is_denied_basename(d)]

        for fn in filenames:
            full = Path(dirpath) / fn
            rel = os.path.relpath(full, root).replace("\\", "/")
            if not is_path_allowed(rel):
                rejections.append(f"excluded:{rel}")
                continue
            secret_reason = scan_file_for_secrets(full)
            if secret_reason:
                raise BackupAllowlistError(
                    f"Refusing backup: secret-like content in {rel} ({secret_reason})"
                )
            allowed.append(full)
    return allowed, rejections


def tar_filter_factory(state_dir: str):
    """Return a tar filter that enforces the allowlist relative to state_dir."""
    root = Path(state_dir)

    def _filter(info):
        # info.name is arcname-relative (often ".")
        name = info.name
        if name in (".", ""):
            return info
        # When arcname is ".", members look like "./sessions/..." or "sessions/..."
        rel = name.lstrip("./")
        if rel.endswith((".lock", ".tmp", ".pid", ".socket")):
            return None
        if "__pycache__" in rel or rel.endswith(".pyc"):
            return None
        if not is_path_allowed(rel):
            return None
        # Content scan for regular files
        if info.isfile():
            candidate = root / rel
            if candidate.is_file():
                reason = scan_file_for_secrets(candidate)
                if reason:
                    raise BackupAllowlistError(
                        f"Refusing backup: secret-like content in {rel} ({reason})"
                    )
        return info

    return _filter