CM-61986-add-mcp-and-email-enrichment-from-claude-json

Author: RoniCycode

cycode/cli/apps/ai_guardrails/scan/claude_config.py

@@ -0,0 +1,86 @@
+"""Reader for ~/.claude.json configuration file.
+
+Extracts user email and MCP server configuration from the Claude Code
+global config file for use in AI guardrails scan enrichment.
+"""
+
+import json
+from pathlib import Path
+from typing import Optional
+
+from cycode.logger import get_logger
+
+logger = get_logger('AI Guardrails Claude Config')
+
+_CLAUDE_CONFIG_PATH = Path.home() / '.claude.json'
+
+_SERVER_TYPE_MAPPING = {
+    'stdio': 'Local',
+    'http': 'Remote',
+}
+
+
+def load_claude_config(config_path: Optional[Path] = None) -> Optional[dict]:
+    """Load and parse ~/.claude.json.
+
+    Args:
+        config_path: Override path for testing. Defaults to ~/.claude.json.
+
+    Returns:
+        Parsed dict or None if file is missing or invalid.
+    """
+    path = config_path or _CLAUDE_CONFIG_PATH
+    if not path.exists():
+        logger.debug('Claude config file not found', extra={'path': str(path)})
+        return None
+    try:
+        content = path.read_text(encoding='utf-8')
+        return json.loads(content)
+    except Exception as e:
+        logger.debug('Failed to load Claude config file', exc_info=e)
+        return None
+
+
+def get_user_email(config: dict) -> Optional[str]:
+    """Extract user email from Claude config.
+
+    Reads oauthAccount.emailAddress from the config dict.
+    """
+    return config.get('oauthAccount', {}).get('emailAddress')
+
+
+def get_mcp_servers(config: dict) -> list[dict]:
+    """Extract top-level MCP servers from Claude config and map to backend DTO shape.
+
+    Reads the top-level mcpServers dict and transforms each entry to match
+    the McpServerPayloadDTO expected by ai-security-manager:
+        - name: server key name
+        - server_type: "Local" (stdio) or "Remote" (http)
+        - command: executable command (stdio servers)
+        - args: command arguments (stdio servers)
+        - url: server URL (http servers)
+
+    Returns:
+        List of dicts matching McpServerPayloadDTO shape. Empty list if none found.
+    """
+    mcp_servers = config.get('mcpServers', {})
+    if not isinstance(mcp_servers, dict):
+        return []
+
+    result = []
+    for name, server_config in mcp_servers.items():
+        if not isinstance(server_config, dict):
+            continue
+
+        raw_type = server_config.get('type', '')
+        server_type = _SERVER_TYPE_MAPPING.get(raw_type, raw_type)
+
+        result.append({
+            'name': name,
+            'server_type': server_type,
+            'command': server_config.get('command'),
+            'args': server_config.get('args'),
+            'url': server_config.get('url'),
+        })
+
+    return result

cycode/cli/apps/ai_guardrails/scan/payload.py

@@ -7,6 +7,7 @@
 from typing import Optional
 
 from cycode.cli.apps.ai_guardrails.consts import AIIDEType
