From 2b7f6c235597d4c44527546ebba10bb87386a70b Mon Sep 17 00:00:00 2001 From: Ilan Lidovski Date: Mon, 27 Jul 2026 15:05:45 +0300 Subject: [PATCH 1/2] CM-68943 Send ai-guardrails hook context with the scan so report-mode findings become violations The scan now carries the hook's context inside the generic scan_parameters metadata bag under an `ai_guardrails` key, which lets secret-detector attribute the finding to the developer's device and, in report mode, promote it to a real violation instead of a scan-scoped detection. - `detection_source` is sent from `SECRETS_BLOCK_REASON_BY_EVENT_TYPE`, so the violation and the ai-security-manager hook event describe the same event with the same vocabulary. The three handlers now look the reason up through that map instead of hardcoding it, and `_ArgScanFeature` no longer carries a `block_reason` alongside its event type. - `GuardrailsMode` replaces `InstallMode`: the same report/block pair describes both the installed mode and a single event's effective mode, which `get_effective_mode` now derives from the policy in one call. - A generation id is minted per hook event when the IDE dialect does not supply one (Copilot), since it scopes the violation's identity. - The device serial is cached in a temp file, keyed by username, so repeated hook invocations don't re-run the platform lookup. - Copilot's sessionStart hook drops its timeoutSec. Co-Authored-By: Claude Opus 5 (1M context) --- cycode/cli/apps/ai_guardrails/consts.py | 9 +- cycode/cli/apps/ai_guardrails/ides/copilot.py | 4 +- .../cli/apps/ai_guardrails/install_command.py | 14 +-- .../cli/apps/ai_guardrails/scan/handlers.py | 113 ++++++++++++++---- .../apps/ai_guardrails/scan/scan_command.py | 4 + cycode/cli/apps/ai_guardrails/scan/types.py | 9 ++ cycode/cli/utils/host_info.py | 36 ++++++ .../ai_guardrails/scan/test_handlers.py | 81 ++++++++++++- 8 files changed, 229 insertions(+), 41 deletions(-) diff --git a/cycode/cli/apps/ai_guardrails/consts.py b/cycode/cli/apps/ai_guardrails/consts.py index 8018fa73..4c962767 100644 --- a/cycode/cli/apps/ai_guardrails/consts.py +++ b/cycode/cli/apps/ai_guardrails/consts.py @@ -10,8 +10,13 @@ class PolicyMode(str, Enum): WARN = 'warn' -class InstallMode(str, Enum): - """Installation mode for ai-guardrails install command.""" +class GuardrailsMode(str, Enum): + """Guardrails enforcement mode. + + Used both as the ai-guardrails install-command mode and as the per-event + effective mode reported to the server (the ai_guardrails scan parameter's + `mode` field) + """ REPORT = 'report' BLOCK = 'block' diff --git a/cycode/cli/apps/ai_guardrails/ides/copilot.py b/cycode/cli/apps/ai_guardrails/ides/copilot.py index 12ef8f89..2cd6a427 100644 --- a/cycode/cli/apps/ai_guardrails/ides/copilot.py +++ b/cycode/cli/apps/ai_guardrails/ides/copilot.py @@ -333,9 +333,7 @@ def entry(command: str) -> dict: return { 'version': 1, 'hooks': { - 'sessionStart': [ - {'type': 'command', 'command': _SESSION_START_COMMAND, 'timeoutSec': _HOOK_TIMEOUT_SEC} - ], + 'sessionStart': [{'type': 'command', 'command': _SESSION_START_COMMAND}], 'userPromptSubmitted': [entry(_SCAN_PROMPT_COMMAND)], 'preToolUse': [entry(_SCAN_TOOL_COMMAND)], }, diff --git a/cycode/cli/apps/ai_guardrails/install_command.py b/cycode/cli/apps/ai_guardrails/install_command.py index 0ee5aacb..155cf83a 100644 --- a/cycode/cli/apps/ai_guardrails/install_command.py +++ b/cycode/cli/apps/ai_guardrails/install_command.py @@ -6,7 +6,7 @@ import typer from cycode.cli.apps.ai_guardrails.command_utils import console, resolve_repo_path, validate_scope -from cycode.cli.apps.ai_guardrails.consts import InstallMode, PolicyMode +from cycode.cli.apps.ai_guardrails.consts import GuardrailsMode, PolicyMode from cycode.cli.apps.ai_guardrails.hooks_manager import create_policy_file, install_hooks from cycode.cli.apps.ai_guardrails.ides import DEFAULT_IDE_NAME, IDES, resolve_ides @@ -40,14 +40,14 @@ def install_command( ), ] = None, mode: Annotated[ - InstallMode, + GuardrailsMode, typer.Option( '--mode', '-m', help='Installation mode: "report" for async non-blocking hooks with warn policy, ' '"block" for sync blocking hooks.', ), - ] = InstallMode.REPORT, + ] = GuardrailsMode.REPORT, ) -> None: """Install AI guardrails hooks for supported IDEs. @@ -65,7 +65,7 @@ def install_command( repo_path = resolve_repo_path(scope, repo_path) ides_to_install = resolve_ides(ide) - report_mode = mode == InstallMode.REPORT + report_mode = mode == GuardrailsMode.REPORT results: list[tuple[str, bool, str]] = [] for current_ide in ides_to_install: @@ -83,7 +83,7 @@ def install_command( all_success = False if any_success: - policy_mode = PolicyMode.WARN if mode == InstallMode.REPORT else PolicyMode.BLOCK + policy_mode = PolicyMode.WARN if mode == GuardrailsMode.REPORT else PolicyMode.BLOCK _install_policy(scope, repo_path, policy_mode) _print_next_steps(results, mode) @@ -99,7 +99,7 @@ def _install_policy(scope: str, repo_path: Optional[Path], policy_mode: PolicyMo console.print(f'[red]✗[/] {policy_message}', style='bold red') -def _print_next_steps(results: list[tuple[str, bool, str]], mode: InstallMode) -> None: +def _print_next_steps(results: list[tuple[str, bool, str]], mode: GuardrailsMode) -> None: console.print() console.print('[bold]Next steps:[/]') successful_ides = [name for name, success, _ in results if success] @@ -107,7 +107,7 @@ def _print_next_steps(results: list[tuple[str, bool, str]], mode: InstallMode) - console.print(f'1. Restart {ide_list} to activate the hooks') console.print('2. (Optional) Customize policy in ~/.cycode/ai-guardrails.yaml') console.print() - if mode == InstallMode.REPORT: + if mode == GuardrailsMode.REPORT: console.print('[dim]Report mode: hooks run async (non-blocking) and policy is set to warn.[/]') else: console.print('[dim]The hooks will scan prompts, file reads, and MCP tool calls for secrets.[/]') diff --git a/cycode/cli/apps/ai_guardrails/scan/handlers.py b/cycode/cli/apps/ai_guardrails/scan/handlers.py index ab05482a..e82c61d7 100644 --- a/cycode/cli/apps/ai_guardrails/scan/handlers.py +++ b/cycode/cli/apps/ai_guardrails/scan/handlers.py @@ -17,24 +17,29 @@ import typer -from cycode.cli.apps.ai_guardrails.consts import PolicyMode +from cycode.cli.apps.ai_guardrails.consts import GuardrailsMode, PolicyMode from cycode.cli.apps.ai_guardrails.ides.base import HookDecision from cycode.cli.apps.ai_guardrails.scan.payload import AIHookPayload from cycode.cli.apps.ai_guardrails.scan.policy import get_policy_value -from cycode.cli.apps.ai_guardrails.scan.types import AiHookEventType, AIHookOutcome, BlockReason +from cycode.cli.apps.ai_guardrails.scan.types import ( + SECRETS_BLOCK_REASON_BY_EVENT_TYPE, + AiHookEventType, + AIHookOutcome, + BlockReason, +) from cycode.cli.apps.ai_guardrails.scan.utils import is_denied_path, truncate_utf8 from cycode.cli.apps.scan.code_scanner import _get_scan_documents_thread_func from cycode.cli.apps.scan.scan_parameters import get_scan_parameters from cycode.cli.cli_types import ScanTypeOption, SeverityOption from cycode.cli.files_collector.file_excluder import is_path_configured_in_exclusions from cycode.cli.models import Document +from cycode.cli.utils.host_info import get_hostname, get_serial_number from cycode.cli.utils.progress_bar import DummyProgressBar, ScanProgressBarSection from cycode.cli.utils.scan_utils import build_violation_summary from cycode.logger import get_logger logger = get_logger('AI Guardrails') - HandlerFn = Callable[[typer.Context, AIHookPayload, dict], HookDecision] @@ -47,7 +52,7 @@ def handle_before_submit_prompt(ctx: typer.Context, payload: AIHookPayload, poli ai_client.create_event(payload, AiHookEventType.PROMPT, AIHookOutcome.ALLOWED) return HookDecision.allow(AiHookEventType.PROMPT) - mode = get_policy_value(policy, 'mode', default=PolicyMode.BLOCK) + effective_mode = get_effective_mode(policy, prompt_config) prompt = payload.prompt or '' max_bytes = get_policy_value(policy, 'secrets', 'max_bytes', default=200000) timeout_ms = get_policy_value(policy, 'secrets', 'timeout_ms', default=30000) @@ -59,12 +64,18 @@ def handle_before_submit_prompt(ctx: typer.Context, payload: AIHookPayload, poli error_message = None try: - violation_summary, scan_id = _scan_text_for_secrets(ctx, clipped, timeout_ms) + violation_summary, scan_id = _scan_text_for_secrets( + ctx, + clipped, + timeout_ms, + payload=payload, + event_type=AiHookEventType.PROMPT, + effective_mode=effective_mode, + ) if violation_summary: - block_reason = BlockReason.SECRETS_IN_PROMPT - action = get_policy_value(prompt_config, 'action', default=PolicyMode.BLOCK) - if action == PolicyMode.BLOCK and mode == PolicyMode.BLOCK: + block_reason = SECRETS_BLOCK_REASON_BY_EVENT_TYPE[AiHookEventType.PROMPT] + if effective_mode == GuardrailsMode.BLOCK: outcome = AIHookOutcome.BLOCKED user_message = f'{violation_summary}. Remove secrets before sending.' return HookDecision.deny(AiHookEventType.PROMPT, user_message) @@ -97,9 +108,8 @@ def handle_before_read_file(ctx: typer.Context, payload: AIHookPayload, policy: ai_client.create_event(payload, AiHookEventType.FILE_READ, AIHookOutcome.ALLOWED) return HookDecision.allow(AiHookEventType.FILE_READ) - mode = get_policy_value(policy, 'mode', default=PolicyMode.BLOCK) file_path = payload.file_path or '' - action = get_policy_value(file_read_config, 'action', default=PolicyMode.BLOCK) + effective_mode = get_effective_mode(policy, file_read_config) scan_id = None block_reason = None @@ -110,7 +120,7 @@ def handle_before_read_file(ctx: typer.Context, payload: AIHookPayload, policy: is_sensitive_path = is_denied_path(file_path, policy) if is_sensitive_path: block_reason = BlockReason.SENSITIVE_PATH - if mode == PolicyMode.BLOCK and action == PolicyMode.BLOCK: + if effective_mode == GuardrailsMode.BLOCK: outcome = AIHookOutcome.BLOCKED user_message = f'Cycode blocked sending {file_path} to the AI (sensitive path policy).' return HookDecision.deny( @@ -133,10 +143,12 @@ def handle_before_read_file(ctx: typer.Context, payload: AIHookPayload, policy: outcome = AIHookOutcome.ALLOWED if get_policy_value(file_read_config, 'scan_content', default=True): - violation_summary, scan_id = _scan_path_for_secrets(ctx, file_path, policy) + violation_summary, scan_id = _scan_path_for_secrets( + ctx, file_path, policy, payload=payload, effective_mode=effective_mode + ) if violation_summary: - block_reason = BlockReason.SECRETS_IN_FILE - if mode == PolicyMode.BLOCK and action == PolicyMode.BLOCK: + block_reason = SECRETS_BLOCK_REASON_BY_EVENT_TYPE[AiHookEventType.FILE_READ] + if effective_mode == GuardrailsMode.BLOCK: outcome = AIHookOutcome.BLOCKED user_message = f'Cycode blocked reading {file_path}. {violation_summary}' return HookDecision.deny( @@ -191,7 +203,6 @@ class _ArgScanFeature: policy_key: str # 'mcp' or 'command_exec' scan_key: str # 'scan_arguments' or 'scan_command' event_type: AiHookEventType - block_reason: BlockReason deny_message: Callable[[str], str] deny_agent_message: str ask_message: Callable[[str], str] @@ -213,11 +224,10 @@ def _handle_arg_scan( ai_client.create_event(payload, feature.event_type, AIHookOutcome.ALLOWED) return HookDecision.allow(feature.event_type) - mode = get_policy_value(policy, 'mode', default=PolicyMode.BLOCK) max_bytes = get_policy_value(policy, 'secrets', 'max_bytes', default=200000) timeout_ms = get_policy_value(policy, 'secrets', 'timeout_ms', default=30000) clipped = truncate_utf8(scan_text, max_bytes) - action = get_policy_value(feature_config, 'action', default=PolicyMode.BLOCK) + effective_mode = get_effective_mode(policy, feature_config) scan_id = None block_reason = None @@ -226,10 +236,17 @@ def _handle_arg_scan( try: if get_policy_value(feature_config, feature.scan_key, default=True): - violation_summary, scan_id = _scan_text_for_secrets(ctx, clipped, timeout_ms) + violation_summary, scan_id = _scan_text_for_secrets( + ctx, + clipped, + timeout_ms, + payload=payload, + event_type=feature.event_type, + effective_mode=effective_mode, + ) if violation_summary: - block_reason = feature.block_reason - if mode == PolicyMode.BLOCK and action == PolicyMode.BLOCK: + block_reason = SECRETS_BLOCK_REASON_BY_EVENT_TYPE[feature.event_type] + if effective_mode == GuardrailsMode.BLOCK: outcome = AIHookOutcome.BLOCKED return HookDecision.deny( feature.event_type, @@ -275,7 +292,6 @@ def handle_before_mcp_execution(ctx: typer.Context, payload: AIHookPayload, poli policy_key='mcp', scan_key='scan_arguments', event_type=AiHookEventType.MCP_EXECUTION, - block_reason=BlockReason.SECRETS_IN_MCP_ARGS, deny_message=lambda v: f'Cycode blocked MCP tool call "{tool}". {v}', deny_agent_message='Do not pass secrets to tools. Use secret references (name/id) instead.', ask_message=lambda v: f'{v} in MCP tool call "{tool}". Allow execution?', @@ -295,6 +311,36 @@ def get_handler_for_event(event_type: str) -> Optional[HandlerFn]: return handlers.get(event_type) +def get_effective_mode(policy: dict, feature_config: dict) -> GuardrailsMode: + """The event only blocks when both the global mode and the per-guardrail action are block.""" + mode = get_policy_value(policy, 'mode', default=PolicyMode.BLOCK) + action = get_policy_value(feature_config, 'action', default=PolicyMode.BLOCK) + return GuardrailsMode.BLOCK if (mode == PolicyMode.BLOCK and action == PolicyMode.BLOCK) else GuardrailsMode.REPORT + + +def build_ai_guardrails_scan_parameters( + ctx: typer.Context, + paths: Optional[tuple[str, ...]], + payload: AIHookPayload, + event_type: AiHookEventType, + effective_mode: GuardrailsMode, +) -> dict: + scan_parameters = get_scan_parameters(ctx, paths) + scan_parameters.setdefault('metadata', {})['ai_guardrails'] = { + 'mode': effective_mode.value, + 'ide_provider': payload.ide_provider, + 'detection_source': SECRETS_BLOCK_REASON_BY_EVENT_TYPE[event_type].value, + 'device_id': get_serial_number(), + 'device_hostname': get_hostname(), + 'conversation_id': payload.conversation_id, + 'generation_id': payload.generation_id, + 'ide_user_email': payload.ide_user_email, + 'mcp_server_name': payload.mcp_server_name, + 'mcp_tool_name': payload.mcp_tool_name, + } + return scan_parameters + + def _setup_scan_context(ctx: typer.Context) -> typer.Context: """Set up minimal context for scan_documents without progress bars or printing.""" ctx.obj['progress_bar'] = DummyProgressBar([ScanProgressBarSection]) @@ -345,7 +391,14 @@ def _perform_scan( return None, scan_id -def _scan_text_for_secrets(ctx: typer.Context, text: str, timeout_ms: int) -> tuple[Optional[str], Optional[str]]: +def _scan_text_for_secrets( + ctx: typer.Context, + text: str, + timeout_ms: int, + payload: AIHookPayload, + event_type: AiHookEventType, + effective_mode: GuardrailsMode, +) -> tuple[Optional[str], Optional[str]]: """Scan text content for secrets using Cycode CLI.""" if not text: return None, None @@ -353,10 +406,17 @@ def _scan_text_for_secrets(ctx: typer.Context, text: str, timeout_ms: int) -> tu document = Document(path='prompt-content.txt', content=text, is_git_diff_format=False) scan_ctx = _setup_scan_context(ctx) timeout_seconds = timeout_ms / 1000.0 - return _perform_scan(scan_ctx, [document], get_scan_parameters(scan_ctx, None), timeout_seconds) + scan_parameters = build_ai_guardrails_scan_parameters(scan_ctx, None, payload, event_type, effective_mode) + return _perform_scan(scan_ctx, [document], scan_parameters, timeout_seconds) -def _scan_path_for_secrets(ctx: typer.Context, file_path: str, policy: dict) -> tuple[Optional[str], Optional[str]]: +def _scan_path_for_secrets( + ctx: typer.Context, + file_path: str, + policy: dict, + payload: AIHookPayload, + effective_mode: GuardrailsMode, +) -> tuple[Optional[str], Optional[str]]: """Scan a file path for secrets.""" if not file_path or not os.path.isfile(file_path): return None, None @@ -375,4 +435,7 @@ def _scan_path_for_secrets(ctx: typer.Context, file_path: str, policy: dict) -> document = Document(path=os.path.basename(file_path), content=content, is_git_diff_format=False) scan_ctx = _setup_scan_context(ctx) - return _perform_scan(scan_ctx, [document], get_scan_parameters(scan_ctx, (file_path,)), timeout_seconds) + scan_parameters = build_ai_guardrails_scan_parameters( + scan_ctx, (file_path,), payload, AiHookEventType.FILE_READ, effective_mode + ) + return _perform_scan(scan_ctx, [document], scan_parameters, timeout_seconds) diff --git a/cycode/cli/apps/ai_guardrails/scan/scan_command.py b/cycode/cli/apps/ai_guardrails/scan/scan_command.py index 1a389d5e..1c0c42b8 100644 --- a/cycode/cli/apps/ai_guardrails/scan/scan_command.py +++ b/cycode/cli/apps/ai_guardrails/scan/scan_command.py @@ -8,6 +8,7 @@ """ from typing import Annotated, Optional, Union +from uuid import uuid4 import click import typer @@ -125,6 +126,9 @@ def scan_command( return unified_payload = ide_integration.parse_hook_payload(payload) + if not unified_payload.generation_id: + # Not every IDE dialect provides a generation id (e.g. Copilot) + unified_payload.generation_id = str(uuid4()) event_name = unified_payload.event_name logger.debug( 'Processing AI guardrails hook', diff --git a/cycode/cli/apps/ai_guardrails/scan/types.py b/cycode/cli/apps/ai_guardrails/scan/types.py index da42ed23..5d18e07d 100644 --- a/cycode/cli/apps/ai_guardrails/scan/types.py +++ b/cycode/cli/apps/ai_guardrails/scan/types.py @@ -41,3 +41,12 @@ class BlockReason(StrEnum): SECRETS_IN_MCP_ARGS = 'secrets_in_mcp_args' SENSITIVE_PATH = 'sensitive_path' SCAN_FAILURE = 'scan_failure' + + +# The reason each event type yields when a secret is found in it. Also travels with the scan as +# `detection_source`, so the violation and the hook event are labelled from the same vocabulary. +SECRETS_BLOCK_REASON_BY_EVENT_TYPE: dict[AiHookEventType, BlockReason] = { + AiHookEventType.PROMPT: BlockReason.SECRETS_IN_PROMPT, + AiHookEventType.FILE_READ: BlockReason.SECRETS_IN_FILE, + AiHookEventType.MCP_EXECUTION: BlockReason.SECRETS_IN_MCP_ARGS, +} diff --git a/cycode/cli/utils/host_info.py b/cycode/cli/utils/host_info.py index 23737b7a..0f39972f 100644 --- a/cycode/cli/utils/host_info.py +++ b/cycode/cli/utils/host_info.py @@ -3,6 +3,8 @@ import re import socket import subprocess +import tempfile +from pathlib import Path from typing import Optional from cycode.logger import get_logger @@ -93,6 +95,19 @@ def get_last_login_user() -> Optional[str]: def get_serial_number() -> Optional[str]: + # The serial is immutable hardware info, but resolving it shells out (ioreg/WMI) + # and this runs in a fresh process per AI hook event - cache it on disk. + cached = _read_serial_number_cache() + if cached: + return cached + + serial = _resolve_serial_number() + if serial: + _write_serial_number_cache(serial) + return serial + + +def _resolve_serial_number() -> Optional[str]: try: system = platform.system() if system == 'Darwin': @@ -104,6 +119,27 @@ def get_serial_number() -> Optional[str]: return None +def _serial_number_cache_path() -> Path: + # The username suffix avoids collisions on OSes with a shared temp dir + return Path(tempfile.gettempdir()) / f'.cycode-device-serial-{getpass.getuser()}' + + +def _read_serial_number_cache() -> Optional[str]: + try: + return _serial_number_cache_path().read_text(encoding='utf-8').strip() or None + except Exception: + return None + + +def _write_serial_number_cache(serial: str) -> None: + try: + cache_path = _serial_number_cache_path() + cache_path.parent.mkdir(parents=True, exist_ok=True) + cache_path.write_text(serial, encoding='utf-8') + except Exception as e: + logger.debug('Failed to cache serial number', exc_info=e) + + def _get_macos_serial_number() -> Optional[str]: output = _run(['ioreg', '-c', 'IOPlatformExpertDevice', '-d', '2']) if not output: diff --git a/tests/cli/commands/ai_guardrails/scan/test_handlers.py b/tests/cli/commands/ai_guardrails/scan/test_handlers.py index 01f8790c..401482ac 100644 --- a/tests/cli/commands/ai_guardrails/scan/test_handlers.py +++ b/tests/cli/commands/ai_guardrails/scan/test_handlers.py @@ -7,10 +7,14 @@ import pytest import typer +from cycode.cli.apps.ai_guardrails.consts import GuardrailsMode from cycode.cli.apps.ai_guardrails.ides.base import DecisionAction, HookDecision from cycode.cli.apps.ai_guardrails.scan.handlers import ( _perform_scan, _scan_path_for_secrets, + _scan_text_for_secrets, + build_ai_guardrails_scan_parameters, + get_effective_mode, handle_before_mcp_execution, handle_before_read_file, handle_before_submit_prompt, @@ -359,18 +363,26 @@ def test_handle_before_read_file_sensitive_path_scan_disabled_warns( assert call_args.kwargs['block_reason'] == BlockReason.SENSITIVE_PATH -def test_scan_path_for_secrets_directory(mock_ctx: MagicMock, default_policy: dict[str, Any], fs: Any) -> None: +def test_scan_path_for_secrets_directory( + mock_ctx: MagicMock, default_policy: dict[str, Any], mock_payload: AIHookPayload, fs: Any +) -> None: """Test that _scan_path_for_secrets returns (None, None) for directories.""" fs.create_dir('/path/to/some_directory') - result = _scan_path_for_secrets(mock_ctx, '/path/to/some_directory', default_policy) + result = _scan_path_for_secrets( + mock_ctx, '/path/to/some_directory', default_policy, payload=mock_payload, effective_mode=GuardrailsMode.BLOCK + ) assert result == (None, None) @patch('cycode.cli.apps.ai_guardrails.scan.handlers._perform_scan') def test_scan_path_for_secrets_skips_path_configured_in_exclusions( - mock_perform_scan: MagicMock, mock_ctx: MagicMock, default_policy: dict[str, Any], fs: Any + mock_perform_scan: MagicMock, + mock_ctx: MagicMock, + default_policy: dict[str, Any], + mock_payload: AIHookPayload, + fs: Any, ) -> None: """Test that a path ignored via `cycode ignore --by-path` is not scanned.""" # `cycode ignore --by-path` stores absolute paths; on Windows that includes the drive prefix @@ -383,7 +395,9 @@ def test_scan_path_for_secrets_skips_path_configured_in_exclusions( 'cycode.cli.files_collector.file_excluder.configuration_manager.get_exclusions_by_scan_type', return_value={'paths': [excluded_dir]}, ): - result = _scan_path_for_secrets(mock_ctx, file_path, default_policy) + result = _scan_path_for_secrets( + mock_ctx, file_path, default_policy, payload=mock_payload, effective_mode=GuardrailsMode.BLOCK + ) assert result == (None, None) mock_perform_scan.assert_not_called() @@ -512,3 +526,62 @@ def test_handle_before_mcp_execution_scan_disabled( assert result == HookDecision.allow(AiHookEventType.MCP_EXECUTION) mock_scan.assert_not_called() + + +def test_get_effective_mode_block_only_when_both_mode_and_action_block() -> None: + """The event blocks only when both the global mode and the per-guardrail action are block.""" + assert get_effective_mode({'mode': 'block'}, {'action': 'block'}) == GuardrailsMode.BLOCK + assert get_effective_mode({'mode': 'block'}, {'action': 'warn'}) == GuardrailsMode.REPORT + assert get_effective_mode({'mode': 'warn'}, {'action': 'block'}) == GuardrailsMode.REPORT + assert get_effective_mode({'mode': 'warn'}, {'action': 'warn'}) == GuardrailsMode.REPORT + + +@patch('cycode.cli.apps.ai_guardrails.scan.handlers.get_serial_number', return_value='SER-123') +@patch('cycode.cli.apps.ai_guardrails.scan.handlers.get_hostname', return_value='test-host') +def test_build_ai_guardrails_scan_parameters( + mock_hostname: MagicMock, mock_serial: MagicMock, mock_ctx: MagicMock, mock_payload: AIHookPayload +) -> None: + """The built scan parameters embed the full hook context alongside the standard scan parameters.""" + mock_ctx.info_name = 'ai_guardrails' + + params = build_ai_guardrails_scan_parameters( + mock_ctx, None, mock_payload, AiHookEventType.PROMPT, effective_mode=GuardrailsMode.REPORT + ) + + assert params['command_type'] == 'ai_guardrails' + assert params['metadata']['ai_guardrails'] == { + 'mode': 'report', + 'ide_provider': 'cursor', + 'detection_source': 'secrets_in_prompt', + 'device_id': 'SER-123', + 'device_hostname': 'test-host', + 'conversation_id': 'test-conv-id', + 'generation_id': 'test-gen-id', + 'ide_user_email': 'test@example.com', + 'mcp_server_name': None, + 'mcp_tool_name': None, + } + + +@patch('cycode.cli.apps.ai_guardrails.scan.handlers._perform_scan') +def test_scan_text_for_secrets_injects_ai_guardrails_scan_parameter( + mock_perform_scan: MagicMock, mock_ctx: MagicMock, mock_payload: AIHookPayload +) -> None: + """The scan parameters sent to the server include the ai_guardrails context.""" + mock_ctx.obj['progress_bar'] = MagicMock() + mock_perform_scan.return_value = (None, 'scan-id-123') + + _scan_text_for_secrets( + mock_ctx, + 'some text', + 1000, + payload=mock_payload, + event_type=AiHookEventType.PROMPT, + effective_mode=GuardrailsMode.REPORT, + ) + + ai_guardrails = mock_perform_scan.call_args.args[2]['metadata']['ai_guardrails'] + assert ai_guardrails['mode'] == 'report' + assert ai_guardrails['detection_source'] == 'secrets_in_prompt' + assert ai_guardrails['conversation_id'] == 'test-conv-id' + assert ai_guardrails['generation_id'] == 'test-gen-id' From 04c89c7a6bd0e8b9760def9175d152690c248729 Mon Sep 17 00:00:00 2001 From: Ilan Lidovski Date: Thu, 30 Jul 2026 15:32:01 +0300 Subject: [PATCH 2/2] CM-68943 Harden the device serial cache and document guardrails data collection The cache is now created 0600 via mkstemp and moved into place with os.replace, so a concurrent hook can't read a half-written file and the write can't be redirected by a symlink planted in the shared temp dir. README gains an AI Guardrails section stating that scanning is server-side and listing the device, user and environment data each event reports. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 30 +++++++++++++++++++++++++++++- cycode/cli/utils/host_info.py | 20 ++++++++++++++++++-- 2 files changed, 47 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index d48c4fcc..4c7ff06b 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,9 @@ This guide walks you through both installation and usage. 1. [Discovering Commands](#discovering-commands) 2. [Examples](#platform-examples) 3. [Notes & Limitations](#platform-notes--limitations) -6. [Scan Command](#scan-command) +6. [AI Guardrails](#ai-guardrails-beta) + 1. [Data Collected by AI Guardrails](#data-collected-by-ai-guardrails) +7. [Scan Command](#scan-command) 1. [Running a Scan](#running-a-scan) 1. [Options](#options) 1. [Severity Threshold](#severity-option) @@ -702,6 +704,32 @@ cycode platform projects list --page-size 100 | jq '.items[].name' - **Override the cache TTL** with `CYCODE_SPEC_CACHE_TTL=`. +# AI Guardrails \[BETA\] + +AI Guardrails installs hooks into supported AI coding agents (Claude Code, Cursor, Copilot, Codex) so that +prompts, files the agent reads, and MCP tool arguments are scanned for secrets before they reach the model. + +## Data Collected by AI Guardrails + +Scanning happens server-side, so the scanned content leaves the machine: the prompt text, the contents of +files the agent reads, and MCP tool arguments are sent to your Cycode tenant to be checked for secrets. + +Each event is also reported with context about the developer and the machine, so a finding can be attributed +to the device and user it came from. Some of this is personal data: + +- **Device identifiers** — the machine's hostname and hardware serial number. +- **User identifiers** — the email address of the user signed in to the AI coding agent, and the local + operating-system username. +- **Environment details** — operating system and version, the AI agent, its version and the model in use, + the contents of the agent's MCP configuration files, and its enabled plugins. + +The hardware serial number is cached in a local temporary file, readable only by the user who ran the +command, so repeated hook invocations don't re-query the hardware. + +If collecting this data is not acceptable in your environment, do not install the guardrails hooks +(`cycode ai-guardrails uninstall` removes hooks that are already installed). + + # Scan Command ## Running a Scan diff --git a/cycode/cli/utils/host_info.py b/cycode/cli/utils/host_info.py index 0f39972f..60a36bed 100644 --- a/cycode/cli/utils/host_info.py +++ b/cycode/cli/utils/host_info.py @@ -1,4 +1,5 @@ import getpass +import os import platform import re import socket @@ -13,6 +14,8 @@ _SUBPROCESS_TIMEOUT_SEC = 5 +_SERIAL_NUMBER_CACHE_FILE_NAME = '.cycode-device-serial' + _PLATFORM_NAMES = {'Darwin': 'macOS', 'Windows': 'Windows', 'Linux': 'Linux'} @@ -134,8 +137,21 @@ def _read_serial_number_cache() -> Optional[str]: def _write_serial_number_cache(serial: str) -> None: try: cache_path = _serial_number_cache_path() - cache_path.parent.mkdir(parents=True, exist_ok=True) - cache_path.write_text(serial, encoding='utf-8') + + # The serial identifies the machine, and the temp dir is shared, so the cache is created + # readable by its owner alone (what mkstemp does) and moved into place atomically - a hook + # racing another one never reads a half-written cache, and the rename can't be redirected + # by a symlink planted at the destination the way an in-place write could. + file_descriptor, temp_path = tempfile.mkstemp( + dir=cache_path.parent, prefix=f'{_SERIAL_NUMBER_CACHE_FILE_NAME}.' + ) + try: + with os.fdopen(file_descriptor, 'w', encoding='utf-8') as temp_file: + temp_file.write(serial) + os.replace(temp_path, cache_path) + except Exception: + Path(temp_path).unlink(missing_ok=True) + raise except Exception as e: logger.debug('Failed to cache serial number', exc_info=e)