diff --git a/claude-code/gateway/setup.py b/claude-code/gateway/setup.py index 32ffa69..660d72f 100644 --- a/claude-code/gateway/setup.py +++ b/claude-code/gateway/setup.py @@ -283,9 +283,21 @@ def write_unbound_config(api_key: str, urls: dict = None) -> bool: return False -def remove_hooks_unbound_script() -> None: - """Remove ~/.claude/hooks/unbound.py if present (leftover from hooks setup).""" - script_path = Path.home() / ".claude" / "hooks" / "unbound.py" +def _resolve_claude_config_dir(config_dir_arg: Optional[str] = None) -> Path: + """Resolve Claude Code's config dir: $CLAUDE_CONFIG_DIR (env wins), else the + --config-dir arg the CLI forwards, else the default ~/.claude.""" + value = (os.environ.get("CLAUDE_CONFIG_DIR") or "").strip() or None + if not value and config_dir_arg: + value = config_dir_arg.strip() or None + if not value: + return Path.home() / ".claude" + return Path(value).expanduser().resolve() + + +def remove_hooks_unbound_script(config_dir: Path = None) -> None: + """Remove /hooks/unbound.py if present (leftover from hooks setup).""" + config_dir = config_dir or (Path.home() / ".claude") + script_path = config_dir / "hooks" / "unbound.py" if script_path.exists(): try: script_path.unlink() @@ -294,12 +306,12 @@ def remove_hooks_unbound_script() -> None: debug_print(f"Failed to remove {script_path}: {e}") -def setup_claude_key_helper() -> bool: +def setup_claude_key_helper(config_dir: Path = None) -> bool: """ - Create ~/.claude/anthropic_key.sh that echoes UNBOUND_API_KEY and - update ~/.claude/settings.json with apiKeyHelper pointing to that script. + Create /anthropic_key.sh that echoes UNBOUND_API_KEY and + update /settings.json with apiKeyHelper pointing to that script. """ - claude_dir = Path.home() / ".claude" + claude_dir = config_dir or (Path.home() / ".claude") settings_path = claude_dir / "settings.json" key_helper_path = claude_dir / "anthropic_key.sh" @@ -326,8 +338,10 @@ def setup_claude_key_helper() -> bool: if "hooks" in settings: del settings["hooks"] - # Update apiKeyHelper - settings["apiKeyHelper"] = "~/.claude/anthropic_key.sh" + if claude_dir.resolve() == (Path.home() / ".claude").resolve(): + settings["apiKeyHelper"] = "~/.claude/anthropic_key.sh" + else: + settings["apiKeyHelper"] = str(key_helper_path) settings_path.write_text(json.dumps(settings, indent=2), encoding="utf-8") return True @@ -428,12 +442,13 @@ def _clear_path(path: Path, label: str) -> str: return "failed" -def remove_api_key_helper_setting() -> str: +def remove_api_key_helper_setting(config_dir: Path = None) -> str: """Remove apiKeyHelper from settings.json. Returns "cleared", "not_found", or "failed". """ - settings_path = Path.home() / ".claude" / "settings.json" + config_dir = config_dir or (Path.home() / ".claude") + settings_path = config_dir / "settings.json" if not settings_path.exists(): return "not_found" try: @@ -451,8 +466,9 @@ def remove_api_key_helper_setting() -> str: return "failed" -def clear_setup() -> bool: +def clear_setup(config_dir: Path = None) -> bool: """Undo all changes made by the setup script.""" + config_dir = config_dir or (Path.home() / ".claude") print("=" * 60) print("Claude Code - Clearing Setup") print("=" * 60) @@ -468,19 +484,29 @@ def clear_setup() -> bool: print(f"Failed to clear {label}") any_failed = True - _r = _clear_path(Path.home() / ".claude" / "anthropic_key.sh", "Claude anthropic_key.sh") + _r = _clear_path(config_dir / "anthropic_key.sh", "Claude anthropic_key.sh") if _r == "cleared": any_cleared = True elif _r == "failed": any_failed = True - settings_status = remove_api_key_helper_setting() + settings_status = remove_api_key_helper_setting(config_dir) if settings_status == "cleared": any_cleared = True elif settings_status == "failed": print("Failed to clear apiKeyHelper in settings.json") any_failed = True + # When the config dir was relocated, also strip enforcement left behind in the + # default ~/.claude so clearing leaves nothing that fires if Claude later runs + # without CLAUDE_CONFIG_DIR set. + default_dir = Path.home() / ".claude" + if config_dir.resolve() != default_dir.resolve(): + if _clear_path(default_dir / "anthropic_key.sh", "Claude anthropic_key.sh (~/.claude)") == "cleared": + any_cleared = True + if remove_api_key_helper_setting(default_dir) == "cleared": + any_cleared = True + if any_cleared: print("Cleared") elif not any_failed: @@ -593,12 +619,13 @@ def get_device_identifier() -> Optional[str]: return None -def detect_install_state() -> str: +def detect_install_state(config_dir: Path = None) -> str: """User-level install state (informational): 'persisted' if this tool's Unbound setup already exists on this device, else 'fresh'. User-level setups are never tamper-eligible, so 'tampered' is never reported.""" + config_dir = config_dir or (Path.home() / ".claude") try: - return "persisted" if (Path.home() / ".claude" / "anthropic_key.sh").exists() else "fresh" + return "persisted" if (config_dir / "anthropic_key.sh").exists() else "fresh" except Exception as e: debug_print(f"detect_install_state failed: {e}") return "fresh" @@ -671,16 +698,19 @@ def main(): parser.add_argument("--clear", action="store_true", help="Undo all changes made by the setup script") parser.add_argument("--debug", action="store_true", help="Show detailed debug information") parser.add_argument("--api-key", dest="api_key", help="API key (skip browser auth)") + parser.add_argument("--config-dir", dest="config_dir", help="Claude Code config dir (defaults to $CLAUDE_CONFIG_DIR or ~/.claude)") args, _ = parser.parse_known_args() args.gateway_url = normalize_url(args.gateway_url) args.backend_url = normalize_url(args.backend_url) + config_dir = _resolve_claude_config_dir(args.config_dir) + if args.debug: DEBUG = True debug_print("Debug mode enabled") if args.clear: - return clear_setup() + return clear_setup(config_dir) if check_enterprise_hooks_conflict(): print("\nāŒ Skipped — Claude Code is managed by your organization (MDM).") @@ -702,7 +732,7 @@ def main(): pass # Remove leftover hooks setup artifacts - remove_hooks_unbound_script() + remove_hooks_unbound_script(config_dir) api_key = args.api_key if not api_key: @@ -742,14 +772,14 @@ def main(): return False debug_print("ANTHROPIC_BASE_URL set successfully") - _install_state = detect_install_state() + _install_state = detect_install_state(config_dir) _device_id = get_device_identifier() write_unbound_config(api_key, urls={"base_url": args.backend_url, "gateway_url": args.gateway_url, "frontend_url": normalize_url(args.domain) if args.domain else None}) # Configure Claude Code helper files debug_print("Setting up Claude key helper...") - if not setup_claude_key_helper(): + if not setup_claude_key_helper(config_dir): return False debug_print("Claude key helper configured") diff --git a/claude-code/gateway/test_setup.py b/claude-code/gateway/test_setup.py new file mode 100644 index 0000000..6052579 --- /dev/null +++ b/claude-code/gateway/test_setup.py @@ -0,0 +1,105 @@ +import importlib.util +import json +import os +import tempfile +import unittest +from unittest import mock +from pathlib import Path + +_SPEC = importlib.util.spec_from_file_location( + "gateway_setup", os.path.join(os.path.dirname(__file__), "setup.py") +) +gw = importlib.util.module_from_spec(_SPEC) +_SPEC.loader.exec_module(gw) + + +class TestResolveClaudeConfigDir(unittest.TestCase): + + def test_env_wins(self): + with tempfile.TemporaryDirectory() as d: + with mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": d}): + self.assertEqual(gw._resolve_claude_config_dir(None), Path(d).resolve()) + + def test_env_takes_precedence_over_arg(self): + with tempfile.TemporaryDirectory() as d: + with mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": d}): + self.assertEqual(gw._resolve_claude_config_dir("/other/dir"), Path(d).resolve()) + + def test_arg_used_when_env_absent(self): + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop("CLAUDE_CONFIG_DIR", None) + self.assertEqual(gw._resolve_claude_config_dir("/opt/cc"), Path("/opt/cc").resolve()) + + def test_default_fallback(self): + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop("CLAUDE_CONFIG_DIR", None) + self.assertEqual(gw._resolve_claude_config_dir(None), Path.home() / ".claude") + + def test_blank_env_falls_back(self): + with mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": " "}): + self.assertEqual(gw._resolve_claude_config_dir(None), Path.home() / ".claude") + + +class TestKeyHelperUnderConfigDir(unittest.TestCase): + def test_custom_dir_writes_there_with_absolute_helper(self): + with tempfile.TemporaryDirectory() as d: + cc = Path(d) / "cc" + gw.setup_claude_key_helper(cc) + self.assertTrue((cc / "anthropic_key.sh").exists()) + settings = json.loads((cc / "settings.json").read_text()) + self.assertEqual(settings["apiKeyHelper"], str(cc / "anthropic_key.sh")) + + def test_default_dir_keeps_portable_helper(self): + with tempfile.TemporaryDirectory() as home: + with mock.patch.object(gw.Path, "home", staticmethod(lambda: Path(home))): + default_dir = Path(home) / ".claude" + gw.setup_claude_key_helper(default_dir) + settings = json.loads((default_dir / "settings.json").read_text()) + self.assertEqual(settings["apiKeyHelper"], "~/.claude/anthropic_key.sh") + + def test_detect_install_state_honors_config_dir(self): + with tempfile.TemporaryDirectory() as d: + cc = Path(d) / "cc" + self.assertEqual(gw.detect_install_state(cc), "fresh") + gw.setup_claude_key_helper(cc) + self.assertEqual(gw.detect_install_state(cc), "persisted") + + def test_apikeyhelper_portable_when_dir_equals_default_via_realpath(self): + # Passing the default dir (even pre-resolution) must still yield the + # portable ~/.claude form, not an absolute realpath. + with tempfile.TemporaryDirectory() as home: + with mock.patch.object(gw.Path, "home", staticmethod(lambda: Path(home))): + gw.setup_claude_key_helper(Path(home) / ".claude") + settings = json.loads((Path(home) / ".claude" / "settings.json").read_text()) + self.assertEqual(settings["apiKeyHelper"], "~/.claude/anthropic_key.sh") + + +class TestClearSweepsLegacyDir(unittest.TestCase): + def test_clear_relocated_also_clears_default_claude(self): + with tempfile.TemporaryDirectory() as home: + home = Path(home) + with mock.patch.object(gw.Path, "home", staticmethod(lambda: home)): + legacy = home / ".claude" + legacy.mkdir(parents=True) + (legacy / "anthropic_key.sh").write_text("echo x") + (legacy / "settings.json").write_text(json.dumps({"apiKeyHelper": "~/.claude/anthropic_key.sh"})) + cc = home / "cc" + gw.setup_claude_key_helper(cc) + gw.clear_setup(cc) + # active dir cleared + self.assertFalse((cc / "anthropic_key.sh").exists()) + # legacy ~/.claude swept too + self.assertFalse((legacy / "anthropic_key.sh").exists()) + self.assertNotIn("apiKeyHelper", json.loads((legacy / "settings.json").read_text())) + + def test_clear_default_dir_does_not_double_sweep(self): + with tempfile.TemporaryDirectory() as home: + home = Path(home) + with mock.patch.object(gw.Path, "home", staticmethod(lambda: home)): + gw.setup_claude_key_helper(home / ".claude") + gw.clear_setup(home / ".claude") + self.assertFalse((home / ".claude" / "anthropic_key.sh").exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/claude-code/hooks/setup.py b/claude-code/hooks/setup.py index bfe5d75..65f2770 100644 --- a/claude-code/hooks/setup.py +++ b/claude-code/hooks/setup.py @@ -62,6 +62,18 @@ def normalize_url(domain: str) -> str: return url.rstrip('/') +def _resolve_claude_config_dir(argv) -> Path: + value = (os.environ.get("CLAUDE_CONFIG_DIR") or "").strip() or None + if not value: + for i, arg in enumerate(argv): + if arg == "--config-dir" and i + 1 < len(argv) and not argv[i + 1].startswith("--"): + value = argv[i + 1].strip() or None + break + if not value: + return Path.home() / ".claude" + return Path(value).expanduser().resolve() + + def get_shell_rc_file() -> Path: system = platform.system().lower() shell = os.environ.get("SHELL", "").lower() @@ -309,9 +321,10 @@ def write_unbound_config(api_key: str, urls: dict = None) -> bool: return False -def remove_gateway_artifacts() -> None: - """Remove ~/.claude/anthropic_key.sh if present (leftover from gateway setup).""" - key_helper_path = Path.home() / ".claude" / "anthropic_key.sh" +def remove_gateway_artifacts(config_dir: Path = None) -> None: + """Remove anthropic_key.sh if present (leftover from gateway setup).""" + config_dir = config_dir or (Path.home() / ".claude") + key_helper_path = config_dir / "anthropic_key.sh" if key_helper_path.exists(): try: key_helper_path.unlink() @@ -351,8 +364,9 @@ def rewrite_gateway_url_in_file(path: Path, gateway_url: str) -> None: debug_print(f"Could not rewrite gateway URL in {path}: {e}") -def setup_hooks(gateway_url: str = DEFAULT_GATEWAY_URL): - hooks_dir = Path.home() / ".claude" / "hooks" +def setup_hooks(gateway_url: str = DEFAULT_GATEWAY_URL, config_dir: Path = None): + config_dir = config_dir or (Path.home() / ".claude") + hooks_dir = config_dir / "hooks" script_path = hooks_dir / "unbound.py" # print("\nšŸ“„ Downloading unbound.py script...") @@ -396,9 +410,10 @@ def _command_targets_hook(command: str, target: Path) -> bool: return os.path.normcase(os.path.normpath(tokens[0])) == normalized_target -def configure_claude_settings() -> bool: - settings_path = Path.home() / ".claude" / "settings.json" - +def configure_claude_settings(config_dir: Path = None) -> bool: + config_dir = config_dir or (Path.home() / ".claude") + settings_path = config_dir / "settings.json" + try: if settings_path.exists(): with open(settings_path, 'r', encoding='utf-8') as f: @@ -411,7 +426,7 @@ def configure_claude_settings() -> bool: if "apiKeyHelper" in settings: del settings["apiKeyHelper"] - script_path = Path.home() / ".claude" / "hooks" / "unbound.py" + script_path = config_dir / "hooks" / "unbound.py" # On Windows, invoke via the launcher and quote the path (handles spaces # like C:\Users\Jane Doe\ or C:\Program Files\). Use `py -3` if present, @@ -543,13 +558,14 @@ def _hook(entry: dict) -> dict: return False -def remove_hooks_from_settings() -> str: +def remove_hooks_from_settings(config_dir: Path = None) -> str: """Remove the unbound hooks from settings.json. Returns "cleared", "not_found", or "failed". """ - settings_path = Path.home() / ".claude" / "settings.json" - script_path = Path.home() / ".claude" / "hooks" / "unbound.py" + config_dir = config_dir or (Path.home() / ".claude") + settings_path = config_dir / "settings.json" + script_path = config_dir / "hooks" / "unbound.py" if not settings_path.exists(): return "not_found" @@ -612,8 +628,9 @@ def _clear_path(path: Path, label: str) -> str: return "failed" -def clear_setup() -> bool: +def clear_setup(config_dir: Path = None) -> bool: """Undo all changes made by the setup script.""" + config_dir = config_dir or (Path.home() / ".claude") print("=" * 60) print("Claude Code Hooks - Clearing Setup") print("=" * 60) @@ -628,15 +645,15 @@ def clear_setup() -> bool: print("Failed to clear API_KEY") any_failed = True - _r = _clear_path(Path.home() / ".claude" / "hooks" / "unbound.py", "Claude unbound.py hook") + _r = _clear_path(config_dir / "hooks" / "unbound.py", "Claude unbound.py hook") if _r == "cleared": any_cleared = True elif _r == "failed": any_failed = True for extra in ( - Path.home() / ".claude" / "hooks" / "unbound-setup.py", - Path.home() / ".claude" / "hooks" / ".last_updated", + config_dir / "hooks" / "unbound-setup.py", + config_dir / "hooks" / ".last_updated", ): _r = _clear_path(extra, str(extra)) if _r == "cleared": @@ -644,13 +661,23 @@ def clear_setup() -> bool: elif _r == "failed": any_failed = True - settings_status = remove_hooks_from_settings() + settings_status = remove_hooks_from_settings(config_dir) if settings_status == "cleared": any_cleared = True elif settings_status == "failed": print("Failed to clear Unbound hooks in settings.json") any_failed = True + # When the config dir was relocated, also strip enforcement left behind in the + # default ~/.claude so clearing leaves nothing that fires if Claude later runs + # without CLAUDE_CONFIG_DIR set. + default_dir = Path.home() / ".claude" + if config_dir.resolve() != default_dir.resolve(): + if _clear_path(default_dir / "hooks" / "unbound.py", "Claude unbound.py hook (~/.claude)") == "cleared": + any_cleared = True + if remove_hooks_from_settings(default_dir) == "cleared": + any_cleared = True + if any_cleared: print("Cleared") elif not any_failed: @@ -762,12 +789,13 @@ def get_device_identifier() -> Optional[str]: return None -def detect_install_state() -> str: +def detect_install_state(config_dir: Path = None) -> str: """User-level install state (informational): 'persisted' if this tool's Unbound setup already exists on this device, else 'fresh'. User-level setups are never tamper-eligible, so 'tampered' is never reported.""" + config_dir = config_dir or (Path.home() / ".claude") try: - return "persisted" if (Path.home() / ".claude" / "hooks" / "unbound.py").exists() else "fresh" + return "persisted" if (config_dir / "hooks" / "unbound.py").exists() else "fresh" except Exception as e: debug_print(f"detect_install_state failed: {e}") return "fresh" @@ -965,16 +993,16 @@ def _backfill_upload_chunk(api_key: str, backend_url: str, sessions: List[Dict]) return True -def _backfill_state_path(home: Path) -> Path: - return home / '.claude' / 'hooks' / BACKFILL_STATE_FILE +def _backfill_state_path(config_dir: Path) -> Path: + return config_dir / 'hooks' / BACKFILL_STATE_FILE -def _backfill_read_cutoff(home: Path) -> float: +def _backfill_read_cutoff(config_dir: Path) -> float: """mtime cutoff for transcript selection: the last successful backfill when cached (so cron reruns only seed sessions touched since), else 30 days ago.""" default_cutoff = time.time() - (BACKFILL_MAX_AGE_DAYS * 86400) try: - last = float(_backfill_state_path(home).read_text().strip()) + last = float(_backfill_state_path(config_dir).read_text().strip()) except (OSError, ValueError): return default_cutoff # Ignore corrupt or future timestamps (clock skew). @@ -983,11 +1011,11 @@ def _backfill_read_cutoff(home: Path) -> float: return last -def _backfill_write_cutoff(home: Path, ts: float) -> None: +def _backfill_write_cutoff(config_dir: Path, ts: float) -> None: # Write via temp + atomic replace so an overlapping cron run never reads a # half-written timestamp. try: - path = _backfill_state_path(home) + path = _backfill_state_path(config_dir) path.parent.mkdir(parents=True, exist_ok=True) tmp = path.parent / f'{path.name}.{os.getpid()}.tmp' tmp.write_text(str(ts)) @@ -1107,17 +1135,18 @@ def _backfill_slice_session(session: Dict, max_chunk_bytes: int): start_idx = last_fit_end -def run_backfill(api_key: str, backend_url: str) -> None: - """Walk ~/.claude/projects and seed historical sessions. Never raises.""" +def run_backfill(api_key: str, backend_url: str, config_dir: Path = None) -> None: + """Walk config_dir/projects and seed historical sessions. Never raises.""" if os.environ.get('UNBOUND_BACKFILL_DISABLED') == '1': debug_print("UNBOUND_BACKFILL_DISABLED=1 — skipping backfill") return try: - home = Path.home() + if config_dir is None: + config_dir = Path.home() / '.claude' started_at = time.time() - cutoff_mtime = _backfill_read_cutoff(home) - projects_root = home / '.claude' / 'projects' + cutoff_mtime = _backfill_read_cutoff(config_dir) + projects_root = config_dir / 'projects' sessions: List[Dict] = [] capped = False if projects_root.exists(): @@ -1132,7 +1161,7 @@ def run_backfill(api_key: str, backend_url: str) -> None: if session: sessions.append(session) if not sessions: - _backfill_write_cutoff(home, started_at) + _backfill_write_cutoff(config_dir, started_at) print("[backfill] No past sessions found.") return @@ -1179,7 +1208,7 @@ def _flush(): print(f"[backfill] Done — queued {sessions_sent} past sessions ({failed} chunks failed).") else: if not capped: - _backfill_write_cutoff(home, started_at) + _backfill_write_cutoff(config_dir, started_at) print(f"[backfill] Done — queued {sessions_sent} past sessions for processing.") except Exception as e: print(f"[backfill] Skipped due to error: {e}", file=sys.stderr) @@ -1197,8 +1226,10 @@ def main(): DEBUG = True debug_print("Debug mode enabled") + config_dir = _resolve_claude_config_dir(sys.argv) + if clear_mode: - return clear_setup() + return clear_setup(config_dir) if check_enterprise_hooks_conflict(): print("\nāŒ Skipped — Claude Code is managed by your organization (MDM).") @@ -1269,7 +1300,7 @@ def main(): remove_env_var(var_name) except Exception: pass - remove_gateway_artifacts() + remove_gateway_artifacts(config_dir) debug_print("Setting UNBOUND_CLAUDE_API_KEY environment variable...") success, message = set_env_var("UNBOUND_CLAUDE_API_KEY", api_key) @@ -1278,19 +1309,19 @@ def main(): return False debug_print("UNBOUND_CLAUDE_API_KEY set successfully") - _install_state = detect_install_state() + _install_state = detect_install_state(config_dir) _device_id = get_device_identifier() write_unbound_config(api_key, urls={"base_url": backend_url, "gateway_url": gateway_url, "frontend_url": normalize_url(domain) if domain else None}) debug_print("Setting up hooks...") - if not setup_hooks(gateway_url=gateway_url): + if not setup_hooks(gateway_url=gateway_url, config_dir=config_dir): print("āŒ Failed to setup hooks") return False debug_print("Hooks downloaded successfully") debug_print("Configuring Claude settings...") - if not configure_claude_settings(): + if not configure_claude_settings(config_dir=config_dir): print("āŒ Failed to configure Claude settings") return False debug_print("Claude settings configured successfully") @@ -1302,7 +1333,7 @@ def main(): notify_setup_complete(api_key, "claude-code", backend_url=backend_url, install_state=_install_state, serial_number=_device_id) if backfill_mode: - run_backfill(api_key, backend_url) + run_backfill(api_key, backend_url, config_dir) rc_path = get_shell_rc_file() if rc_path is not None: diff --git a/claude-code/hooks/test_setup.py b/claude-code/hooks/test_setup.py index b66e70d..6c255d2 100644 --- a/claude-code/hooks/test_setup.py +++ b/claude-code/hooks/test_setup.py @@ -146,12 +146,13 @@ class TestBackfillCutoffCache(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.home = Path(self._tmp.name) + self.config_dir = self.home / ".claude" self.addCleanup(self._tmp.cleanup) def test_read_cutoff_defaults_to_max_age_when_no_file(self): """No cache file -> fall back to BACKFILL_MAX_AGE_DAYS ago (first run).""" import setup - cutoff = setup._backfill_read_cutoff(self.home) + cutoff = setup._backfill_read_cutoff(self.config_dir) expected = time.time() - (setup.BACKFILL_MAX_AGE_DAYS * 86400) self.assertAlmostEqual(cutoff, expected, delta=5) @@ -159,25 +160,25 @@ def test_write_then_read_roundtrip(self): """A persisted timestamp is read back as the cutoff on the next run.""" import setup ts = time.time() - 3600 - setup._backfill_write_cutoff(self.home, ts) - self.assertTrue(setup._backfill_state_path(self.home).exists()) - self.assertAlmostEqual(setup._backfill_read_cutoff(self.home), ts, delta=0.01) + setup._backfill_write_cutoff(self.config_dir, ts) + self.assertTrue(setup._backfill_state_path(self.config_dir).exists()) + self.assertAlmostEqual(setup._backfill_read_cutoff(self.config_dir), ts, delta=0.01) def test_read_cutoff_ignores_corrupt_value(self): """A non-numeric cache file falls back to the default window.""" import setup - path = setup._backfill_state_path(self.home) + path = setup._backfill_state_path(self.config_dir) path.parent.mkdir(parents=True, exist_ok=True) path.write_text("not-a-number") expected = time.time() - (setup.BACKFILL_MAX_AGE_DAYS * 86400) - self.assertAlmostEqual(setup._backfill_read_cutoff(self.home), expected, delta=5) + self.assertAlmostEqual(setup._backfill_read_cutoff(self.config_dir), expected, delta=5) def test_read_cutoff_ignores_future_timestamp(self): """A future timestamp (clock skew) is rejected for the default window.""" import setup - setup._backfill_write_cutoff(self.home, time.time() + 10000) + setup._backfill_write_cutoff(self.config_dir, time.time() + 10000) expected = time.time() - (setup.BACKFILL_MAX_AGE_DAYS * 86400) - self.assertAlmostEqual(setup._backfill_read_cutoff(self.home), expected, delta=5) + self.assertAlmostEqual(setup._backfill_read_cutoff(self.config_dir), expected, delta=5) def test_iter_transcripts_respects_cutoff(self): """Only transcripts modified at/after the cutoff are yielded.""" @@ -199,8 +200,8 @@ def test_iter_transcripts_respects_cutoff(self): def test_write_is_atomic_and_leaves_no_temp(self): """The atomic write produces the final file and no leftover .tmp.""" import setup - setup._backfill_write_cutoff(self.home, 123.0) - path = setup._backfill_state_path(self.home) + setup._backfill_write_cutoff(self.config_dir, 123.0) + path = setup._backfill_state_path(self.config_dir) self.assertEqual(path.read_text(), "123.0") self.assertEqual(list(path.parent.glob("*.tmp")), []) @@ -208,15 +209,27 @@ def test_cutoff_not_advanced_when_session_cap_fires(self): """When the per-run session cap is hit, the cutoff must NOT advance, or the unprocessed older files would be skipped forever next run.""" import setup - root = self.home / ".claude" / "projects" + root = self.config_dir / "projects" root.mkdir(parents=True) for i in range(3): (root / f"s{i}.jsonl").write_text('{"sessionId":"x%d"}\n' % i) with patch.object(setup, "BACKFILL_MAX_SESSIONS_PER_RUN", 2), \ - patch.object(setup, "_backfill_upload_chunk", return_value=True), \ - patch.object(Path, "home", return_value=self.home): - setup.run_backfill("key", "https://backend") - self.assertFalse(setup._backfill_state_path(self.home).exists()) + patch.object(setup, "_backfill_upload_chunk", return_value=True): + setup.run_backfill("key", "https://backend", self.config_dir) + self.assertFalse(setup._backfill_state_path(self.config_dir).exists()) + + def test_run_backfill_reads_custom_config_dir_projects(self): + """With a custom config_dir, backfill walks config_dir/projects and writes + the cutoff there — not under ~/.claude.""" + import setup + custom = self.home / "custom-cc" + root = custom / "projects" + root.mkdir(parents=True) + (root / "s.jsonl").write_text('{"sessionId":"x"}\n') + with patch.object(setup, "_backfill_upload_chunk", return_value=True): + setup.run_backfill("key", "https://backend", custom) + self.assertTrue(setup._backfill_state_path(custom).exists()) + self.assertFalse(setup._backfill_state_path(self.config_dir).exists()) class TestMdmBackfillCutoff(unittest.TestCase): @@ -360,6 +373,86 @@ def fake_run_as_user(username, fn, *args, **kwargs): ) +class TestResolveClaudeConfigDir(unittest.TestCase): + """CLAUDE_CONFIG_DIR env > --config-dir arg > ~/.claude.""" + + def test_env_beats_arg_and_home(self): + import setup + with patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": "/env/cc"}): + result = setup._resolve_claude_config_dir(["x", "--config-dir", "/arg/cc"]) + self.assertEqual(result, Path("/env/cc").resolve()) + + def test_arg_used_when_no_env(self): + import setup + env = {k: v for k, v in os.environ.items() if k != "CLAUDE_CONFIG_DIR"} + with patch.dict(os.environ, env, clear=True): + result = setup._resolve_claude_config_dir(["x", "--config-dir", "/arg/cc"]) + self.assertEqual(result, Path("/arg/cc").resolve()) + + def test_env_used_when_no_arg(self): + import setup + with patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": "/env/cc"}): + result = setup._resolve_claude_config_dir(["x"]) + self.assertEqual(result, Path("/env/cc").resolve()) + + def test_home_default_when_arg_and_env_absent(self): + import setup + env = {k: v for k, v in os.environ.items() if k != "CLAUDE_CONFIG_DIR"} + with patch.dict(os.environ, env, clear=True): + result = setup._resolve_claude_config_dir(["x"]) + self.assertEqual(result, Path.home() / ".claude") + + def test_relative_value_is_absolutized(self): + import setup + result = setup._resolve_claude_config_dir(["x", "--config-dir", "rel/cc"]) + self.assertEqual(result, Path("rel/cc").resolve()) + + +class TestInstallUnderResolvedDir(unittest.TestCase): + """Hooks + settings + baked command must land under the resolved config dir.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.home = Path(self.tmp) / "home" + self.home.mkdir(parents=True) + self.config_dir = Path(self.tmp) / "custom-cc" + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_settings_and_hook_command_under_config_dir(self): + import setup + with patch.object(Path, "home", staticmethod(lambda: self.home)), \ + patch.object(setup, "download_file", lambda url, dest: dest.parent.mkdir(parents=True, exist_ok=True) or dest.write_text("# hook") or True): + self.assertTrue(setup.setup_hooks(config_dir=self.config_dir)) + self.assertTrue(setup.configure_claude_settings(config_dir=self.config_dir)) + + hook_path = self.config_dir / "hooks" / "unbound.py" + settings_path = self.config_dir / "settings.json" + self.assertTrue(hook_path.exists()) + self.assertTrue(settings_path.exists()) + settings = json.loads(settings_path.read_text()) + cmd = settings["hooks"]["PreToolUse"][0]["hooks"][0]["command"] + self.assertEqual(cmd, str(hook_path)) + self.assertNotIn(str(self.home / ".claude"), cmd) + + def test_backward_compat_no_env_uses_home_claude(self): + import setup + env = {k: v for k, v in os.environ.items() if k != "CLAUDE_CONFIG_DIR"} + with patch.dict(os.environ, env, clear=True), \ + patch.object(Path, "home", staticmethod(lambda: self.home)), \ + patch.object(setup, "download_file", lambda url, dest: dest.parent.mkdir(parents=True, exist_ok=True) or dest.write_text("# hook") or True): + config_dir = setup._resolve_claude_config_dir(["x"]) + self.assertTrue(setup.setup_hooks(config_dir=config_dir)) + self.assertTrue(setup.configure_claude_settings(config_dir=config_dir)) + + hook_path = self.home / ".claude" / "hooks" / "unbound.py" + self.assertTrue(hook_path.exists()) + settings = json.loads((self.home / ".claude" / "settings.json").read_text()) + cmd = settings["hooks"]["PreToolUse"][0]["hooks"][0]["command"] + self.assertEqual(cmd, str(hook_path)) + + class TestCommandTargetsHook(unittest.TestCase): def setUp(self): from setup import _command_targets_hook diff --git a/claude-code/hooks/unbound.py b/claude-code/hooks/unbound.py index 8117e73..0cf19fa 100644 --- a/claude-code/hooks/unbound.py +++ b/claude-code/hooks/unbound.py @@ -18,22 +18,35 @@ UNBOUND_GATEWAY_URL = os.environ.get( "UNBOUND_GATEWAY_URL", "https://api.getunbound.ai" ).rstrip("/") -AUDIT_LOG = Path.home() / ".claude" / "hooks" / "agent-audit.log" -ERROR_LOG = Path.home() / ".claude" / "hooks" / "error.log" -LAST_REPORT_FILE = Path.home() / ".claude" / "hooks" / ".last_error_report" +_env_config_dir = (os.environ.get("CLAUDE_CONFIG_DIR") or "").strip() +_config_dir_is_default = not _env_config_dir +_CONFIG_DIR = Path(_env_config_dir or (Path.home() / ".claude")).expanduser().resolve() +AUDIT_LOG = _CONFIG_DIR / "hooks" / "agent-audit.log" +ERROR_LOG = _CONFIG_DIR / "hooks" / "error.log" +LAST_REPORT_FILE = _CONFIG_DIR / "hooks" / ".last_error_report" ALLOWED_NON_MCP_HOOK_NAMES = ['Bash', 'Read', 'Write', 'Edit'] # MCP tools (mcp__*) are always checked separately NATIVE_FILE_TOOLS = {'Read', 'Write', 'Edit'} MCP_TOOL_PREFIX = 'mcp__' + +def _relocated_or_legacy(relocated: Path, legacy: Path) -> Path: + # Whether Claude relocates .claude.json / plugins under CLAUDE_CONFIG_DIR is + # version-dependent, so read from the relocated dir when it actually has the + # artifact, else the default ~/.claude location. Keeps both paths consistent. + if not _config_dir_is_default and relocated.exists(): + return relocated + return legacy + + # CoWork built-in tools that are exposed under mcp__ COWORK_BUILTIN_MCP_SERVERS = frozenset({ 'workspace', 'cowork', 'cowork-onboarding', 'visualize', 'scheduled-tasks', 'plugins', 'mcp-registry', 'session_info', 'skills', }) -CLAUDE_MCP_CONFIG_PATH = Path.home() / ".claude.json" -CLAUDE_PLUGIN_CACHE_DIR = Path.home() / ".claude" / "plugins" / "cache" -POLICY_CACHE_FILE = Path.home() / ".claude" / "hooks" / ".policy_cache.json" +CLAUDE_MCP_CONFIG_PATH = _relocated_or_legacy(_CONFIG_DIR / ".claude.json", Path.home() / ".claude.json") +CLAUDE_PLUGIN_CACHE_DIR = _relocated_or_legacy(_CONFIG_DIR / "plugins" / "cache", Path.home() / ".claude" / "plugins" / "cache") +POLICY_CACHE_FILE = _CONFIG_DIR / "hooks" / ".policy_cache.json" CACHE_TTL_SECONDS = 300 POLICY_CHECK_FAILURE_DEFAULT = 'allow' POLICY_CHECK_FAILURE_BLOCK_REASON = 'policy engine unavailable — please retry' @@ -62,7 +75,7 @@ SELF_UPDATE_INTERVAL_SECONDS = 2 * 3600 SELF_UPDATE_LOCK_TTL_SECONDS = 30 SELF_UPDATE_CURL_TIMEOUT = 10 -SELF_SCRIPT_PATH = Path.home() / ".claude" / "hooks" / "unbound.py" +SELF_SCRIPT_PATH = _CONFIG_DIR / "hooks" / "unbound.py" SELF_UPDATE_STATE_PATH = SELF_SCRIPT_PATH.parent / ".self_update_check" SELF_UPDATE_LOCK_PATH = SELF_SCRIPT_PATH.parent / ".self_update.lock" @@ -256,7 +269,7 @@ def append_to_audit_log(event_data: Dict): pass -_APPROVAL_MARKER_FILE = Path.home() / ".claude" / "hooks" / ".approval_pending" +_APPROVAL_MARKER_FILE = _CONFIG_DIR / "hooks" / ".approval_pending" def _is_approval_retry(command: str) -> bool: