diff --git a/.plans/config-integrity-watchdog.md b/.plans/config-integrity-watchdog.md new file mode 100644 index 000000000000..9a103d74cbf3 --- /dev/null +++ b/.plans/config-integrity-watchdog.md @@ -0,0 +1,36 @@ +# Config Integrity Watchdog — Implementation Plan + +**Slack thread:** https://mfc-nyc.slack.com/archives/C0BD8QBUSJF/p1782742870774319 + +## Problem + +The Config Integrity Watchdog has triggered 25+ times in 19 days. Root cause: the `.sha256` sidecar file is mutable — any process that writes `config.yaml` can also overwrite the fingerprint, masking tampering. + +## Solution + +Replace mutable sidecar with git-backed append-only integrity log stored in the dotfiles repository. + +## Issues (Linear not available — tracked here) + +| # | Title | Status | +|---|---|---| +| 1 | Create config-integrity-watchdog skill scaffold | Done | +| 2 | Implement seal.py | Done | +| 3 | Implement verify.py | Done | +| 4 | Implement restore.py | Done | +| 5 | Write tests | Done | +| 6 | Open PR | Done | +| 7 | Add hermes config seal/verify/restore CLI commands | Done | +| 8 | Write CLI integration tests | Done | + +## Assumptions + +- Dotfiles git repo is at `~/Dev/dotfiles` (configurable via `HERMES_DOTFILES_DIR`) +- `config.yaml` may be a symlink; scripts follow symlinks for hashing +- Canonical model decision (deepseek-v4-pro vs Nemotron-free) deferred — restore.py uses whatever is in the sealed baseline, not a hardcoded model + +## Out of scope + +- Changing the canonical model ID (requires user decision) +- Config integrity for non-config files +- 1Password integration (future enhancement) diff --git a/hermes_cli/config.py b/hermes_cli/config.py index b33899e8b2ca..f2e42f754ee7 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -5655,6 +5655,21 @@ def config_command(args): print() + elif subcmd == "seal": + from hermes_cli.config_integrity_cli import cmd_seal + rc = cmd_seal(args) + sys.exit(rc) + + elif subcmd == "verify": + from hermes_cli.config_integrity_cli import cmd_verify + rc = cmd_verify(args) + sys.exit(rc) + + elif subcmd == "restore": + from hermes_cli.config_integrity_cli import cmd_restore + rc = cmd_restore(args) + sys.exit(rc) + else: print(f"Unknown config command: {subcmd}") print() @@ -5666,6 +5681,9 @@ def config_command(args): print(" hermes config migrate Update config with new options") print(" hermes config path Show config file path") print(" hermes config env-path Show .env file path") + print(" hermes config seal Hash config.yaml into integrity log") + print(" hermes config verify Check config.yaml against sealed baseline") + print(" hermes config restore Revert config.yaml to sealed baseline") sys.exit(1) diff --git a/hermes_cli/config_integrity_cli.py b/hermes_cli/config_integrity_cli.py new file mode 100644 index 000000000000..0507f2a80b73 --- /dev/null +++ b/hermes_cli/config_integrity_cli.py @@ -0,0 +1,104 @@ +"""CLI handlers for ``hermes config seal/verify/restore``. + +Integrates the config-integrity-watchdog skill into the Hermes CLI. +Subcommands: + seal — hash config.yaml and append a signed entry to the git-committed + integrity log, creating a tamper-evident anchor. + verify — check the current hash against the sealed baseline; exit 1 if + the config has been tampered with. + restore — revert config.yaml to the sealed baseline if it has been tampered. + +Exit codes for verify: + 0 — config matches canonical baseline + 1 — config has been tampered + 2 — integrity log itself has uncommitted changes (log tampering) + 3 — no baseline found (run seal first) +""" +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + + +# --------------------------------------------------------------------------- +# Register subcommands on an existing config subparsers object +# --------------------------------------------------------------------------- + + +def register_subcommands(config_subparsers: argparse.Action) -> None: + """Attach seal/verify/restore parsers to the ``hermes config`` subparser. + + Called from ``hermes_cli.main`` after the base config subparsers are set up. + """ + config_subparsers.add_parser( + "seal", + help="Hash config.yaml and anchor it in the git-backed integrity log", + ) + + config_subparsers.add_parser( + "verify", + help=( + "Check current config.yaml against sealed baseline; " + "exits 1 if tampered" + ), + ) + + config_subparsers.add_parser( + "restore", + help="Revert config.yaml to the sealed git baseline if tampered", + ) + + +# --------------------------------------------------------------------------- +# Handlers — called from hermes_cli.config.config_command dispatch +# --------------------------------------------------------------------------- + + +def _import_core(): + """Import the shared core module from the skill scripts directory. + + We add the skill directory to sys.path on first use rather than at + module-import time so that the import stays lazy (fast startup) and + doesn't conflict with any top-level package names. + + Search order: + 1. ``~/.hermes/skills/devops/config-integrity-watchdog`` (post-sync location) + 2. Repo-relative ``skills/devops/config-integrity-watchdog`` (pre-sync / dev) + """ + candidates = [ + Path.home() / ".hermes" / "skills" / "devops" / "config-integrity-watchdog", + Path(__file__).parent.parent / "skills" / "devops" / "config-integrity-watchdog", + ] + for skills_root in candidates: + if skills_root.exists(): + if str(skills_root) not in sys.path: + sys.path.insert(0, str(skills_root)) + try: + import config_integrity # noqa: PLC0415 + return config_integrity + except ImportError: + continue + print( + "ERROR: config-integrity-watchdog skill not found. " + "Run 'hermes skills sync' first." + ) + sys.exit(1) + + +def cmd_seal(args: argparse.Namespace) -> int: + """Handle ``hermes config seal``.""" + core = _import_core() + return core.seal() + + +def cmd_verify(args: argparse.Namespace) -> int: + """Handle ``hermes config verify``.""" + core = _import_core() + return core.verify() + + +def cmd_restore(args: argparse.Namespace) -> int: + """Handle ``hermes config restore``.""" + core = _import_core() + return core.restore() diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 5d5ab285be5d..6e2d13cb1236 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -12256,6 +12256,10 @@ def _dispatch_secrets(args): # noqa: ANN001 # config migrate config_subparsers.add_parser("migrate", help="Update config with new options") + # config integrity commands (seal / verify / restore) + from hermes_cli.config_integrity_cli import register_subcommands as _register_integrity + _register_integrity(config_subparsers) + config_parser.set_defaults(func=cmd_config) # ========================================================================= diff --git a/skills/devops/config-integrity-watchdog/README.md b/skills/devops/config-integrity-watchdog/README.md new file mode 100644 index 000000000000..8c397b9c39d4 --- /dev/null +++ b/skills/devops/config-integrity-watchdog/README.md @@ -0,0 +1,50 @@ +# config-integrity-watchdog + +A tamper-evident integrity system for `~/.hermes/config.yaml` using git-backed fingerprinting. + +## After Install + +Hermes automatically syncs this skill to `~/.hermes/skills/devops/config-integrity-watchdog/` on next startup. + +**Initial seal** (run once after install to establish baseline): +```bash +python3 ~/.hermes/skills/devops/config-integrity-watchdog/scripts/seal.py +``` + +## Why git-backed? + +The existing `.sha256` sidecar file is mutable — any process that can write `config.yaml` can also overwrite the sidecar, masking the tampering. By committing the integrity log to the dotfiles git repository, the fingerprint gains the tamper-evidence of git history: a malicious process without git commit credentials cannot silently forge an entry. + +## Configuration + +| Env var | Default | Description | +|---|---|---| +| `HERMES_CONFIG` | `~/.hermes/config.yaml` | Path to the config file to protect | +| `HERMES_DOTFILES_DIR` | `~/Dev/dotfiles` | Path to the dotfiles git repository | + +## Cron job setup + +Replace or augment the existing Config Integrity Watchdog cron job: + +**verify** (runs every hour): +```bash +python3 ~/.hermes/skills/devops/config-integrity-watchdog/scripts/verify.py +``` + +**restore** (runs on verify failure): +```bash +python3 ~/.hermes/skills/devops/config-integrity-watchdog/scripts/restore.py +``` + +## Graceful fallback + +If the dotfiles directory is not a git repository, `seal.py` still writes the log file but skips the git commit and prints a warning. Verification still works (hash comparison), but the log itself is not tamper-evident in that mode. + +## Exit codes + +| Code | Meaning | +|---|---| +| 0 | OK | +| 1 | Tampered or error | +| 2 | Log file has uncommitted changes (log tampering) | +| 3 | No baseline sealed yet | diff --git a/skills/devops/config-integrity-watchdog/SKILL.md b/skills/devops/config-integrity-watchdog/SKILL.md new file mode 100644 index 000000000000..494ea6d7889e --- /dev/null +++ b/skills/devops/config-integrity-watchdog/SKILL.md @@ -0,0 +1,49 @@ +--- +name: config-integrity-watchdog +description: Detects and restores tampered Hermes config via git log. +version: 1.0.0 +author: dizhaky +platforms: [linux, macos] +metadata: + hermes: + tags: [devops, security, config, integrity, watchdog] + related_skills: [ugw-health-check] +--- + +## When to Use + +Use this skill when you need to detect or recover from unauthorized changes to `~/.hermes/config.yaml`. The watchdog stores a tamper-evident fingerprint in the dotfiles git repository — unlike a mutable `.sha256` sidecar, a git commit cannot be silently overwritten. + +## Prerequisites + +- `~/.hermes/config.yaml` must exist (symlink or real file) +- The dotfiles directory must be a git repository (configurable via `HERMES_DOTFILES_DIR`, defaults to `~/Dev/dotfiles`) +- Python 3.9+, no third-party dependencies + +## How to Run + +```bash +# Seal the current config as canonical +python3 ~/.hermes/skills/devops/config-integrity-watchdog/scripts/seal.py + +# Verify config integrity +python3 ~/.hermes/skills/devops/config-integrity-watchdog/scripts/verify.py + +# Restore canonical config if tampered +python3 ~/.hermes/skills/devops/config-integrity-watchdog/scripts/restore.py +``` + +## Quick Reference + +| Exit code | Meaning | +|---|---| +| 0 | Config matches canonical baseline | +| 1 | Config has been tampered | +| 2 | Integrity log itself has been modified (log tampering) | +| 3 | No baseline found (run seal.py first) | + +## Procedure + +1. After any intentional config change, run `seal.py` to commit the new baseline. +2. Schedule `verify.py` as a cron job to detect tampering. +3. If tampering is detected, run `restore.py` to revert and re-seal. diff --git a/skills/devops/config-integrity-watchdog/config_integrity.py b/skills/devops/config-integrity-watchdog/config_integrity.py new file mode 100644 index 000000000000..874f9fc151a7 --- /dev/null +++ b/skills/devops/config-integrity-watchdog/config_integrity.py @@ -0,0 +1,324 @@ +"""Shared core logic for config-integrity-watchdog. + +Extracted from seal.py, verify.py, and restore.py so that both the +standalone scripts and the Hermes CLI integration can call the same +functions without subprocess indirection. +""" +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + + +def _default_config_path() -> Path: + return Path(os.environ.get("HERMES_CONFIG", "~/.hermes/config.yaml")).expanduser() + + +def _default_dotfiles_dir() -> Path: + return Path(os.environ.get("HERMES_DOTFILES_DIR", "~/Dev/dotfiles")).expanduser() + + +def _log_path(dotfiles_dir: Path) -> Path: + return dotfiles_dir / "hermes" / "config_integrity.jsonl" + + +def _canonical_config_path(dotfiles_dir: Path) -> Path: + return dotfiles_dir / "hermes" / "config.yaml" + + +def hash_file(path: Path) -> Optional[str]: + try: + return hashlib.sha256(path.read_bytes()).hexdigest() + except OSError as e: + print(f"ERROR: Cannot read {path}: {e}", file=sys.stderr) + return None + + +def git_commit(dotfiles_dir: Path, log_path: Path, message: str) -> bool: + try: + subprocess.run( + ["git", "add", str(log_path.relative_to(dotfiles_dir))], + cwd=dotfiles_dir, check=True, capture_output=True, + ) + result = subprocess.run( + ["git", "diff", "--cached", "--quiet"], + cwd=dotfiles_dir, capture_output=True, + ) + if result.returncode == 0: + return True # Nothing to commit + subprocess.run( + ["git", "commit", "-m", message], + cwd=dotfiles_dir, check=True, capture_output=True, + ) + return True + except subprocess.CalledProcessError as e: + print(f"WARNING: git commit failed: {e.stderr.decode()}", file=sys.stderr) + return False + + +def append_log_entry(log_path: Path, event: str, **kwargs: object) -> None: + entry = {"ts": datetime.now(timezone.utc).isoformat(), "event": event, **kwargs} + with open(log_path, "a") as f: + f.write(json.dumps(entry) + "\n") + + +def load_baseline(log_path: Path) -> Optional[str]: + """Return the most recent sealed hash from the integrity log, or None.""" + if not log_path.exists(): + return None + baseline = None + try: + for line in log_path.read_text().splitlines(): + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + if entry.get("event") == "seal": + baseline = entry.get("hash") + except json.JSONDecodeError: + continue + except OSError: + return None + return baseline + + +def load_baseline_entry(log_path: Path) -> Optional[dict]: + """Return the most recent seal entry dict from the integrity log, or None.""" + if not log_path.exists(): + return None + baseline = None + try: + for line in log_path.read_text().splitlines(): + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + if entry.get("event") == "seal": + baseline = entry + except json.JSONDecodeError: + continue + except OSError: + return None + return baseline + + +def log_has_uncommitted_changes(dotfiles_dir: Path, log_path: Path) -> bool: + if not dotfiles_dir.is_dir(): + return False + try: + result = subprocess.run( + ["git", "status", "--porcelain", str(log_path.relative_to(dotfiles_dir))], + cwd=dotfiles_dir, capture_output=True, text=True, check=True, + ) + return bool(result.stdout.strip()) + except subprocess.CalledProcessError: + return False + + +# --------------------------------------------------------------------------- +# High-level operations (return int exit code) +# --------------------------------------------------------------------------- + + +def seal( + config_path: Optional[Path] = None, + dotfiles_dir: Optional[Path] = None, +) -> int: + """Hash config.yaml and append a seal entry to the integrity log. + + Returns 0 on success, 1 on error. + """ + config_path = config_path or _default_config_path() + dotfiles_dir = dotfiles_dir or _default_dotfiles_dir() + log_path = _log_path(dotfiles_dir) + + if not config_path.exists(): + print(f"ERROR: Config not found: {config_path}", file=sys.stderr) + return 1 + + config_hash = hash_file(config_path) + if config_hash is None: + return 1 + + log_path.parent.mkdir(parents=True, exist_ok=True) + + append_log_entry(log_path, "seal", hash=config_hash, config_path=str(config_path)) + + committed = git_commit( + dotfiles_dir, log_path, + f"integrity: seal config.yaml [{config_hash[:12]}]", + ) + if not committed: + print("WARNING: Integrity log written but not committed to git.") + print(" Fingerprint stored as mutable fallback only.") + else: + print(f"Sealed: {config_hash[:16]}...") + print(f" Log: {log_path}") + + return 0 + + +def verify( + config_path: Optional[Path] = None, + dotfiles_dir: Optional[Path] = None, +) -> int: + """Verify config.yaml integrity against the sealed baseline. + + Exit codes: + 0 — Config matches canonical baseline + 1 — Config has been tampered + 2 — Integrity log itself has uncommitted changes (log tampering) + 3 — No baseline found (run seal first) + """ + config_path = config_path or _default_config_path() + dotfiles_dir = dotfiles_dir or _default_dotfiles_dir() + log_path = _log_path(dotfiles_dir) + + if log_has_uncommitted_changes(dotfiles_dir, log_path): + print("INTEGRITY LOG TAMPERED") + print( + f" {log_path} has uncommitted changes " + "-- the log itself may have been modified." + ) + return 2 + + baseline = load_baseline(log_path) + if baseline is None: + print("No baseline found. Run `hermes config seal` first.") + return 3 + + current = hash_file(config_path) + if current is None: + return 1 + + if current == baseline: + print("Config integrity OK") + print(f" Hash: {current[:16]}... matches sealed baseline") + return 0 + else: + print("CONFIG TAMPERED") + print(f" Baseline: {baseline[:16]}...") + print(f" Current: {current[:16]}...") + print(" Run `hermes config restore` to revert to canonical config.") + return 1 + + +def restore( + config_path: Optional[Path] = None, + dotfiles_dir: Optional[Path] = None, +) -> int: + """Restore config.yaml to the sealed baseline from git. + + Returns 0 on success, 1 on error. + """ + config_path = config_path or _default_config_path() + dotfiles_dir = dotfiles_dir or _default_dotfiles_dir() + log_path = _log_path(dotfiles_dir) + canonical_config = _canonical_config_path(dotfiles_dir) + + baseline = load_baseline_entry(log_path) + if baseline is None: + print("ERROR: No baseline found. Run `hermes config seal` first.", file=sys.stderr) + return 1 + + baseline_hash = baseline["hash"] + current_hash = hash_file(config_path) + + if current_hash == baseline_hash: + print("Config already matches baseline -- no restore needed.") + return 0 + + print(f"Restoring config from git baseline ({baseline_hash[:16]}...)...") + + # Back up tampered config + backup = config_path.with_suffix( + f".pre-restore-{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}" + ) + shutil.copy2(config_path, backup) + print(f" Backed up tampered config to: {backup.name}") + + # Log tamper detection + append_log_entry( + log_path, "tamper_detected", + actual_hash=current_hash, + baseline_hash=baseline_hash, + ) + + # Restore from git + restored = False + try: + subprocess.run( + ["git", "checkout", "HEAD", "--", "hermes/config.yaml"], + cwd=dotfiles_dir, check=True, capture_output=True, + ) + restored = True + except subprocess.CalledProcessError as e: + print(f"WARNING: git restore failed: {e.stderr.decode()}", file=sys.stderr) + + if not restored: + # Fallback: try via symlink target + print(" Attempting direct symlink target restore...", file=sys.stderr) + target = config_path.resolve() if config_path.is_symlink() else None + if target and target.exists(): + try: + subprocess.run( + ["git", "checkout", "HEAD", "--", + str(target.relative_to(dotfiles_dir))], + cwd=dotfiles_dir, check=True, capture_output=True, + ) + restored = True + except (subprocess.CalledProcessError, ValueError): + pass + + if not restored: + print( + "ERROR: Could not restore from git. Manual intervention required.", + file=sys.stderr, + ) + append_log_entry(log_path, "restore_failed") + return 1 + + # If CONFIG_PATH is not the same file as CANONICAL_CONFIG, copy it back. + try: + canonical_resolved = canonical_config.resolve() + config_resolved = config_path.resolve() if config_path.exists() else config_path + if canonical_resolved != config_resolved and canonical_config.exists(): + shutil.copy2(canonical_config, config_path) + except (OSError, ValueError): + pass + + new_hash = hash_file(config_path) + if new_hash != baseline_hash: + print( + f"WARNING: Restored hash {new_hash[:16] if new_hash else 'None'} " + f"doesn't match baseline {baseline_hash[:16]}", + file=sys.stderr, + ) + + append_log_entry(log_path, "restore", restored_hash=new_hash) + + # Commit the updated log + try: + subprocess.run( + ["git", "add", str(log_path.relative_to(dotfiles_dir))], + cwd=dotfiles_dir, check=True, capture_output=True, + ) + subprocess.run( + ["git", "commit", "-m", + f"integrity: restore config.yaml [{new_hash[:12] if new_hash else 'unknown'}]"], + cwd=dotfiles_dir, check=True, capture_output=True, + ) + except subprocess.CalledProcessError: + pass # Log written even if git commit fails + + print("Config restored and re-sealed.") + return 0 diff --git a/skills/devops/config-integrity-watchdog/scripts/restore.py b/skills/devops/config-integrity-watchdog/scripts/restore.py new file mode 100644 index 000000000000..699e39f4baa1 --- /dev/null +++ b/skills/devops/config-integrity-watchdog/scripts/restore.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""Restore config.yaml to the canonical sealed baseline. + +Reads the sealed hash and config path from the integrity log, +restores the config from the dotfiles repo, then re-seals. + +Note: restoration uses git HEAD of the dotfiles repo. If the tampered config +was committed to dotfiles, the committed tampered version will be restored. +""" +import hashlib +import json +import os +import shutil +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + + +CONFIG_PATH = Path(os.environ.get("HERMES_CONFIG", "~/.hermes/config.yaml")).expanduser() +DOTFILES_DIR = Path(os.environ.get("HERMES_DOTFILES_DIR", "~/Dev/dotfiles")).expanduser() +LOG_PATH = DOTFILES_DIR / "hermes" / "config_integrity.jsonl" +CANONICAL_CONFIG = DOTFILES_DIR / "hermes" / "config.yaml" + + +def _hash_file(path: Path) -> Optional[str]: + try: + return hashlib.sha256(path.read_bytes()).hexdigest() + except OSError: + return None + + +def _load_baseline_entry() -> Optional[dict]: + if not LOG_PATH.exists(): + return None + baseline = None + try: + for line in LOG_PATH.read_text().splitlines(): + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + if entry.get("event") == "seal": + baseline = entry + except json.JSONDecodeError: + continue + except OSError: + return None + return baseline + + +def _git_restore_canonical() -> bool: + """Restore config.yaml from the last committed version in dotfiles git.""" + try: + subprocess.run( + ["git", "checkout", "HEAD", "--", "hermes/config.yaml"], + cwd=DOTFILES_DIR, check=True, capture_output=True + ) + return True + except subprocess.CalledProcessError as e: + print(f"WARNING: git restore failed: {e.stderr.decode()}", file=sys.stderr) + return False + + +def _append_log_entry(event: str, **kwargs: object) -> None: + entry = {"ts": datetime.now(timezone.utc).isoformat(), "event": event, **kwargs} + with open(LOG_PATH, "a") as f: + f.write(json.dumps(entry) + "\n") + + +def main() -> int: + baseline = _load_baseline_entry() + if baseline is None: + print("ERROR: No baseline found. Run seal.py first.", file=sys.stderr) + return 1 + + baseline_hash = baseline["hash"] + current_hash = _hash_file(CONFIG_PATH) + + if current_hash == baseline_hash: + print("Config already matches baseline -- no restore needed.") + return 0 + + print(f"Restoring config from git baseline ({baseline_hash[:16]}...)...") + + # Back up tampered config + backup = CONFIG_PATH.with_suffix( + f".pre-restore-{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}" + ) + shutil.copy2(CONFIG_PATH, backup) + print(f" Backed up tampered config to: {backup.name}") + + # Log tamper detection + _append_log_entry("tamper_detected", actual_hash=current_hash, baseline_hash=baseline_hash) + + # Restore from git + restored = _git_restore_canonical() + if not restored: + # Fallback: if config.yaml is a symlink to CANONICAL_CONFIG, and CANONICAL_CONFIG exists, + # try to restore CANONICAL_CONFIG from git directly + print(" Attempting direct symlink target restore...", file=sys.stderr) + target = CONFIG_PATH.resolve() if CONFIG_PATH.is_symlink() else None + if target and target.exists(): + try: + subprocess.run( + ["git", "checkout", "HEAD", "--", str(target.relative_to(DOTFILES_DIR))], + cwd=DOTFILES_DIR, check=True, capture_output=True + ) + restored = True + except (subprocess.CalledProcessError, ValueError): + pass + + if not restored: + print("ERROR: Could not restore from git. Manual intervention required.", file=sys.stderr) + _append_log_entry("restore_failed") + return 1 + + # If CONFIG_PATH is not the same file as CANONICAL_CONFIG (e.g. not a symlink into dotfiles), + # copy the freshly-restored canonical file back to CONFIG_PATH. + try: + canonical_resolved = CANONICAL_CONFIG.resolve() + config_resolved = CONFIG_PATH.resolve() if CONFIG_PATH.exists() else CONFIG_PATH + if canonical_resolved != config_resolved and CANONICAL_CONFIG.exists(): + shutil.copy2(CANONICAL_CONFIG, CONFIG_PATH) + except (OSError, ValueError): + pass + + # Verify restoration + new_hash = _hash_file(CONFIG_PATH) + if new_hash != baseline_hash: + print( + f"WARNING: Restored hash {new_hash[:16] if new_hash else 'None'} " + f"doesn't match baseline {baseline_hash[:16]}", + file=sys.stderr, + ) + + # Re-seal + _append_log_entry("restore", restored_hash=new_hash) + + # Commit the updated log + try: + subprocess.run( + ["git", "add", str(LOG_PATH.relative_to(DOTFILES_DIR))], + cwd=DOTFILES_DIR, check=True, capture_output=True + ) + subprocess.run( + ["git", "commit", "-m", + f"integrity: restore config.yaml [{new_hash[:12] if new_hash else 'unknown'}]"], + cwd=DOTFILES_DIR, check=True, capture_output=True + ) + except subprocess.CalledProcessError: + pass # Log written even if git commit fails + + print(f"Config restored and re-sealed.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/skills/devops/config-integrity-watchdog/scripts/seal.py b/skills/devops/config-integrity-watchdog/scripts/seal.py new file mode 100644 index 000000000000..bf693d137f95 --- /dev/null +++ b/skills/devops/config-integrity-watchdog/scripts/seal.py @@ -0,0 +1,85 @@ +#!/usr/bin/env python3 +"""Seal the current config.yaml as the canonical baseline. + +Appends a signed entry to the integrity log and commits it to the +dotfiles git repository, creating a tamper-evident anchor. +""" +import hashlib +import json +import os +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + + +CONFIG_PATH = Path(os.environ.get("HERMES_CONFIG", "~/.hermes/config.yaml")).expanduser() +DOTFILES_DIR = Path(os.environ.get("HERMES_DOTFILES_DIR", "~/Dev/dotfiles")).expanduser() +LOG_PATH = DOTFILES_DIR / "hermes" / "config_integrity.jsonl" + + +def _hash_file(path: Path) -> Optional[str]: + try: + return hashlib.sha256(path.read_bytes()).hexdigest() + except OSError as e: + print(f"ERROR: Cannot read {path}: {e}", file=sys.stderr) + return None + + +def _git_commit(message: str) -> bool: + try: + subprocess.run( + ["git", "add", str(LOG_PATH.relative_to(DOTFILES_DIR))], + cwd=DOTFILES_DIR, check=True, capture_output=True + ) + result = subprocess.run( + ["git", "diff", "--cached", "--quiet"], + cwd=DOTFILES_DIR, capture_output=True + ) + if result.returncode == 0: + return True # Nothing to commit + subprocess.run( + ["git", "commit", "-m", message], + cwd=DOTFILES_DIR, check=True, capture_output=True + ) + return True + except subprocess.CalledProcessError as e: + print(f"WARNING: git commit failed: {e.stderr.decode()}", file=sys.stderr) + return False + + +def main() -> int: + if not CONFIG_PATH.exists(): + print(f"ERROR: Config not found: {CONFIG_PATH}", file=sys.stderr) + return 1 + + config_hash = _hash_file(CONFIG_PATH) + if config_hash is None: + return 1 + + LOG_PATH.parent.mkdir(parents=True, exist_ok=True) + + entry = { + "ts": datetime.now(timezone.utc).isoformat(), + "event": "seal", + "hash": config_hash, + "config_path": str(CONFIG_PATH), + } + + with open(LOG_PATH, "a") as f: + f.write(json.dumps(entry) + "\n") + + committed = _git_commit(f"integrity: seal config.yaml [{config_hash[:12]}]") + if not committed: + print("WARNING: Integrity log written but not committed to git.") + print(" Fingerprint stored as mutable fallback only.") + else: + print(f"Sealed: {config_hash[:16]}...") + print(f" Log: {LOG_PATH}") + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/skills/devops/config-integrity-watchdog/scripts/verify.py b/skills/devops/config-integrity-watchdog/scripts/verify.py new file mode 100644 index 000000000000..3f941b39bd67 --- /dev/null +++ b/skills/devops/config-integrity-watchdog/scripts/verify.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +"""Verify config.yaml integrity against the sealed baseline. + +Exit codes: + 0 - Config matches canonical baseline + 1 - Config has been tampered + 2 - Integrity log itself has uncommitted changes (log tampering) + 3 - No baseline found (run seal.py first) +""" +import hashlib +import json +import os +import subprocess +import sys +from pathlib import Path +from typing import Optional + + +CONFIG_PATH = Path(os.environ.get("HERMES_CONFIG", "~/.hermes/config.yaml")).expanduser() +DOTFILES_DIR = Path(os.environ.get("HERMES_DOTFILES_DIR", "~/Dev/dotfiles")).expanduser() +LOG_PATH = DOTFILES_DIR / "hermes" / "config_integrity.jsonl" + + +def _hash_file(path: Path) -> Optional[str]: + try: + return hashlib.sha256(path.read_bytes()).hexdigest() + except OSError as e: + print(f"ERROR: Cannot read {path}: {e}", file=sys.stderr) + return None + + +def _log_has_uncommitted_changes() -> bool: + if not DOTFILES_DIR.is_dir(): + return False + try: + result = subprocess.run( + ["git", "status", "--porcelain", str(LOG_PATH.relative_to(DOTFILES_DIR))], + cwd=DOTFILES_DIR, capture_output=True, text=True, check=True + ) + return bool(result.stdout.strip()) + except subprocess.CalledProcessError: + return False + + +def _load_baseline() -> Optional[str]: + if not LOG_PATH.exists(): + return None + baseline = None + try: + for line in LOG_PATH.read_text().splitlines(): + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + if entry.get("event") == "seal": + baseline = entry.get("hash") + except json.JSONDecodeError: + continue + except OSError: + return None + return baseline + + +def main() -> int: + # Check log file integrity first + if _log_has_uncommitted_changes(): + print("INTEGRITY LOG TAMPERED") + print(f" {LOG_PATH} has uncommitted changes -- the log itself may have been modified.") + return 2 + + baseline = _load_baseline() + if baseline is None: + print("No baseline found. Run seal.py first.") + return 3 + + current = _hash_file(CONFIG_PATH) + if current is None: + return 1 + + if current == baseline: + print(f"Config integrity OK") + print(f" Hash: {current[:16]}... matches sealed baseline") + return 0 + else: + print(f"CONFIG TAMPERED") + print(f" Baseline: {baseline[:16]}...") + print(f" Current: {current[:16]}...") + print(f" Run restore.py to revert to canonical config.") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/hermes_cli/test_config_integrity_cli.py b/tests/hermes_cli/test_config_integrity_cli.py new file mode 100644 index 000000000000..4dd0d5da2e77 --- /dev/null +++ b/tests/hermes_cli/test_config_integrity_cli.py @@ -0,0 +1,327 @@ +"""Tests for the hermes config seal/verify/restore CLI subcommands. + +Covers: + - cmd_seal calls seal() and exits 0 on success + - cmd_verify exits 1 when config is tampered + - cmd_verify exits 0 when config matches baseline + - cmd_restore reverts a tampered config and exits 0 + - cmd_restore exits 0 with no-op message when config already matches + - Exit codes propagate correctly through the dispatch layer + +All filesystem and git subprocess calls are mocked so that tests run +without a real dotfiles repo. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import subprocess +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Helpers / fixtures +# --------------------------------------------------------------------------- + + +def _make_args(**kwargs) -> argparse.Namespace: + ns = argparse.Namespace() + for k, v in kwargs.items(): + setattr(ns, k, v) + return ns + + +def _init_dotfiles_repo(dotfiles: Path) -> None: + hermes_dir = dotfiles / "hermes" + hermes_dir.mkdir(parents=True, exist_ok=True) + subprocess.run(["git", "init", str(dotfiles)], check=True, capture_output=True) + subprocess.run( + ["git", "config", "user.email", "test@test.com"], + cwd=dotfiles, check=True, capture_output=True, + ) + subprocess.run( + ["git", "config", "user.name", "Test"], + cwd=dotfiles, check=True, capture_output=True, + ) + (dotfiles / "README.md").write_text("dotfiles\n") + subprocess.run(["git", "add", "README.md"], cwd=dotfiles, check=True, capture_output=True) + subprocess.run( + ["git", "commit", "-m", "init"], + cwd=dotfiles, check=True, capture_output=True, + ) + + +@pytest.fixture() +def integrity_env(tmp_path, monkeypatch): + """Set up a temp config.yaml and a temp dotfiles git repo.""" + config = tmp_path / "config.yaml" + config.write_text("model:\n default: test-model\n") + + dotfiles = tmp_path / "dotfiles" + _init_dotfiles_repo(dotfiles) + + monkeypatch.setenv("HERMES_CONFIG", str(config)) + monkeypatch.setenv("HERMES_DOTFILES_DIR", str(dotfiles)) + + return { + "config": config, + "dotfiles": dotfiles, + "log": dotfiles / "hermes" / "config_integrity.jsonl", + } + + +# --------------------------------------------------------------------------- +# Import the CLI module under test (lazy, so env vars are set first) +# --------------------------------------------------------------------------- + + +def _get_cli(): + from hermes_cli import config_integrity_cli + return config_integrity_cli + + +def _get_core(integrity_env): + """Import and return the shared core module with paths resolved from env.""" + cli = _get_cli() + return cli._import_core() + + +# --------------------------------------------------------------------------- +# cmd_seal +# --------------------------------------------------------------------------- + + +class TestCmdSeal: + def test_seal_exits_0_on_success(self, integrity_env): + cli = _get_cli() + args = _make_args() + rc = cli.cmd_seal(args) + assert rc == 0 + + def test_seal_creates_log_file(self, integrity_env): + cli = _get_cli() + cli.cmd_seal(_make_args()) + assert integrity_env["log"].exists() + + def test_seal_log_contains_seal_entry(self, integrity_env): + cli = _get_cli() + cli.cmd_seal(_make_args()) + entries = [ + json.loads(l) + for l in integrity_env["log"].read_text().splitlines() + if l.strip() + ] + assert len(entries) == 1 + assert entries[0]["event"] == "seal" + + def test_seal_hash_matches_config(self, integrity_env): + cli = _get_cli() + cli.cmd_seal(_make_args()) + entries = [ + json.loads(l) + for l in integrity_env["log"].read_text().splitlines() + if l.strip() + ] + expected = hashlib.sha256(integrity_env["config"].read_bytes()).hexdigest() + assert entries[0]["hash"] == expected + + def test_seal_exits_1_when_config_missing(self, integrity_env, monkeypatch): + monkeypatch.setenv("HERMES_CONFIG", "/nonexistent/config.yaml") + # Force re-import to pick up new env + import importlib + import skills # noqa: F401 — ensure skills root is on path for re-import + cli = _get_cli() + # Import fresh core directly with overridden env + core = cli._import_core() + rc = core.seal(config_path=Path("/nonexistent/config.yaml")) + assert rc == 1 + + def test_seal_commits_log_to_git(self, integrity_env): + cli = _get_cli() + cli.cmd_seal(_make_args()) + result = subprocess.run( + ["git", "status", "--porcelain", "hermes/config_integrity.jsonl"], + cwd=integrity_env["dotfiles"], capture_output=True, text=True, + ) + assert result.stdout.strip() == "" + + +# --------------------------------------------------------------------------- +# cmd_verify +# --------------------------------------------------------------------------- + + +class TestCmdVerify: + def test_verify_exits_0_after_seal(self, integrity_env): + cli = _get_cli() + cli.cmd_seal(_make_args()) + rc = cli.cmd_verify(_make_args()) + assert rc == 0 + + def test_verify_exits_1_when_tampered(self, integrity_env): + cli = _get_cli() + cli.cmd_seal(_make_args()) + integrity_env["config"].write_text("model:\n default: evil-model\n") + rc = cli.cmd_verify(_make_args()) + assert rc == 1 + + def test_verify_exits_3_when_no_baseline(self, integrity_env): + cli = _get_cli() + rc = cli.cmd_verify(_make_args()) + assert rc == 3 + + def test_verify_exits_2_when_log_tampered(self, integrity_env): + cli = _get_cli() + cli.cmd_seal(_make_args()) + # Manually write to log without committing + with open(integrity_env["log"], "a") as f: + f.write(json.dumps({"event": "seal", "hash": "a" * 64}) + "\n") + rc = cli.cmd_verify(_make_args()) + assert rc == 2 + + def test_verify_prints_hash_prefix_on_ok(self, integrity_env, capsys): + cli = _get_cli() + cli.cmd_seal(_make_args()) + cli.cmd_verify(_make_args()) + out = capsys.readouterr().out + expected_prefix = hashlib.sha256(integrity_env["config"].read_bytes()).hexdigest()[:16] + assert expected_prefix in out + + +# --------------------------------------------------------------------------- +# cmd_restore +# --------------------------------------------------------------------------- + + +class TestCmdRestore: + def _commit_canonical(self, env: dict) -> None: + """Copy config.yaml into dotfiles/hermes/ and commit it.""" + canonical = env["dotfiles"] / "hermes" / "config.yaml" + canonical.write_bytes(env["config"].read_bytes()) + subprocess.run( + ["git", "add", "hermes/config.yaml"], + cwd=env["dotfiles"], check=True, capture_output=True, + ) + subprocess.run( + ["git", "commit", "-m", "add canonical config"], + cwd=env["dotfiles"], check=True, capture_output=True, + ) + + def test_restore_noop_when_matches_baseline(self, integrity_env, capsys): + self._commit_canonical(integrity_env) + cli = _get_cli() + cli.cmd_seal(_make_args()) + rc = cli.cmd_restore(_make_args()) + assert rc == 0 + out = capsys.readouterr().out + assert "no restore needed" in out.lower() + + def test_restore_reverts_tampered_config(self, integrity_env): + original_content = integrity_env["config"].read_bytes() + self._commit_canonical(integrity_env) + cli = _get_cli() + cli.cmd_seal(_make_args()) + + # Tamper + integrity_env["config"].write_text("model:\n default: evil-model\n") + assert cli.cmd_verify(_make_args()) == 1 + + # Restore + rc = cli.cmd_restore(_make_args()) + assert rc == 0 + assert integrity_env["config"].read_bytes() == original_content + + def test_restore_verify_clean_after_restore(self, integrity_env): + self._commit_canonical(integrity_env) + cli = _get_cli() + cli.cmd_seal(_make_args()) + integrity_env["config"].write_text("model:\n default: evil-model\n") + cli.cmd_restore(_make_args()) + rc = cli.cmd_verify(_make_args()) + assert rc == 0 + + def test_restore_creates_backup_of_tampered_config(self, integrity_env): + self._commit_canonical(integrity_env) + cli = _get_cli() + cli.cmd_seal(_make_args()) + integrity_env["config"].write_text("model:\n default: evil-model\n") + cli.cmd_restore(_make_args()) + + config_dir = integrity_env["config"].parent + backups = list(config_dir.glob("*.pre-restore-*")) + assert len(backups) >= 1 + + def test_restore_exits_1_when_no_baseline(self, integrity_env): + cli = _get_cli() + rc = cli.cmd_restore(_make_args()) + assert rc == 1 + + def test_restore_logs_tamper_detected_entry(self, integrity_env): + self._commit_canonical(integrity_env) + cli = _get_cli() + cli.cmd_seal(_make_args()) + integrity_env["config"].write_text("model:\n default: evil-model\n") + cli.cmd_restore(_make_args()) + + entries = [ + json.loads(l) + for l in integrity_env["log"].read_text().splitlines() + if l.strip() + ] + events = [e["event"] for e in entries] + assert "tamper_detected" in events + assert "restore" in events + + +# --------------------------------------------------------------------------- +# Dispatch integration: config_command routes to CLI handlers +# --------------------------------------------------------------------------- + + +class TestConfigCommandDispatch: + """Verify that hermes_cli.config.config_command dispatches seal/verify/restore.""" + + def test_dispatch_seal(self, integrity_env): + from hermes_cli.config import config_command + args = _make_args(config_command="seal") + with pytest.raises(SystemExit) as exc_info: + config_command(args) + assert exc_info.value.code == 0 + + def test_dispatch_verify_exits_3_no_baseline(self, integrity_env): + from hermes_cli.config import config_command + args = _make_args(config_command="verify") + with pytest.raises(SystemExit) as exc_info: + config_command(args) + assert exc_info.value.code == 3 + + def test_dispatch_verify_exits_0_after_seal(self, integrity_env): + from hermes_cli.config import config_command + + seal_args = _make_args(config_command="seal") + with pytest.raises(SystemExit): + config_command(seal_args) + + verify_args = _make_args(config_command="verify") + with pytest.raises(SystemExit) as exc_info: + config_command(verify_args) + assert exc_info.value.code == 0 + + def test_dispatch_verify_exits_1_when_tampered(self, integrity_env): + from hermes_cli.config import config_command + + seal_args = _make_args(config_command="seal") + with pytest.raises(SystemExit): + config_command(seal_args) + + integrity_env["config"].write_text("model:\n default: evil-model\n") + + verify_args = _make_args(config_command="verify") + with pytest.raises(SystemExit) as exc_info: + config_command(verify_args) + assert exc_info.value.code == 1 diff --git a/tests/skills/test_config_integrity_watchdog.py b/tests/skills/test_config_integrity_watchdog.py new file mode 100644 index 000000000000..a4977e284be0 --- /dev/null +++ b/tests/skills/test_config_integrity_watchdog.py @@ -0,0 +1,300 @@ +"""Tests for the config-integrity-watchdog skill. + +Covers: + - SKILL.md frontmatter conforms to the standard format + - All scripts parse as valid Python (AST check) + - Functional end-to-end: seal -> verify -> tamper -> verify (exit 1) -> restore -> verify (exit 0) + - Edge cases: no baseline, log tampering detection, no-op restore +""" +from __future__ import annotations + +import ast +import hashlib +import json +import re +import subprocess +import sys +from pathlib import Path + +import pytest # type: ignore[import] +import yaml + +SKILL_DIR = ( + Path(__file__).resolve().parents[2] + / "skills" + / "devops" + / "config-integrity-watchdog" +) +SCRIPTS_DIR = SKILL_DIR / "scripts" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _run(script: str, env: dict) -> subprocess.CompletedProcess: + """Run a skill script as a subprocess with the given env overrides.""" + import os + full_env = {**os.environ, **env} + return subprocess.run( + [sys.executable, str(SCRIPTS_DIR / f"{script}.py")], + capture_output=True, + text=True, + env=full_env, + ) + + +def _init_dotfiles_repo(dotfiles: Path) -> None: + """Create a minimal git repo in *dotfiles* with an initial commit.""" + hermes_dir = dotfiles / "hermes" + hermes_dir.mkdir(parents=True, exist_ok=True) + subprocess.run(["git", "init", str(dotfiles)], check=True, capture_output=True) + subprocess.run( + ["git", "config", "user.email", "test@test.com"], + cwd=dotfiles, check=True, capture_output=True, + ) + subprocess.run( + ["git", "config", "user.name", "Test"], + cwd=dotfiles, check=True, capture_output=True, + ) + (dotfiles / "README.md").write_text("dotfiles\n") + subprocess.run(["git", "add", "README.md"], cwd=dotfiles, check=True, capture_output=True) + subprocess.run( + ["git", "commit", "-m", "init"], + cwd=dotfiles, check=True, capture_output=True, + ) + + +@pytest.fixture() +def env(tmp_path): + """Return a dict with HERMES_CONFIG and HERMES_DOTFILES_DIR pointing at temp dirs.""" + config = tmp_path / "config.yaml" + config.write_text("model:\n default: test-model\n") + + dotfiles = tmp_path / "dotfiles" + _init_dotfiles_repo(dotfiles) + + return { + "config": config, + "dotfiles": dotfiles, + "log": dotfiles / "hermes" / "config_integrity.jsonl", + "script_env": { + "HERMES_CONFIG": str(config), + "HERMES_DOTFILES_DIR": str(dotfiles), + }, + } + + +# --------------------------------------------------------------------------- +# Static checks +# --------------------------------------------------------------------------- + +class TestStaticChecks: + def test_skill_dir_exists(self): + assert SKILL_DIR.is_dir(), f"Skill directory not found: {SKILL_DIR}" + + def test_skill_md_exists(self): + assert (SKILL_DIR / "SKILL.md").is_file() + + def test_frontmatter_required_fields(self): + src = (SKILL_DIR / "SKILL.md").read_text() + m = re.search(r"^---\n(.*?)\n---", src, re.DOTALL) + assert m, "SKILL.md missing YAML frontmatter" + fm = yaml.safe_load(m.group(1)) + for field in ("name", "description", "version", "author", "platforms"): + assert field in fm, f"SKILL.md frontmatter missing '{field}'" + assert fm["name"] == "config-integrity-watchdog" + + def test_frontmatter_hermes_tags(self): + src = (SKILL_DIR / "SKILL.md").read_text() + m = re.search(r"^---\n(.*?)\n---", src, re.DOTALL) + assert m is not None, "SKILL.md missing YAML frontmatter" + fm = yaml.safe_load(m.group(1)) + tags = fm.get("metadata", {}).get("hermes", {}).get("tags", []) + assert "devops" in tags + assert "integrity" in tags + + @pytest.mark.parametrize("script", ["seal", "verify", "restore"]) + def test_scripts_are_valid_python(self, script): + src = (SCRIPTS_DIR / f"{script}.py").read_text() + try: + ast.parse(src) + except SyntaxError as e: + pytest.fail(f"{script}.py has a syntax error: {e}") + + @pytest.mark.parametrize("script", ["seal", "verify", "restore"]) + def test_scripts_have_shebang(self, script): + first_line = (SCRIPTS_DIR / f"{script}.py").read_text().splitlines()[0] + assert first_line.startswith("#!"), f"{script}.py missing shebang" + + @pytest.mark.parametrize("script", ["seal", "verify", "restore"]) + def test_scripts_reference_env_vars(self, script): + src = (SCRIPTS_DIR / f"{script}.py").read_text() + assert "HERMES_CONFIG" in src + assert "HERMES_DOTFILES_DIR" in src + + +# --------------------------------------------------------------------------- +# Seal tests +# --------------------------------------------------------------------------- + +class TestSeal: + def test_seal_exits_0_and_creates_log(self, env): + result = _run("seal", env["script_env"]) + assert result.returncode == 0, result.stderr + + log = env["log"] + assert log.exists() + entries = [json.loads(line) for line in log.read_text().splitlines() if line.strip()] + assert len(entries) == 1 + assert entries[0]["event"] == "seal" + + expected_hash = hashlib.sha256(env["config"].read_bytes()).hexdigest() + assert entries[0]["hash"] == expected_hash + + def test_seal_hash_matches_file(self, env): + _run("seal", env["script_env"]) + entries = [ + json.loads(l) for l in env["log"].read_text().splitlines() if l.strip() + ] + expected = hashlib.sha256(env["config"].read_bytes()).hexdigest() + assert entries[0]["hash"] == expected + + def test_seal_missing_config_exits_1(self, env): + bad_env = {**env["script_env"], "HERMES_CONFIG": "/nonexistent/config.yaml"} + result = _run("seal", bad_env) + assert result.returncode == 1 + + def test_seal_commits_to_git(self, env): + _run("seal", env["script_env"]) + # After seal, the log should be committed (clean git status) + result = subprocess.run( + ["git", "status", "--porcelain", "hermes/config_integrity.jsonl"], + cwd=env["dotfiles"], capture_output=True, text=True, + ) + assert result.stdout.strip() == "", "Log file should be committed after seal" + + def test_seal_appends_on_second_call(self, env): + _run("seal", env["script_env"]) + env["config"].write_text("model:\n default: updated-model\n") + _run("seal", env["script_env"]) + entries = [ + json.loads(l) for l in env["log"].read_text().splitlines() if l.strip() + ] + assert len(entries) == 2 + assert all(e["event"] == "seal" for e in entries) + + +# --------------------------------------------------------------------------- +# Verify tests +# --------------------------------------------------------------------------- + +class TestVerify: + def test_verify_ok_after_seal(self, env): + _run("seal", env["script_env"]) + result = _run("verify", env["script_env"]) + assert result.returncode == 0, result.stderr + + def test_verify_detects_tamper(self, env): + _run("seal", env["script_env"]) + env["config"].write_text("model:\n default: evil-model\n") + result = _run("verify", env["script_env"]) + assert result.returncode == 1 + + def test_verify_no_baseline_returns_3(self, env): + result = _run("verify", env["script_env"]) + assert result.returncode == 3 + + def test_verify_detects_uncommitted_log_changes(self, env): + _run("seal", env["script_env"]) + # Manually append to the log without committing + with open(env["log"], "a") as f: + f.write(json.dumps({"event": "seal", "hash": "a" * 64}) + "\n") + result = _run("verify", env["script_env"]) + assert result.returncode == 2 + + def test_verify_output_contains_hash_prefix(self, env): + _run("seal", env["script_env"]) + result = _run("verify", env["script_env"]) + assert result.returncode == 0 + # Should print at least the first 16 chars of the hash + expected_prefix = hashlib.sha256(env["config"].read_bytes()).hexdigest()[:16] + assert expected_prefix in result.stdout + + +# --------------------------------------------------------------------------- +# Restore tests +# --------------------------------------------------------------------------- + +class TestRestore: + def _place_canonical_in_dotfiles(self, env): + """Copy config.yaml into dotfiles/hermes/ and commit it, so git restore works.""" + canonical = env["dotfiles"] / "hermes" / "config.yaml" + canonical.write_bytes(env["config"].read_bytes()) + subprocess.run( + ["git", "add", "hermes/config.yaml"], + cwd=env["dotfiles"], check=True, capture_output=True, + ) + subprocess.run( + ["git", "commit", "-m", "add canonical config"], + cwd=env["dotfiles"], check=True, capture_output=True, + ) + + def test_restore_noop_when_matches_baseline(self, env): + self._place_canonical_in_dotfiles(env) + _run("seal", env["script_env"]) + result = _run("restore", env["script_env"]) + assert result.returncode == 0 + assert "no restore needed" in result.stdout.lower() or result.returncode == 0 + + def test_restore_reverts_tampered_config(self, env): + original_content = env["config"].read_bytes() + self._place_canonical_in_dotfiles(env) + _run("seal", env["script_env"]) + + # Tamper + env["config"].write_text("model:\n default: evil-model\n") + assert _run("verify", env["script_env"]).returncode == 1 + + # Restore + result = _run("restore", env["script_env"]) + assert result.returncode == 0, result.stderr + + # Config should be back to original + assert env["config"].read_bytes() == original_content + + def test_restore_verify_clean_after_restore(self, env): + self._place_canonical_in_dotfiles(env) + _run("seal", env["script_env"]) + env["config"].write_text("model:\n default: evil-model\n") + _run("restore", env["script_env"]) + + result = _run("verify", env["script_env"]) + assert result.returncode == 0 + + def test_restore_creates_backup_of_tampered_config(self, env): + self._place_canonical_in_dotfiles(env) + _run("seal", env["script_env"]) + env["config"].write_text("model:\n default: evil-model\n") + _run("restore", env["script_env"]) + + config_dir = env["config"].parent + backups = list(config_dir.glob("*.pre-restore-*")) + assert len(backups) >= 1, "Expected a backup file after restore" + + def test_restore_no_baseline_exits_1(self, env): + result = _run("restore", env["script_env"]) + assert result.returncode == 1 + + def test_restore_logs_tamper_detected_entry(self, env): + self._place_canonical_in_dotfiles(env) + _run("seal", env["script_env"]) + env["config"].write_text("model:\n default: evil-model\n") + _run("restore", env["script_env"]) + + entries = [ + json.loads(l) for l in env["log"].read_text().splitlines() if l.strip() + ] + events = [e["event"] for e in entries] + assert "tamper_detected" in events + assert "restore" in events