diff --git a/hooks/.gitkeep b/hooks/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/hooks/secrets_guard.py b/hooks/secrets_guard.py new file mode 100644 index 0000000..c539df4 --- /dev/null +++ b/hooks/secrets_guard.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python3 +"""Universal secrets-read guard for Claude Code, Codex, and Cursor hooks. + +One stdlib-only file, no sibling imports — safe to copy anywhere. Detects the +calling agent from the stdin payload shape and answers in that agent's dialect: + +- Claude Code / Codex (PreToolUse): stdin {"tool_name": ..., "tool_input": {...}}, + deny via {"hookSpecificOutput": {"permissionDecision": "deny", ...}} on stdout, + allow via silence (exit 0, no output). +- Cursor (beforeReadFile / beforeShellExecution): stdin has "hook_event_name" set + to one of those event names with "file_path"/"command" at the top level, + deny via {"permission": "deny", ...}, allow via {"permission": "allow"}. + +Blocks reads of dlt secrets (secrets.toml, *.secrets.toml) and dotenv files +(.env, .env.* except .env.example/.env.template/.env.sample). +""" + +import json +import os +import shlex +import sys + +_CURSOR_EVENTS = {"beforeReadFile", "beforeShellExecution"} +_ALLOWED_ENV_SUFFIXES = {"example", "template", "sample"} + +DENY_MESSAGE = ( + "Blocked: direct access to secrets/env files is not allowed. " + "Use the dlt-workspace-mcp `secrets_view_redacted` tool to inspect values (redacted) " + "or `secrets_update_fragment` to write placeholders. See the setup-secrets skill." +) + + +def is_blocked_path(path: str) -> bool: + """True if the basename of `path` looks like a dlt secrets or dotenv file.""" + name = os.path.basename(path.replace("\\", "/")) + + if name == "secrets.toml" or name.endswith(".secrets.toml"): + return True + + if name == ".env": + return True + if name.startswith(".env."): + suffix = name.rsplit(".", 1)[-1] + return suffix not in _ALLOWED_ENV_SUFFIXES + + return False + + +def command_is_blocked(command: str) -> bool: + """True if any token in a shell command string looks like a blocked path.""" + try: + tokens = shlex.split(command) + except ValueError: + tokens = command.split() + + return any(is_blocked_path(token) for token in tokens) + + +def _claude_codex_blocked(payload: dict) -> bool: + tool_name = payload.get("tool_name", "") + tool_input = payload.get("tool_input") or {} + + if tool_name == "Read": + return is_blocked_path(tool_input.get("file_path", "")) + + if tool_name == "Grep": + paths = tool_input.get("paths") or [] + if tool_input.get("path"): + paths = [*paths, tool_input["path"]] + return any(is_blocked_path(p) for p in paths) + + if tool_name == "Bash": + return command_is_blocked(tool_input.get("command", "")) + + return False + + +def _cursor_blocked(payload: dict) -> bool: + if payload["hook_event_name"] == "beforeReadFile": + return is_blocked_path(payload.get("file_path", "")) + return command_is_blocked(payload.get("command", "")) + + +def main() -> None: + payload = json.load(sys.stdin) + + if payload.get("hook_event_name") in _CURSOR_EVENTS: + if _cursor_blocked(payload): + print(json.dumps({"permission": "deny", "user_message": DENY_MESSAGE})) + else: + print(json.dumps({"permission": "allow"})) + return + + if _claude_codex_blocked(payload): + print( + json.dumps( + { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": DENY_MESSAGE, + } + } + ) + ) + + +if __name__ == "__main__": + try: + main() + except Exception: + # Fail open: a bug in this guard must never block normal tool use. + # For Cursor a reply is expected, but hooks fail open there too when + # no valid JSON is produced. + pass diff --git a/pyproject.toml b/pyproject.toml index e6550ee..77dde31 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -89,7 +89,8 @@ path = "hatch_build.py" [tool.hatch.build.targets.wheel] packages = ["src/dlthub_init"] -# Ship the single source-of-truth skills/ (repo root) into the wheel without a -# committed duplicate. Read back at runtime as dlthub_init/_bundled_skills. +# Ship the single source-of-truth skills/ and hooks/ (repo root) into the wheel +# without committed duplicates. Read back at runtime as dlthub_init/_bundled_*. [tool.hatch.build.targets.wheel.force-include] "skills" = "dlthub_init/_bundled_skills" +"hooks" = "dlthub_init/_bundled_hooks" diff --git a/scripts/generate_skills.py b/scripts/generate_skills.py index e5b79f7..9dfefdf 100644 --- a/scripts/generate_skills.py +++ b/scripts/generate_skills.py @@ -18,6 +18,7 @@ SCRIPT_PATH = Path(__file__).resolve() SKILLS_DIR = SCRIPT_PATH.parent.parent / "skills" +HOOKS_DIR = SCRIPT_PATH.parent.parent / "hooks" WORKBENCH_REPO = os.environ.get("DLTHUB_WORKBENCH_REPO", "https://github.com/dlt-hub/dlthub-ai-workbench.git") WORKBENCH_BRANCH = "master" @@ -59,6 +60,23 @@ def _reset_skills_dir() -> None: shutil.rmtree(entry) if entry.is_dir() else entry.unlink() +def _copy_hooks(workbench: Path) -> int: + """Sync workbench/init/hooks/*.py into hooks/. Tolerates refs that predate + the hooks directory (keeps whatever is committed and warns instead).""" + source = workbench / "workbench" / "init" / "hooks" + scripts = sorted(source.glob("*.py")) if source.is_dir() else [] + if not scripts: + print("warning: workbench ref has no init hook scripts; keeping committed hooks/") + return 0 + HOOKS_DIR.mkdir(exist_ok=True) + for entry in HOOKS_DIR.iterdir(): + if entry.name != _KEEP: + shutil.rmtree(entry) if entry.is_dir() else entry.unlink() + for script in scripts: + shutil.copy2(script, HOOKS_DIR / script.name) + return len(scripts) + + def _copy_toolkit_skills(workbench: Path, toolkits: tuple[str, ...]) -> dict[str, str]: collected: dict[str, str] = {} for toolkit in toolkits: @@ -90,7 +108,10 @@ def main() -> int: _git(workbench, "checkout", "--quiet", full) _reset_skills_dir() collected = _copy_toolkit_skills(workbench, toolkits) + hook_count = _copy_hooks(workbench) print(f"Wrote {len(collected)} skill(s) from [{', '.join(toolkits)}]: {', '.join(sorted(collected))}") + if hook_count: + print(f"Wrote {hook_count} hook script(s) from init") return 0 diff --git a/src/dlthub_init/cli.py b/src/dlthub_init/cli.py index 1ad9af9..7588267 100644 --- a/src/dlthub_init/cli.py +++ b/src/dlthub_init/cli.py @@ -21,6 +21,7 @@ ) from .errors import CollisionError, UvError, WorkspaceError from .prompts import confirm +from .hooks import install_hooks from .scaffold import apply_scaffold, resolve_target from .skills import install_skills from .uv import execute_uv_install, find_uv, run_uv_sync @@ -146,9 +147,12 @@ def run(args: argparse.Namespace) -> None: plan = apply_scaffold(project_dir, scaffold=scaffold, flags=flags) installed_skills = install_skills(project_dir) + configured_hooks = install_hooks(project_dir) print_summary(plan) if installed_skills: substep_detail(strings.MSG_SKILLS_INSTALLED.format(count=len(installed_skills))) + if configured_hooks: + substep_detail(strings.MSG_HOOKS_INSTALLED.format(agents=", ".join(configured_hooks))) synced = _maybe_sync(project_dir, args, verbose=verbose) print_next_steps(project_dir, synced=synced, uv_installed=find_uv() is not None) diff --git a/src/dlthub_init/hooks.py b/src/dlthub_init/hooks.py new file mode 100644 index 0000000..9c67c18 --- /dev/null +++ b/src/dlthub_init/hooks.py @@ -0,0 +1,150 @@ +"""Install the bundled secrets-read guard hook into a scaffolded workspace. + +The canonical guard script lives once in the repo-root hooks/ directory (synced +from the dltHub AI workbench by scripts/generate_skills.py). The wheel +force-includes it at dlthub_init/_bundled_hooks/; a source checkout reads it +from repo-root hooks/. The script is copied to .agents/hooks/ and referenced by +per-agent hook configs (.claude/settings.json, .cursor/hooks.json, +.codex/hooks.json) that all point at that single copy. + +Agents give no common guarantee about the hook process's working directory: +Claude Code exposes $CLAUDE_PROJECT_DIR for project-level settings; Cursor and +Codex expose nothing, so their commands locate the project root via +`git rev-parse --show-toplevel`, falling back to the current directory. +Codex hooks are written to .codex/hooks.json rather than .codex/config.toml +because repo-local config.toml hooks do not fire in interactive sessions +(openai/codex#17532). +""" + +from __future__ import annotations + +import json +import shutil +from pathlib import Path +from typing import Any + +from .errors import ScaffoldError +from . import strings + +_PACKAGE_DIR = Path(__file__).resolve().parent +_BUNDLED_HOOKS = _PACKAGE_DIR / "_bundled_hooks" +_SOURCE_HOOKS = _PACKAGE_DIR.parents[1] / "hooks" + +_SCRIPT_NAME = "secrets_guard.py" +CANONICAL_DIR = Path(".agents") / "hooks" + +_CLAUDE_COMMAND = 'python3 "$CLAUDE_PROJECT_DIR"/.agents/hooks/secrets_guard.py' +_GIT_ROOT_COMMAND = 'python3 "$(git rev-parse --show-toplevel 2>/dev/null || pwd)"/.agents/hooks/secrets_guard.py' + + +def hooks_source() -> Path | None: + if _BUNDLED_HOOKS.is_dir(): + return _BUNDLED_HOOKS + if _SOURCE_HOOKS.is_dir(): + return _SOURCE_HOOKS + return None + + +def install_hooks(project_dir: Path) -> list[str]: + """Copy the guard script into .agents/hooks/ and register it with each agent. + + Merges into existing config files without clobbering unrelated content, and + skips any config that already references the guard script. Configs that + exist but are not valid JSON are left untouched. Returns the names of the + agent configs newly written or updated. + """ + source = hooks_source() + if source is None or not (source / _SCRIPT_NAME).is_file(): + return [] + + try: + script_dest = project_dir / CANONICAL_DIR / _SCRIPT_NAME + if not script_dest.exists(): + script_dest.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source / _SCRIPT_NAME, script_dest) + + configured: list[str] = [] + if _configure_claude(project_dir): + configured.append("claude") + if _configure_cursor(project_dir): + configured.append("cursor") + if _configure_codex(project_dir): + configured.append("codex") + except OSError as exc: + raise ScaffoldError(strings.ERROR_WRITE_FAILED.format(path=project_dir, reason=exc)) from exc + return configured + + +def _pretooluse_entry(command: str, matcher: str) -> dict[str, Any]: + return { + "matcher": matcher, + "hooks": [{"type": "command", "command": command}], + } + + +def _configure_claude(project_dir: Path) -> bool: + path = project_dir / ".claude" / "settings.json" + settings = _load_json_object(path) + if settings is None: + return False + + pre_tool_use = settings.setdefault("hooks", {}).setdefault("PreToolUse", []) + if _references_guard(pre_tool_use): + return False + pre_tool_use.append(_pretooluse_entry(_CLAUDE_COMMAND, "Read|Grep|Bash")) + _write_json(path, settings) + return True + + +def _configure_cursor(project_dir: Path) -> bool: + path = project_dir / ".cursor" / "hooks.json" + config = _load_json_object(path) + if config is None: + return False + + config.setdefault("version", 1) + hooks = config.setdefault("hooks", {}) + changed = False + for event in ("beforeReadFile", "beforeShellExecution"): + entries = hooks.setdefault(event, []) + if not _references_guard(entries): + entries.append({"command": _GIT_ROOT_COMMAND}) + changed = True + if changed: + _write_json(path, config) + return changed + + +def _configure_codex(project_dir: Path) -> bool: + path = project_dir / ".codex" / "hooks.json" + config = _load_json_object(path) + if config is None: + return False + + pre_tool_use = config.setdefault("hooks", {}).setdefault("PreToolUse", []) + if _references_guard(pre_tool_use): + return False + # Codex has no dedicated Read/Grep tool; file access goes through shell + pre_tool_use.append(_pretooluse_entry(_GIT_ROOT_COMMAND, "Bash")) + _write_json(path, config) + return True + + +def _references_guard(entries: object) -> bool: + return _SCRIPT_NAME in json.dumps(entries) + + +def _load_json_object(path: Path) -> dict[str, Any] | None: + """Read a JSON config, returning {} if absent and None if unusable.""" + if not path.exists(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + return data if isinstance(data, dict) else None + + +def _write_json(path: Path, data: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") diff --git a/src/dlthub_init/strings.py b/src/dlthub_init/strings.py index 6c0ff07..abb848e 100644 --- a/src/dlthub_init/strings.py +++ b/src/dlthub_init/strings.py @@ -44,6 +44,7 @@ MSG_UNEXPECTED_ERROR_HINT = "[dim]Re-run with --verbose to see the full traceback.[/dim]" MSG_HEADER = "Initializing a dltHub workspace in [bold]{project_dir}[/bold]" MSG_SKILLS_INSTALLED = "Added {count} skill(s) to .agents/skills/ (linked into .claude/skills/)" +MSG_HOOKS_INSTALLED = "Added secrets-read guard hook to .agents/hooks/ (configured: {agents})" MSG_INSTALLING_DEPS = "Installing dependencies" MSG_INSTALLED_DEPS = "Installed dependencies into .venv" MSG_SKIPPED_SYNC = "\n[yellow]Skipped[/yellow] dependency sync." diff --git a/tests/test_hooks.py b/tests/test_hooks.py new file mode 100644 index 0000000..a3178fe --- /dev/null +++ b/tests/test_hooks.py @@ -0,0 +1,119 @@ +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +import dlthub_init.hooks as hooks + + +class InstallHooksTest(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + root = Path(self._tmp.name) + self.project_dir = root / "ws" + self.project_dir.mkdir() + self.source = root / "src_hooks" + self.source.mkdir() + (self.source / "secrets_guard.py").write_text("# guard\n", encoding="utf-8") + + def tearDown(self): + self._tmp.cleanup() + + def _install(self): + with patch("dlthub_init.hooks.hooks_source", return_value=self.source): + return hooks.install_hooks(self.project_dir) + + def _read_json(self, rel): + return json.loads((self.project_dir / rel).read_text(encoding="utf-8")) + + def test_copies_script_into_agents_hooks(self): + self._install() + self.assertTrue((self.project_dir / ".agents/hooks/secrets_guard.py").exists()) + + def test_configures_all_three_agents(self): + configured = self._install() + self.assertEqual(configured, ["claude", "cursor", "codex"]) + + claude = self._read_json(".claude/settings.json") + self.assertEqual(claude["hooks"]["PreToolUse"][0]["matcher"], "Read|Grep|Bash") + self.assertIn("$CLAUDE_PROJECT_DIR", claude["hooks"]["PreToolUse"][0]["hooks"][0]["command"]) + + cursor = self._read_json(".cursor/hooks.json") + self.assertEqual(cursor["version"], 1) + for event in ("beforeReadFile", "beforeShellExecution"): + self.assertIn("secrets_guard.py", cursor["hooks"][event][0]["command"]) + + codex = self._read_json(".codex/hooks.json") + self.assertEqual(codex["hooks"]["PreToolUse"][0]["matcher"], "Bash") + self.assertIn("secrets_guard.py", codex["hooks"]["PreToolUse"][0]["hooks"][0]["command"]) + + def test_merges_into_existing_claude_settings(self): + settings_path = self.project_dir / ".claude/settings.json" + settings_path.parent.mkdir(parents=True) + settings_path.write_text(json.dumps({"permissions": {"allow": ["Bash(ls *)"]}}), encoding="utf-8") + self._install() + merged = self._read_json(".claude/settings.json") + self.assertEqual(merged["permissions"]["allow"], ["Bash(ls *)"]) + self.assertIn("secrets_guard.py", json.dumps(merged["hooks"])) + + def test_skips_agents_already_configured(self): + first = self._install() + second = self._install() + self.assertEqual(first, ["claude", "cursor", "codex"]) + self.assertEqual(second, []) + claude = self._read_json(".claude/settings.json") + self.assertEqual(len(claude["hooks"]["PreToolUse"]), 1) + + def test_leaves_invalid_json_untouched(self): + settings_path = self.project_dir / ".claude/settings.json" + settings_path.parent.mkdir(parents=True) + settings_path.write_text("{not json", encoding="utf-8") + configured = self._install() + self.assertEqual(configured, ["cursor", "codex"]) + self.assertEqual(settings_path.read_text(encoding="utf-8"), "{not json") + + def test_never_overwrites_existing_script(self): + dest = self.project_dir / ".agents/hooks/secrets_guard.py" + dest.parent.mkdir(parents=True) + dest.write_text("USER OWNED", encoding="utf-8") + self._install() + self.assertEqual(dest.read_text(encoding="utf-8"), "USER OWNED") + + def test_no_source_returns_empty(self): + with patch("dlthub_init.hooks.hooks_source", return_value=None): + self.assertEqual(hooks.install_hooks(self.project_dir), []) + + +class HooksSourceTest(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.root = Path(self._tmp.name) + self.bundled = self.root / "bundled" + self.source = self.root / "source" + + def tearDown(self): + self._tmp.cleanup() + + def _resolve(self): + with ( + patch.object(hooks, "_BUNDLED_HOOKS", self.bundled), + patch.object(hooks, "_SOURCE_HOOKS", self.source), + ): + return hooks.hooks_source() + + def test_prefers_bundled(self): + self.bundled.mkdir() + self.source.mkdir() + self.assertEqual(self._resolve(), self.bundled) + + def test_falls_back_to_source(self): + self.source.mkdir() + self.assertEqual(self._resolve(), self.source) + + def test_none_when_neither_exists(self): + self.assertIsNone(self._resolve()) + + +if __name__ == "__main__": + unittest.main()