diff --git a/augment/hooks/unbound.py b/augment/hooks/unbound.py index 63cb562..7eb9cd5 100644 --- a/augment/hooks/unbound.py +++ b/augment/hooks/unbound.py @@ -678,6 +678,121 @@ def _mangle_mcp_token(s: Optional[str]) -> str: return re.sub(r'[^A-Za-z0-9_-]', '_', s or '') +# KEEP IN SYNC: coding-discovery-tool mcp_tools_cache.py + all 5 hook copies — byte-identical, do not diverge. + +_MCP_TOOLS_CACHE_FILENAME = 'mcp-tools-cache.json' +_MCP_TOOLS_CACHE_MAX_BYTES = 2 * 1024 * 1024 +_MCP_CACHE_CODING_TOOL_NAMES = frozenset({'auggie cli'}) +_MCP_CACHE_CODING_TOOL_PREFIXES = ('augment (',) +_UNBOUND_CODING_TOOL = 'Auggie CLI' + + +def compute_mcp_cache_key(name, command, url, args): + subset = {} + clean_name = name.strip() if isinstance(name, str) else '' + if clean_name: + subset['name'] = clean_name + clean_url = url.strip() if isinstance(url, str) else '' + if clean_url: + subset['url'] = clean_url + clean_command = command.strip() if isinstance(command, str) else '' + if clean_command: + subset['command'] = clean_command + if args: + subset['args'] = args + if not subset: + return None + encoded = json.dumps(subset, sort_keys=True, separators=(',', ':')) + return hashlib.sha256(encoded.encode('utf-8')).hexdigest() + + +def _unbound_state_dir_candidates(): + candidates = [Path.home() / '.unbound'] + if hasattr(os, 'getuid'): + candidates.append(Path(f'/var/tmp/unbound-{os.getuid()}')) + else: + candidates.append(Path(tempfile.gettempdir()) / 'unbound') + return candidates + + +def _read_mcp_tools_cache(): + try: + for state_dir in _unbound_state_dir_candidates(): + path = state_dir / _MCP_TOOLS_CACHE_FILENAME + if not path.is_file(): + continue + with open(path, 'rb') as f: + data = f.read(_MCP_TOOLS_CACHE_MAX_BYTES + 1) + if len(data) > _MCP_TOOLS_CACHE_MAX_BYTES: + return {} + parsed = json.loads(data.decode('utf-8')) + return parsed if isinstance(parsed, dict) else {} + except Exception: + pass + return {} + + +def _mcp_cache_entries_for_user(tools): + username = Path.home().name + entries = [] + for key, by_user in tools.items(): + if not isinstance(key, str) or not isinstance(by_user, dict): + continue + k = key.strip().lower() + if k in _MCP_CACHE_CODING_TOOL_NAMES or k.startswith(_MCP_CACHE_CODING_TOOL_PREFIXES): + entry = by_user.get(username) + if isinstance(entry, dict): + entries.append(entry) + return entries + + +_CONTENT_HASH_RE = re.compile(r'^[a-f0-9]{64}$', re.IGNORECASE) + + +def _lookup_tool_content_hash(server_name, mcp_tool, server_cfg): + try: + if not server_name or not mcp_tool or not isinstance(server_cfg, dict): + return None + cache_key = compute_mcp_cache_key( + name=server_name, + command=server_cfg.get('command'), + url=server_cfg.get('url'), + args=server_cfg.get('args'), + ) + if not cache_key: + return None + tools = _read_mcp_tools_cache().get('tools') + if not isinstance(tools, dict): + return None + for entry in _mcp_cache_entries_for_user(tools): + by_tool = entry.get(cache_key) + if not isinstance(by_tool, dict): + continue + content_hash = by_tool.get(mcp_tool) + if isinstance(content_hash, str) and _CONTENT_HASH_RE.match(content_hash): + return content_hash + return None + except Exception: + return None + + +def _attach_tool_content_hash(metadata): + try: + server_cfg = metadata.get('mcp_server_config') + if not isinstance(server_cfg, dict): + return + content_hash = _lookup_tool_content_hash( + metadata.get('mcp_server'), metadata.get('mcp_tool'), server_cfg + ) + if content_hash: + server_cfg['tool_content_hash'] = content_hash + except Exception: + pass + + +# ───────────────────────── end MCP tool risk-scoring section ───────────────── + + # ── Cross-surface Augment MCP config resolution ────────────────────────────── # Augment does NOT embed the server in the MCP tool_name (no `mcp__server__tool`) # and the VS Code extension sends mcp_metadata=null, so an MCP call arrives as a @@ -1215,6 +1330,7 @@ def process_pre_tool_use(event: Dict, api_key: str) -> Dict: metadata['mcp_tool'] = mcp_tool_name if mcp_cfg: metadata['mcp_server_config'] = mcp_cfg + _attach_tool_content_hash(metadata) approval_key = f"{tool_name}:{command}" is_retry = _is_approval_retry(approval_key) @@ -1852,6 +1968,7 @@ def _dispatch_mcp_server_scan(server_name: str, server_config: Dict) -> None: "UNBOUND_API_KEY": api_key, "UNBOUND_MCP_SERVER_JSON": json.dumps(server_config), "UNBOUND_MCP_SERVER_NAME": server_name, + "UNBOUND_CODING_TOOL": _UNBOUND_CODING_TOOL, "UNBOUND_MCP_DOMAIN": backend_url}, } if os.name == "nt": diff --git a/claude-code/hooks/test_tool_content_hash.py b/claude-code/hooks/test_tool_content_hash.py new file mode 100644 index 0000000..9fe9a5f --- /dev/null +++ b/claude-code/hooks/test_tool_content_hash.py @@ -0,0 +1,294 @@ +""" +Tests for the MCP tool risk-scoring section in claude-code/hooks/unbound.py: +the local mcp-tools-cache.json lookup that attaches `tool_content_hash` to the +outgoing mcp_server_config on MCP PreToolUse, and the single name-inclusive +config-hash cache keying shared with the discovery tool. + +The section is embedded identically in every hook variant; a drift-guard test +below asserts the copies stay byte-identical modulo the per-hook constants. +""" + +import hashlib +import json +import re +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +import unbound + +HASH_A = 'a' * 64 +SLACK_CFG = {'url': 'https://mcp.slack.com/mcp', 'type': 'http'} +SLACK_KEY = unbound.compute_mcp_cache_key( + name='slack', command=None, url='https://mcp.slack.com/mcp', args=None, +) +USER = Path.home().name # home_user convention: home-directory basename + + +def _config_hash(subset): + return hashlib.sha256( + json.dumps(subset, sort_keys=True, separators=(',', ':')).encode('utf-8') + ).hexdigest() + + +def _cache_payload(coding_tool='Claude Code', user=USER, cache_key=SLACK_KEY, + tool='post_message', content_hash=HASH_A): + return { + 'updated_at': '2026-07-13T00:00:00Z', + 'tools': {coding_tool: {user: {cache_key: {tool: content_hash}}}}, + } + + +def _metadata(server='slack', tool='post_message', cfg=None): + if cfg is None: + cfg = dict(SLACK_CFG) + return {'mcp_server': server, 'mcp_tool': tool, 'mcp_server_config': cfg} + + +class _CacheDirMixin(unittest.TestCase): + """Points the state-dir resolution at a temp dir the test controls.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.state_dir = Path(self._tmp.name) + patcher = patch.object( + unbound, '_unbound_state_dir_candidates', + return_value=[self.state_dir], + ) + patcher.start() + self.addCleanup(patcher.stop) + self.addCleanup(self._tmp.cleanup) + + def write_cache(self, payload): + path = self.state_dir / 'mcp-tools-cache.json' + path.write_text(payload if isinstance(payload, str) else json.dumps(payload)) + return path + + +class TestComputeMcpCacheKey(unittest.TestCase): + """The single name-inclusive config-hash contract (SPEC §6 v3), shared + byte-identically with coding-discovery-tool's mcp_tools_cache module.""" + + def test_name_only_server(self): + # Empty-config servers (connectors, claude.ai integrations, IDE + # builtins) key on their name alone — no pattern identities. + self.assertEqual( + unbound.compute_mcp_cache_key('claude.ai Atlassian', None, None, None), + _config_hash({'name': 'claude.ai Atlassian'}), + ) + + def test_name_and_command(self): + self.assertEqual( + unbound.compute_mcp_cache_key('srv', 'builtin', None, None), + _config_hash({'name': 'srv', 'command': 'builtin'}), + ) + + def test_full_config(self): + self.assertEqual( + unbound.compute_mcp_cache_key( + 'slack', 'npx', 'https://mcp.slack.com/mcp', ['-y', '@slack/mcp']), + _config_hash({'name': 'slack', 'url': 'https://mcp.slack.com/mcp', + 'command': 'npx', 'args': ['-y', '@slack/mcp']}), + ) + + def test_strings_stripped_and_empty_dropped(self): + self.assertEqual( + unbound.compute_mcp_cache_key(' slack ', ' ', ' https://a.example/m ', None), + _config_hash({'name': 'slack', 'url': 'https://a.example/m'}), + ) + + def test_canonical_json_form_pinned(self): + # sort_keys + compact separators over the non-empty subset. + key = unbound.compute_mcp_cache_key('s', 'node', 'https://a.example', ['x']) + expected = hashlib.sha256( + '{"args":["x"],"command":"node","name":"s","url":"https://a.example"}'.encode('utf-8') + ).hexdigest() + self.assertEqual(key, expected) + + def test_name_changes_the_key(self): + a = unbound.compute_mcp_cache_key('name-one', None, 'https://a.example', None) + b = unbound.compute_mcp_cache_key('name-two', None, 'https://a.example', None) + self.assertNotEqual(a, b) + + def test_all_empty_returns_none(self): + self.assertIsNone(unbound.compute_mcp_cache_key(None, None, None, None)) + self.assertIsNone(unbound.compute_mcp_cache_key('', '', '', [])) + self.assertIsNone(unbound.compute_mcp_cache_key(' ', ' ', ' ', None)) + + +class TestAttachToolContentHash(_CacheDirMixin): + def test_hit_attaches_hash(self): + self.write_cache(_cache_payload()) + md = _metadata() + unbound._attach_tool_content_hash(md) + self.assertEqual(md['mcp_server_config']['tool_content_hash'], HASH_A) + + def test_coding_tool_key_matched_case_insensitively(self): + self.write_cache(_cache_payload(coding_tool='CLAUDE CODE')) + md = _metadata() + unbound._attach_tool_content_hash(md) + self.assertEqual(md['mcp_server_config']['tool_content_hash'], HASH_A) + + def test_cowork_surface_matches_too(self): + self.write_cache(_cache_payload(coding_tool='Claude Cowork')) + md = _metadata() + unbound._attach_tool_content_hash(md) + self.assertEqual(md['mcp_server_config']['tool_content_hash'], HASH_A) + + def test_name_only_server_hit(self): + # Empty-config server (e.g. a connector): keyed on name alone. + self.write_cache(_cache_payload(cache_key=_config_hash({'name': 'Gmail'}))) + md = _metadata(server='Gmail', cfg={'additional_data': {'scope': 'claude-connector'}}) + unbound._attach_tool_content_hash(md) + self.assertEqual(md['mcp_server_config']['tool_content_hash'], HASH_A) + + def test_empty_server_name_attaches_nothing(self): + # No name and an empty config -> no cache key -> no field. + self.write_cache(_cache_payload()) + md = _metadata(server='', cfg={'type': 'http'}) + unbound._attach_tool_content_hash(md) + self.assertNotIn('tool_content_hash', md['mcp_server_config']) + + def test_miss_unknown_tool_omits_field(self): + self.write_cache(_cache_payload()) + md = _metadata(tool='other_tool') + unbound._attach_tool_content_hash(md) + self.assertNotIn('tool_content_hash', md['mcp_server_config']) + + def test_miss_unknown_cache_key_omits_field(self): + self.write_cache(_cache_payload(cache_key=_config_hash({'url': 'https://other.example'}))) + md = _metadata() + unbound._attach_tool_content_hash(md) + self.assertNotIn('tool_content_hash', md['mcp_server_config']) + + def test_miss_other_coding_tool_omits_field(self): + self.write_cache(_cache_payload(coding_tool='Cursor')) + md = _metadata() + unbound._attach_tool_content_hash(md) + self.assertNotIn('tool_content_hash', md['mcp_server_config']) + + def test_miss_other_user_omits_field(self): + self.write_cache(_cache_payload(user=USER + '-someone-else')) + md = _metadata() + unbound._attach_tool_content_hash(md) + self.assertNotIn('tool_content_hash', md['mcp_server_config']) + + def test_missing_cache_file_omits_field(self): + md = _metadata() + unbound._attach_tool_content_hash(md) + self.assertNotIn('tool_content_hash', md['mcp_server_config']) + + def test_corrupt_cache_file_is_a_miss_not_a_crash(self): + self.write_cache('{not json!!') + md = _metadata() + unbound._attach_tool_content_hash(md) + self.assertNotIn('tool_content_hash', md['mcp_server_config']) + + def test_non_dict_cache_json_is_a_miss(self): + self.write_cache('[1, 2, 3]') + md = _metadata() + unbound._attach_tool_content_hash(md) + self.assertNotIn('tool_content_hash', md['mcp_server_config']) + + def test_oversized_cache_file_is_a_miss(self): + payload = _cache_payload() + payload['padding'] = 'x' * (unbound._MCP_TOOLS_CACHE_MAX_BYTES + 1) + self.write_cache(payload) + md = _metadata() + unbound._attach_tool_content_hash(md) + self.assertNotIn('tool_content_hash', md['mcp_server_config']) + + def test_non_sha256_cached_value_not_attached(self): + self.write_cache(_cache_payload(content_hash='not-a-hash')) + md = _metadata() + unbound._attach_tool_content_hash(md) + self.assertNotIn('tool_content_hash', md['mcp_server_config']) + + def test_malformed_cache_shapes_are_a_miss(self): + for tools in ( + 'string', ['list'], + {'Claude Code': 'string'}, + {'Claude Code': {USER: 'string'}}, + {'Claude Code': {USER: {SLACK_KEY: 'string'}}}, + ): + self.write_cache({'tools': tools}) + md = _metadata() + unbound._attach_tool_content_hash(md) + self.assertNotIn('tool_content_hash', md['mcp_server_config']) + + def test_no_config_is_a_noop(self): + md = {'mcp_server': 'slack', 'mcp_tool': 'post_message'} + unbound._attach_tool_content_hash(md) # must not raise + self.assertNotIn('mcp_server_config', md) + + def test_internal_error_never_escapes(self): + self.write_cache(_cache_payload()) + md = _metadata() + with patch.object(unbound, '_lookup_tool_content_hash', side_effect=RuntimeError('boom')): + unbound._attach_tool_content_hash(md) # must not raise + self.assertNotIn('tool_content_hash', md['mcp_server_config']) + + +class TestDispatchPassesCodingTool(unittest.TestCase): + """SPEC §9: the single-server scan dispatch carries this hook's + discovery-report coding-tool name so the scanner's cache write lands under + a key the lookup matches.""" + + def test_env_contains_unbound_coding_tool(self): + with tempfile.TemporaryDirectory() as tmp: + config_path = Path(tmp) / 'config.json' + config_path.write_text(json.dumps( + {'api_key': 'k', 'base_url': 'https://backend.example'})) + fake_bin = Path(tmp) / 'unbound-discovery' + fake_bin.write_text('') + with patch.object(unbound, 'UNBOUND_CONFIG_PATH', config_path), \ + patch.object(unbound, 'RUNNING_FROZEN', True), \ + patch.object(unbound, 'FROZEN_DISCOVERY_BIN', str(fake_bin)), \ + patch.object(unbound.subprocess, 'Popen') as popen: + unbound._dispatch_mcp_server_scan('srv', {'url': 'https://a.example'}) + self.assertTrue(popen.called) + env = popen.call_args.kwargs['env'] + self.assertEqual(env['UNBOUND_CODING_TOOL'], 'Claude Code') + self.assertEqual(unbound._UNBOUND_CODING_TOOL, 'Claude Code') + + +class TestCrossHookSectionConsistency(unittest.TestCase): + """The risk-scoring section is embedded per hook (single-file self-update + constraint). It must stay byte-identical across variants, modulo the three + per-hook coding-tool constants.""" + + START = '# KEEP IN SYNC: coding-discovery-tool mcp_tools_cache.py + all 5 hook copies' + END = '# ───────────────────────── end MCP tool risk-scoring section ─────────────────' + HOOK_FILES = ( + 'claude-code/hooks/unbound.py', + 'codex/hooks/unbound.py', + 'copilot/hooks/unbound.py', + 'augment/hooks/unbound.py', + 'cursor/unbound.py', + ) + + def _section(self, text): + section = text[text.index(self.START):text.index(self.END)] + section = re.sub(r'_MCP_CACHE_CODING_TOOL_NAMES = .*', '', section) + section = re.sub(r'_MCP_CACHE_CODING_TOOL_PREFIXES = .*', '', section) + return re.sub(r'_UNBOUND_CODING_TOOL = .*', '', section) + + def test_sections_identical_across_hooks(self): + repo_root = Path(__file__).resolve().parents[2] + paths = [repo_root / rel for rel in self.HOOK_FILES] + missing = [p for p in paths if not p.is_file()] + if missing: + self.skipTest(f'hook files not found: {missing}') + base = self._section(paths[0].read_text(encoding='utf-8')) + for path in paths[1:]: + with self.subTest(hook=str(path)): + self.assertEqual( + self._section(path.read_text(encoding='utf-8')), base, + f'{path} drifted from {paths[0]} — keep the embedded ' + f'risk-scoring sections in sync across all hook variants', + ) + + +if __name__ == '__main__': + unittest.main() diff --git a/claude-code/hooks/unbound.py b/claude-code/hooks/unbound.py index c5fa43a..257216c 100644 --- a/claude-code/hooks/unbound.py +++ b/claude-code/hooks/unbound.py @@ -1042,6 +1042,121 @@ def _read_mcp_server_config(server_name: str, config_path: Path, cwd: Optional[s return None +# KEEP IN SYNC: coding-discovery-tool mcp_tools_cache.py + all 5 hook copies — byte-identical, do not diverge. + +_MCP_TOOLS_CACHE_FILENAME = 'mcp-tools-cache.json' +_MCP_TOOLS_CACHE_MAX_BYTES = 2 * 1024 * 1024 +_MCP_CACHE_CODING_TOOL_NAMES = frozenset({'claude code', 'claude cowork'}) +_MCP_CACHE_CODING_TOOL_PREFIXES = () +_UNBOUND_CODING_TOOL = 'Claude Code' + + +def compute_mcp_cache_key(name, command, url, args): + subset = {} + clean_name = name.strip() if isinstance(name, str) else '' + if clean_name: + subset['name'] = clean_name + clean_url = url.strip() if isinstance(url, str) else '' + if clean_url: + subset['url'] = clean_url + clean_command = command.strip() if isinstance(command, str) else '' + if clean_command: + subset['command'] = clean_command + if args: + subset['args'] = args + if not subset: + return None + encoded = json.dumps(subset, sort_keys=True, separators=(',', ':')) + return hashlib.sha256(encoded.encode('utf-8')).hexdigest() + + +def _unbound_state_dir_candidates(): + candidates = [Path.home() / '.unbound'] + if hasattr(os, 'getuid'): + candidates.append(Path(f'/var/tmp/unbound-{os.getuid()}')) + else: + candidates.append(Path(tempfile.gettempdir()) / 'unbound') + return candidates + + +def _read_mcp_tools_cache(): + try: + for state_dir in _unbound_state_dir_candidates(): + path = state_dir / _MCP_TOOLS_CACHE_FILENAME + if not path.is_file(): + continue + with open(path, 'rb') as f: + data = f.read(_MCP_TOOLS_CACHE_MAX_BYTES + 1) + if len(data) > _MCP_TOOLS_CACHE_MAX_BYTES: + return {} + parsed = json.loads(data.decode('utf-8')) + return parsed if isinstance(parsed, dict) else {} + except Exception: + pass + return {} + + +def _mcp_cache_entries_for_user(tools): + username = Path.home().name + entries = [] + for key, by_user in tools.items(): + if not isinstance(key, str) or not isinstance(by_user, dict): + continue + k = key.strip().lower() + if k in _MCP_CACHE_CODING_TOOL_NAMES or k.startswith(_MCP_CACHE_CODING_TOOL_PREFIXES): + entry = by_user.get(username) + if isinstance(entry, dict): + entries.append(entry) + return entries + + +_CONTENT_HASH_RE = re.compile(r'^[a-f0-9]{64}$', re.IGNORECASE) + + +def _lookup_tool_content_hash(server_name, mcp_tool, server_cfg): + try: + if not server_name or not mcp_tool or not isinstance(server_cfg, dict): + return None + cache_key = compute_mcp_cache_key( + name=server_name, + command=server_cfg.get('command'), + url=server_cfg.get('url'), + args=server_cfg.get('args'), + ) + if not cache_key: + return None + tools = _read_mcp_tools_cache().get('tools') + if not isinstance(tools, dict): + return None + for entry in _mcp_cache_entries_for_user(tools): + by_tool = entry.get(cache_key) + if not isinstance(by_tool, dict): + continue + content_hash = by_tool.get(mcp_tool) + if isinstance(content_hash, str) and _CONTENT_HASH_RE.match(content_hash): + return content_hash + return None + except Exception: + return None + + +def _attach_tool_content_hash(metadata): + try: + server_cfg = metadata.get('mcp_server_config') + if not isinstance(server_cfg, dict): + return + content_hash = _lookup_tool_content_hash( + metadata.get('mcp_server'), metadata.get('mcp_tool'), server_cfg + ) + if content_hash: + server_cfg['tool_content_hash'] = content_hash + except Exception: + pass + + +# ───────────────────────── end MCP tool risk-scoring section ───────────────── + + def _email_domain(email: Optional[str]) -> Optional[str]: try: if email and '@' in email: @@ -1379,6 +1494,8 @@ def process_pre_tool_use(event: Dict, api_key: str) -> Dict: if _is_uuid(mcp_server_name): metadata['mcp_server_uuid'] = mcp_server_name + _attach_tool_content_hash(metadata) + approval_key = f"{tool_name}:{command}" is_retry = _is_approval_retry(approval_key) @@ -2000,6 +2117,7 @@ def _dispatch_mcp_server_scan(server_name: str, server_config: Dict, cwd: Option "UNBOUND_API_KEY": api_key, "UNBOUND_MCP_SERVER_JSON": json.dumps(server_config), "UNBOUND_MCP_SERVER_NAME": server_name, + "UNBOUND_CODING_TOOL": _UNBOUND_CODING_TOOL, "UNBOUND_MCP_DOMAIN": backend_url}, } if os.name == "nt": diff --git a/codex/hooks/unbound.py b/codex/hooks/unbound.py index e618b9c..5fe815f 100644 --- a/codex/hooks/unbound.py +++ b/codex/hooks/unbound.py @@ -684,6 +684,121 @@ def _augment_script_hash(result, cwd): return result +# KEEP IN SYNC: coding-discovery-tool mcp_tools_cache.py + all 5 hook copies — byte-identical, do not diverge. + +_MCP_TOOLS_CACHE_FILENAME = 'mcp-tools-cache.json' +_MCP_TOOLS_CACHE_MAX_BYTES = 2 * 1024 * 1024 +_MCP_CACHE_CODING_TOOL_NAMES = frozenset({'codex'}) +_MCP_CACHE_CODING_TOOL_PREFIXES = () +_UNBOUND_CODING_TOOL = 'Codex' + + +def compute_mcp_cache_key(name, command, url, args): + subset = {} + clean_name = name.strip() if isinstance(name, str) else '' + if clean_name: + subset['name'] = clean_name + clean_url = url.strip() if isinstance(url, str) else '' + if clean_url: + subset['url'] = clean_url + clean_command = command.strip() if isinstance(command, str) else '' + if clean_command: + subset['command'] = clean_command + if args: + subset['args'] = args + if not subset: + return None + encoded = json.dumps(subset, sort_keys=True, separators=(',', ':')) + return hashlib.sha256(encoded.encode('utf-8')).hexdigest() + + +def _unbound_state_dir_candidates(): + candidates = [Path.home() / '.unbound'] + if hasattr(os, 'getuid'): + candidates.append(Path(f'/var/tmp/unbound-{os.getuid()}')) + else: + candidates.append(Path(tempfile.gettempdir()) / 'unbound') + return candidates + + +def _read_mcp_tools_cache(): + try: + for state_dir in _unbound_state_dir_candidates(): + path = state_dir / _MCP_TOOLS_CACHE_FILENAME + if not path.is_file(): + continue + with open(path, 'rb') as f: + data = f.read(_MCP_TOOLS_CACHE_MAX_BYTES + 1) + if len(data) > _MCP_TOOLS_CACHE_MAX_BYTES: + return {} + parsed = json.loads(data.decode('utf-8')) + return parsed if isinstance(parsed, dict) else {} + except Exception: + pass + return {} + + +def _mcp_cache_entries_for_user(tools): + username = Path.home().name + entries = [] + for key, by_user in tools.items(): + if not isinstance(key, str) or not isinstance(by_user, dict): + continue + k = key.strip().lower() + if k in _MCP_CACHE_CODING_TOOL_NAMES or k.startswith(_MCP_CACHE_CODING_TOOL_PREFIXES): + entry = by_user.get(username) + if isinstance(entry, dict): + entries.append(entry) + return entries + + +_CONTENT_HASH_RE = re.compile(r'^[a-f0-9]{64}$', re.IGNORECASE) + + +def _lookup_tool_content_hash(server_name, mcp_tool, server_cfg): + try: + if not server_name or not mcp_tool or not isinstance(server_cfg, dict): + return None + cache_key = compute_mcp_cache_key( + name=server_name, + command=server_cfg.get('command'), + url=server_cfg.get('url'), + args=server_cfg.get('args'), + ) + if not cache_key: + return None + tools = _read_mcp_tools_cache().get('tools') + if not isinstance(tools, dict): + return None + for entry in _mcp_cache_entries_for_user(tools): + by_tool = entry.get(cache_key) + if not isinstance(by_tool, dict): + continue + content_hash = by_tool.get(mcp_tool) + if isinstance(content_hash, str) and _CONTENT_HASH_RE.match(content_hash): + return content_hash + return None + except Exception: + return None + + +def _attach_tool_content_hash(metadata): + try: + server_cfg = metadata.get('mcp_server_config') + if not isinstance(server_cfg, dict): + return + content_hash = _lookup_tool_content_hash( + metadata.get('mcp_server'), metadata.get('mcp_tool'), server_cfg + ) + if content_hash: + server_cfg['tool_content_hash'] = content_hash + except Exception: + pass + + +# ───────────────────────── end MCP tool risk-scoring section ───────────────── + + def _read_mcp_server_config(server_name, config_path): """ Read an MCP server's config (url, command, args, type) from the codex @@ -866,6 +981,7 @@ def process_pre_tool_use(event: Dict, api_key: str) -> Dict: server_cfg = _read_mcp_server_config(mcp_server, CODEX_CONFIG_PATH) if server_cfg: metadata['mcp_server_config'] = _augment_script_hash(server_cfg, metadata.get('cwd')) + _attach_tool_content_hash(metadata) approval_key = f"{tool_name}:{command}" is_retry = _is_approval_retry(approval_key) @@ -1517,6 +1633,7 @@ def _dispatch_mcp_server_scan(server_name: str, server_config: Dict) -> None: "UNBOUND_API_KEY": api_key, "UNBOUND_MCP_SERVER_JSON": json.dumps(server_config), "UNBOUND_MCP_SERVER_NAME": server_name, + "UNBOUND_CODING_TOOL": _UNBOUND_CODING_TOOL, "UNBOUND_MCP_DOMAIN": backend_url}, } if os.name == "nt": diff --git a/copilot/hooks/unbound.py b/copilot/hooks/unbound.py index 781f4b1..1a5b8ce 100644 --- a/copilot/hooks/unbound.py +++ b/copilot/hooks/unbound.py @@ -662,6 +662,121 @@ def _augment_script_hash(result, cwd): return result +# KEEP IN SYNC: coding-discovery-tool mcp_tools_cache.py + all 5 hook copies — byte-identical, do not diverge. + +_MCP_TOOLS_CACHE_FILENAME = 'mcp-tools-cache.json' +_MCP_TOOLS_CACHE_MAX_BYTES = 2 * 1024 * 1024 +_MCP_CACHE_CODING_TOOL_NAMES = frozenset({'github copilot cli'}) +_MCP_CACHE_CODING_TOOL_PREFIXES = ('github copilot',) +_UNBOUND_CODING_TOOL = 'GitHub Copilot CLI' + + +def compute_mcp_cache_key(name, command, url, args): + subset = {} + clean_name = name.strip() if isinstance(name, str) else '' + if clean_name: + subset['name'] = clean_name + clean_url = url.strip() if isinstance(url, str) else '' + if clean_url: + subset['url'] = clean_url + clean_command = command.strip() if isinstance(command, str) else '' + if clean_command: + subset['command'] = clean_command + if args: + subset['args'] = args + if not subset: + return None + encoded = json.dumps(subset, sort_keys=True, separators=(',', ':')) + return hashlib.sha256(encoded.encode('utf-8')).hexdigest() + + +def _unbound_state_dir_candidates(): + candidates = [Path.home() / '.unbound'] + if hasattr(os, 'getuid'): + candidates.append(Path(f'/var/tmp/unbound-{os.getuid()}')) + else: + candidates.append(Path(tempfile.gettempdir()) / 'unbound') + return candidates + + +def _read_mcp_tools_cache(): + try: + for state_dir in _unbound_state_dir_candidates(): + path = state_dir / _MCP_TOOLS_CACHE_FILENAME + if not path.is_file(): + continue + with open(path, 'rb') as f: + data = f.read(_MCP_TOOLS_CACHE_MAX_BYTES + 1) + if len(data) > _MCP_TOOLS_CACHE_MAX_BYTES: + return {} + parsed = json.loads(data.decode('utf-8')) + return parsed if isinstance(parsed, dict) else {} + except Exception: + pass + return {} + + +def _mcp_cache_entries_for_user(tools): + username = Path.home().name + entries = [] + for key, by_user in tools.items(): + if not isinstance(key, str) or not isinstance(by_user, dict): + continue + k = key.strip().lower() + if k in _MCP_CACHE_CODING_TOOL_NAMES or k.startswith(_MCP_CACHE_CODING_TOOL_PREFIXES): + entry = by_user.get(username) + if isinstance(entry, dict): + entries.append(entry) + return entries + + +_CONTENT_HASH_RE = re.compile(r'^[a-f0-9]{64}$', re.IGNORECASE) + + +def _lookup_tool_content_hash(server_name, mcp_tool, server_cfg): + try: + if not server_name or not mcp_tool or not isinstance(server_cfg, dict): + return None + cache_key = compute_mcp_cache_key( + name=server_name, + command=server_cfg.get('command'), + url=server_cfg.get('url'), + args=server_cfg.get('args'), + ) + if not cache_key: + return None + tools = _read_mcp_tools_cache().get('tools') + if not isinstance(tools, dict): + return None + for entry in _mcp_cache_entries_for_user(tools): + by_tool = entry.get(cache_key) + if not isinstance(by_tool, dict): + continue + content_hash = by_tool.get(mcp_tool) + if isinstance(content_hash, str) and _CONTENT_HASH_RE.match(content_hash): + return content_hash + return None + except Exception: + return None + + +def _attach_tool_content_hash(metadata): + try: + server_cfg = metadata.get('mcp_server_config') + if not isinstance(server_cfg, dict): + return + content_hash = _lookup_tool_content_hash( + metadata.get('mcp_server'), metadata.get('mcp_tool'), server_cfg + ) + if content_hash: + server_cfg['tool_content_hash'] = content_hash + except Exception: + pass + + +# ───────────────────────── end MCP tool risk-scoring section ───────────────── + + def read_copilot_mcp_servers(cwd=None): servers = {} for config_path in _copilot_mcp_config_paths(cwd): @@ -1099,6 +1214,7 @@ def process_pre_tool_use(event, api_key): metadata['mcp_tool'] = mcp_tool if mcp_server_config: metadata['mcp_server_config'] = mcp_server_config + _attach_tool_content_hash(metadata) approval_key = f"{canonical}:{command}" is_retry = _is_approval_retry(approval_key) diff --git a/cursor/unbound.py b/cursor/unbound.py index a266fd9..8f568da 100644 --- a/cursor/unbound.py +++ b/cursor/unbound.py @@ -879,6 +879,121 @@ def _augment_script_hash(result, cwd): return result +# KEEP IN SYNC: coding-discovery-tool mcp_tools_cache.py + all 5 hook copies — byte-identical, do not diverge. + +_MCP_TOOLS_CACHE_FILENAME = 'mcp-tools-cache.json' +_MCP_TOOLS_CACHE_MAX_BYTES = 2 * 1024 * 1024 +_MCP_CACHE_CODING_TOOL_NAMES = frozenset({'cursor', 'cursor cli'}) +_MCP_CACHE_CODING_TOOL_PREFIXES = () +_UNBOUND_CODING_TOOL = 'Cursor' + + +def compute_mcp_cache_key(name, command, url, args): + subset = {} + clean_name = name.strip() if isinstance(name, str) else '' + if clean_name: + subset['name'] = clean_name + clean_url = url.strip() if isinstance(url, str) else '' + if clean_url: + subset['url'] = clean_url + clean_command = command.strip() if isinstance(command, str) else '' + if clean_command: + subset['command'] = clean_command + if args: + subset['args'] = args + if not subset: + return None + encoded = json.dumps(subset, sort_keys=True, separators=(',', ':')) + return hashlib.sha256(encoded.encode('utf-8')).hexdigest() + + +def _unbound_state_dir_candidates(): + candidates = [Path.home() / '.unbound'] + if hasattr(os, 'getuid'): + candidates.append(Path(f'/var/tmp/unbound-{os.getuid()}')) + else: + candidates.append(Path(tempfile.gettempdir()) / 'unbound') + return candidates + + +def _read_mcp_tools_cache(): + try: + for state_dir in _unbound_state_dir_candidates(): + path = state_dir / _MCP_TOOLS_CACHE_FILENAME + if not path.is_file(): + continue + with open(path, 'rb') as f: + data = f.read(_MCP_TOOLS_CACHE_MAX_BYTES + 1) + if len(data) > _MCP_TOOLS_CACHE_MAX_BYTES: + return {} + parsed = json.loads(data.decode('utf-8')) + return parsed if isinstance(parsed, dict) else {} + except Exception: + pass + return {} + + +def _mcp_cache_entries_for_user(tools): + username = Path.home().name + entries = [] + for key, by_user in tools.items(): + if not isinstance(key, str) or not isinstance(by_user, dict): + continue + k = key.strip().lower() + if k in _MCP_CACHE_CODING_TOOL_NAMES or k.startswith(_MCP_CACHE_CODING_TOOL_PREFIXES): + entry = by_user.get(username) + if isinstance(entry, dict): + entries.append(entry) + return entries + + +_CONTENT_HASH_RE = re.compile(r'^[a-f0-9]{64}$', re.IGNORECASE) + + +def _lookup_tool_content_hash(server_name, mcp_tool, server_cfg): + try: + if not server_name or not mcp_tool or not isinstance(server_cfg, dict): + return None + cache_key = compute_mcp_cache_key( + name=server_name, + command=server_cfg.get('command'), + url=server_cfg.get('url'), + args=server_cfg.get('args'), + ) + if not cache_key: + return None + tools = _read_mcp_tools_cache().get('tools') + if not isinstance(tools, dict): + return None + for entry in _mcp_cache_entries_for_user(tools): + by_tool = entry.get(cache_key) + if not isinstance(by_tool, dict): + continue + content_hash = by_tool.get(mcp_tool) + if isinstance(content_hash, str) and _CONTENT_HASH_RE.match(content_hash): + return content_hash + return None + except Exception: + return None + + +def _attach_tool_content_hash(metadata): + try: + server_cfg = metadata.get('mcp_server_config') + if not isinstance(server_cfg, dict): + return + content_hash = _lookup_tool_content_hash( + metadata.get('mcp_server'), metadata.get('mcp_tool'), server_cfg + ) + if content_hash: + server_cfg['tool_content_hash'] = content_hash + except Exception: + pass + + +# ───────────────────────── end MCP tool risk-scoring section ───────────────── + + def _read_mcp_server_config(server_name, config_path): """ Read an MCP server's config (url, command, args) from a config file. @@ -935,6 +1050,8 @@ def process_pre_tool_use_execution(event, api_key, tool_name, command, mcp_serve if mcp_tool is not None: metadata['mcp_tool'] = mcp_tool + _attach_tool_content_hash(metadata) + approval_key = f"{tool_name}:{command}" is_retry = _is_approval_retry(approval_key) @@ -1634,6 +1751,7 @@ def _dispatch_mcp_server_scan(server_name, server_config): "UNBOUND_API_KEY": api_key, "UNBOUND_MCP_SERVER_JSON": json.dumps(server_config), "UNBOUND_MCP_SERVER_NAME": server_name, + "UNBOUND_CODING_TOOL": _UNBOUND_CODING_TOOL, "UNBOUND_MCP_DOMAIN": backend_url}, } if os.name == "nt":