+from cycode.cli.apps.ai_guardrails.scan.claude_config import get_user_email, load_claude_config
 from cycode.cli.apps.ai_guardrails.scan.types import (
     CLAUDE_CODE_EVENT_MAPPING,
     CLAUDE_CODE_EVENT_NAMES,
@@ -207,11 +208,15 @@ def from_claude_code_payload(cls, payload: dict) -> 'AIHookPayload':
         # Extract IDE version, model, and generation ID from transcript file
         ide_version, model, generation_id = _extract_from_claude_transcript(payload.get('transcript_path'))
 
+        # Extract user email from ~/.claude.json
+        claude_config = load_claude_config()
+        ide_user_email = get_user_email(claude_config) if claude_config else None
+
         return cls(
             event_name=canonical_event,
             conversation_id=payload.get('session_id'),
             generation_id=generation_id,
-            ide_user_email=None,  # Claude Code doesn't provide this in hook payload
+            ide_user_email=ide_user_email,
             model=model,
             ide_provider=AIIDEType.CLAUDE_CODE.value,
             ide_version=ide_version,

cycode/cli/apps/ai_guardrails/scan/scan_command.py

@@ -17,6 +17,7 @@
 import typer
 
 from cycode.cli.apps.ai_guardrails.consts import AIIDEType
+from cycode.cli.apps.ai_guardrails.scan.claude_config import get_mcp_servers, load_claude_config
 from cycode.cli.apps.ai_guardrails.scan.handlers import get_handler_for_event
 from cycode.cli.apps.ai_guardrails.scan.payload import AIHookPayload
 from cycode.cli.apps.ai_guardrails.scan.policy import load_policy
@@ -114,6 +115,16 @@ def scan_command(
     try:
         _initialize_clients(ctx)
 
+        if tool == AIIDEType.CLAUDE_CODE:
+            try:
+                claude_config = load_claude_config()
+                if claude_config:
+                    mcp_servers = get_mcp_servers(claude_config)
+                    ai_security_client = ctx.obj['ai_security_client']
+                    ai_security_client.report_mcp_servers(mcp_servers)
+            except Exception as e:
+                logger.debug('Failed to report MCP servers from Claude config', exc_info=e)
+
         handler = get_handler_for_event(event_name)
         if handler is None:
             logger.debug('Unknown hook event, allowing by default', extra={'event_name': event_name})

cycode/cyclient/ai_security_manager_client.py

@@ -17,6 +17,7 @@ class AISecurityManagerClient:
 
     _CONVERSATIONS_PATH = 'v4/ai-security/interactions/conversations'
     _EVENTS_PATH = 'v4/ai-security/interactions/events'
+    _MCP_SERVERS_PATH = 'v4/ai-security/mcp-servers'
 
     def __init__(self, client: CycodeClientBase, service_config: 'AISecurityManagerServiceConfigBase') -> None:
         self.client = client
@@ -54,6 +55,22 @@ def create_conversation(self, payload: 'AIHookPayload') -> Optional[str]:
 
         return conversation_id
 
+    def report_mcp_servers(self, mcp_servers: list[dict]) -> None:
+        """Report MCP servers discovered from IDE config.
+
+        Posts the list of MCP server configurations to the backend.
+        Failures are logged but do not block the hook flow.
+        """
+        if not mcp_servers:
+            return
+
+        body = {'mcp_servers': mcp_servers}
+
+        try:
+            self.client.post(self._build_endpoint_path(self._MCP_SERVERS_PATH), body=body)
+        except Exception as e:
+            logger.debug('Failed to report MCP servers', exc_info=e)
+
     def create_event(
         self,
         payload: 'AIHookPayload',

tests/cli/commands/ai_guardrails/test_claude_config.py

@@ -0,0 +1,119 @@
+"""Tests for Claude Code config file reader."""
+
+import json
+from pathlib import Path
+
+from pyfakefs.fake_filesystem import FakeFilesystem
+
+from cycode.cli.apps.ai_guardrails.scan.claude_config import get_mcp_servers, get_user_email, load_claude_config
+
+
+def test_load_claude_config_valid(fs: FakeFilesystem) -> None:
+    """Test loading a valid ~/.claude.json file."""
+    config = {'oauthAccount': {'emailAddress': 'user@example.com'}}
+    config_path = Path.home() / '.claude.json'
+    fs.create_file(config_path, contents=json.dumps(config))
+
+    result = load_claude_config(config_path)
+    assert result == config
+
+
+def test_load_claude_config_missing_file(fs: FakeFilesystem) -> None:
+    """Test loading when ~/.claude.json does not exist."""
+    fs.create_dir(Path.home())
+    config_path = Path.home() / '.claude.json'
+
+    result = load_claude_config(config_path)
+    assert result is None
+
+
+def test_load_claude_config_corrupt_file(fs: FakeFilesystem) -> None:
+    """Test loading when ~/.claude.json contains invalid JSON."""
+    config_path = Path.home() / '.claude.json'
+    fs.create_file(config_path, contents='not valid json {{{')
+
+    result = load_claude_config(config_path)
+    assert result is None
+
+
+def test_get_user_email_present() -> None:
+    """Test extracting email when oauthAccount.emailAddress exists."""
+    config = {'oauthAccount': {'emailAddress': 'user@example.com'}}
+    assert get_user_email(config) == 'user@example.com'
+
+
+def test_get_user_email_missing_oauth_account() -> None:
+    """Test extracting email when oauthAccount key is missing."""
+    config = {'someOtherKey': 'value'}
+    assert get_user_email(config) is None
+
+
+def test_get_user_email_missing_email_address() -> None:
+    """Test extracting email when oauthAccount exists but emailAddress is missing."""
+    config = {'oauthAccount': {'someOtherField': 'value'}}
+    assert get_user_email(config) is None
+
+
+def test_get_mcp_servers_stdio_and_http() -> None:
+    """Test extracting MCP servers with both stdio and http types."""
+    config = {
+        'mcpServers': {
+            'gitlab': {
+                'type': 'stdio',
+                'command': '/opt/homebrew/bin/gitlab-mcp',
+                'args': ['--verbose'],
+            },
+            'atlassian': {
+                'type': 'http',
+                'url': 'https://mcp.atlassian.com/v1/mcp',
+            },
+        },
+    }
+
+    result = get_mcp_servers(config)
+    assert len(result) == 2
+
+    gitlab = next(s for s in result if s['name'] == 'gitlab')
+    assert gitlab['server_type'] == 'Local'
+    assert gitlab['command'] == '/opt/homebrew/bin/gitlab-mcp'
+    assert gitlab['args'] == ['--verbose']
+    assert gitlab['url'] is None
+
+    atlassian = next(s for s in result if s['name'] == 'atlassian')
+    assert atlassian['server_type'] == 'Remote'
+    assert atlassian['command'] is None
+    assert atlassian['url'] == 'https://mcp.atlassian.com/v1/mcp'
+
+
+def test_get_mcp_servers_empty() -> None:
+    """Test extracting MCP servers when mcpServers is empty."""
+    config = {'mcpServers': {}}
+    assert get_mcp_servers(config) == []
+
+
+def test_get_mcp_servers_missing_key() -> None:
+    """Test extracting MCP servers when mcpServers key is missing."""
+    config = {'someOtherKey': 'value'}
+    assert get_mcp_servers(config) == []
+
+
+def test_get_mcp_servers_invalid_type() -> None:
+    """Test extracting MCP servers when mcpServers is not a dict."""
+    config = {'mcpServers': 'not a dict'}
+    assert get_mcp_servers(config) == []
+
+
+def test_get_mcp_servers_unknown_server_type() -> None:
+    """Test that unknown server types are passed through as-is."""
+    config = {
+        'mcpServers': {
+            'custom': {
+                'type': 'sse',
+                'url': 'https://example.com/sse',
+            },
+        },
+    }
+
+    result = get_mcp_servers(config)
+    assert len(result) == 1
+    assert result[0]['server_type'] == 'sse'