"""Secrets : tar chiffré par passphrase age (pyrage).""" from __future__ import annotations import getpass import io import os import tarfile import tempfile from collections.abc import Callable, Iterator, Sequence from pathlib import Path import pyrage from osx_backup.context import Context from osx_backup.fsutil import DEPENDENCY_DIRS, is_excluded, tree_size from osx_backup.mapfile import GLOBAL_EXCLUDES, load_map from osx_backup.modules.base import Inventory, Module, ModuleResult, ProgressReporter, register SECRETS_FILE = "secrets.tar.age" PRIVATE_DIRS = (".ssh", ".gnupg") def _entry_excludes(ctx: Context) -> dict[str, list[str]]: return {e.path: GLOBAL_EXCLUDES + list(e.exclude) for e in load_map(ctx.map_path) if e.secret} def _tree_totals(ctx: Context, rels: Sequence[Path]) -> tuple[int, int]: """(nombre de fichiers, octets) attendus pour ces entrées, exclusions appliquées.""" excludes = _entry_excludes(ctx) files = size = 0 for rel in rels: f, s = tree_size(ctx.home / rel, excludes.get(rel.as_posix(), GLOBAL_EXCLUDES)) files += f size += s return files, size def collect_secret_paths(ctx: Context) -> list[Path]: rels: list[Path] = [] for entry_path in _entry_excludes(ctx): if (ctx.home / entry_path).exists(): rels.append(Path(entry_path)) github = ctx.home / "Github" if github.is_dir(): skip = set(DEPENDENCY_DIRS) | {".git"} for root, dirs, files in os.walk(github): dirs[:] = [d for d in dirs if d not in skip] for f in files: if f.startswith(".env"): rels.append((Path(root) / f).relative_to(ctx.home)) return rels def get_passphrase(ctx: Context, confirm: bool) -> str: if ctx.passphrase: return ctx.passphrase env = os.environ.get("OSX_BACKUP_PASSPHRASE") if env: return env if ctx.yes: raise RuntimeError("passphrase requise : définir OSX_BACKUP_PASSPHRASE en mode --yes") first = getpass.getpass("Passphrase de l'archive secrets : ") if not first: raise ValueError("passphrase vide") if confirm and getpass.getpass("Confirmer la passphrase : ") != first: raise ValueError("les passphrases diffèrent") ctx.passphrase = first return first def _iter_entry_files( ctx: Context, rel: Path, patterns: Sequence[str] ) -> Iterator[tuple[Path, str]]: """(chemin absolu, arcname) de chaque fichier de l'entrée, exclusions appliquées. On descend nous-mêmes plutôt que de laisser ``tar.add`` récurser : un seul fichier illisible ou un lien cassé y ferait tomber toute l'entrée, alors qu'on veut sauver le reste. Les liens vers un dossier sont suivis (montage type stow/chezmoi) avec garde anti-boucle. """ root = ctx.home / rel if not root.is_dir(): yield root, rel.as_posix() return seen: set[str] = set() for dirpath, dirnames, filenames in os.walk(root, followlinks=True): real = os.path.realpath(dirpath) if real in seen: dirnames[:] = [] continue seen.add(real) rel_root = Path(dirpath).relative_to(root) dirnames[:] = [d for d in dirnames if not is_excluded(rel_root / d, patterns)] for name in sorted(filenames): inner = rel_root / name if is_excluded(inner, patterns): continue yield root / inner, (rel / inner).as_posix() def build_archive( ctx: Context, rels: Sequence[Path], on_file: Callable[[], None] | None = None, warnings: list[str] | None = None, ) -> bytes: """Tar gzip en mémoire des entrées données. Jamais de tar en clair sur le disque de destination. Les liens symboliques sont déréférencés : un ``~/.ssh/config`` pointant sur un dépôt de dotfiles doit redevenir un vrai fichier sur le Mac neuf. Un fichier illisible ou un lien cassé est signalé puis ignoré — le reste de l'entrée est archivé. Si *aucun* fichier n'a pu être ajouté, on lève plutôt que d'écrire une archive vide qui donnerait l'illusion d'une sauvegarde. ``on_file`` est appelé une fois par fichier réellement archivé ; ``warnings``, s'il est fourni, reçoit chaque avertissement pour que l'appelant le remonte dans son ``ModuleResult`` (et donc dans le manifeste) et pas seulement dans la console. """ excludes = _entry_excludes(ctx) added = 0 def note(message: str) -> None: ctx.warn(message) if warnings is not None: warnings.append(message) with tempfile.TemporaryFile() as tmp: with tarfile.open(fileobj=tmp, mode="w:gz", dereference=True) as tar: for rel in rels: patterns = excludes.get(rel.as_posix(), GLOBAL_EXCLUDES) for src, arcname in _iter_entry_files(ctx, rel, patterns): if not src.exists(): note(f"lien cassé, fichier ignoré : ~/{arcname}") continue if not os.access(src, os.R_OK): note(f"fichier illisible, ignoré : ~/{arcname}") continue if not src.is_file(): # Socket (agent ssh) ou fifo : `gettarinfo` rend None pour un socket et # `add` ne fait alors rien sans lever, et l'extraction ne repose que des # fichiers et des liens. Les compter annoncerait plus de fichiers # chiffrés que l'archive n'en contient et que la restauration n'en pose. note(f"fichier spécial (socket ou fifo), ignoré : ~/{arcname}") continue try: tar.add(src, arcname=arcname, recursive=False) except OSError as exc: note(f"archivage impossible, fichier ignoré : ~/{arcname} ({exc})") continue added += 1 if on_file: on_file() if rels and not added: raise RuntimeError( f"aucun fichier des {len(rels)} entrées secrètes n'a pu être archivé" ) tmp.seek(0) return tmp.read() def _safe_member_rel(root: Path, name: str) -> Path | None: """Chemin relatif du membre, ou None s'il sortirait du dossier d'extraction. ``filter="data"`` réécrit les noms absolus au moment d'extraire, mais ``member.name`` reste tel quel : sans cette vérification, ``home / Path("/etc/passwd")`` vaut ``/etc/passwd`` et une archive forgée ferait déplacer un fichier hors de $HOME. """ rel = Path(name) if rel.is_absolute() or ".." in rel.parts: return None if not (root / rel).resolve().is_relative_to(root.resolve()): return None return rel def extract_archive(ctx: Context, data: bytes) -> int: """Extrait l'archive en zone tampon puis pose chaque fichier dans $HOME. Retourne le compte.""" count = 0 with tempfile.TemporaryDirectory() as tmpdir, tarfile.open(fileobj=io.BytesIO(data)) as tar: root = Path(tmpdir) def _filter(member: tarfile.TarInfo, path: str) -> tarfile.TarInfo | None: try: return tarfile.data_filter(member, path) except tarfile.FilterError as exc: ctx.warn(f"membre d'archive refusé à l'extraction : {member.name} ({exc})") return None tar.extractall(tmpdir, filter=_filter) for member in tar.getmembers(): if not (member.isfile() or member.issym()): continue rel = _safe_member_rel(root, member.name) if rel is None: ctx.warn(f"chemin d'archive hors de $HOME, ignoré : {member.name}") continue src = root / rel if not src.exists() and not src.is_symlink(): continue # membre écarté par le filtre, déjà signalé try: ctx.place(src, ctx.home / rel) if not ctx.dry_run and not member.issym(): os.chmod(ctx.home / rel, 0o600) except OSError as exc: ctx.warn(f"restauration impossible, fichier ignoré : ~/{rel} ({exc})") continue count += 1 if not ctx.dry_run: for d in PRIVATE_DIRS: p = ctx.home / d if p.is_dir(): os.chmod(p, 0o700) return count def _write_archive(ctx: Context, encrypted: bytes) -> None: """Écrit `secrets.tar.age` via un temporaire puis `os.replace`. Écrire directement sur la cible la tronque avant d'avoir la nouvelle archive : une panne (disque plein, débranchement, Ctrl-C) détruirait la sauvegarde précédente des secrets sans en produire de nouvelle. """ target = ctx.dest / SECRETS_FILE if ctx.dry_run: ctx.log(f"[dry-run] write {target} ({len(encrypted)} bytes)") return tmp = target.with_name(target.name + ".tmp") try: tmp.write_bytes(encrypted) os.replace(tmp, target) except BaseException: tmp.unlink(missing_ok=True) raise @register class SecretsModule(Module): name = "secrets" description = "Clés SSH/GPG/age, tokens CLI et fichiers .env, archive chiffrée (age)" order = 50 def inventory(self, ctx: Context) -> Inventory: rels = collect_secret_paths(ctx) _, size = _tree_totals(ctx, rels) missing = [p for p in _entry_excludes(ctx) if not (ctx.home / p).exists()] return Inventory( items=len(rels), size_bytes=size, warnings=[f"absent : ~/{p}" for p in missing] ) def backup(self, ctx: Context, progress: ProgressReporter) -> ModuleResult: rels = collect_secret_paths(ctx) if not rels: return ModuleResult.skipped("aucun secret trouvé") if ctx.dry_run: # Rien n'est chiffré ni écrit : demander une passphrase n'aurait aucun objet. expected, size = _tree_totals(ctx, rels) ctx.log(f"[dry-run] archive {ctx.dest / SECRETS_FILE} ({expected} fichiers)") return ModuleResult.ok("[dry-run] archive planifiée", items=expected, size_bytes=size) try: passphrase = get_passphrase(ctx, confirm=True) except (RuntimeError, ValueError) as exc: return ModuleResult.failed(str(exc)) expected, _ = _tree_totals(ctx, rels) task = progress.task("secrets", total=expected) archived = 0 def _tick() -> None: nonlocal archived archived += 1 task.advance() warnings: list[str] = [] try: data = build_archive(ctx, rels, on_file=_tick, warnings=warnings) except RuntimeError as exc: task.finish() return ModuleResult.failed(str(exc), warnings=warnings) encrypted = pyrage.passphrase.encrypt(data, passphrase) _write_archive(ctx, encrypted) task.finish() return ModuleResult.ok( f"{archived} fichiers chiffrés", items=archived, size_bytes=len(encrypted), warnings=warnings, ) def restore(self, ctx: Context, progress: ProgressReporter) -> ModuleResult: archive = ctx.dest / SECRETS_FILE if not archive.exists(): return ModuleResult.skipped("pas d'archive secrets dans la sauvegarde") try: passphrase = get_passphrase(ctx, confirm=False) except (RuntimeError, ValueError) as exc: return ModuleResult.failed(str(exc)) try: data = pyrage.passphrase.decrypt(archive.read_bytes(), passphrase) except Exception: # pyrage lève DecryptError, on reste générique return ModuleResult.failed("passphrase incorrecte ou archive corrompue") task = progress.task("secrets", total=None) try: count = extract_archive(ctx, data) except (tarfile.TarError, OSError) as exc: task.finish() return ModuleResult.failed(f"archive secrets inexploitable : {exc}") task.finish() return ModuleResult.ok(f"{count} fichiers restaurés", items=count)