diff --git a/cycode/cli/apps/ai_guardrails/scan/guardrail_config.py b/cycode/cli/apps/ai_guardrails/scan/guardrail_config.py index 009c72ec..416262b8 100644 --- a/cycode/cli/apps/ai_guardrails/scan/guardrail_config.py +++ b/cycode/cli/apps/ai_guardrails/scan/guardrail_config.py @@ -37,19 +37,11 @@ ) ) -# CLI --ide names to matrix column names; identity for names not listed. -_AGENT_BY_IDE_NAME = {'claude-code': 'claude'} - def get_config_cache_path() -> Path: return Path.home() / CYCODE_CONFIGURATION_DIRECTORY / GUARDRAILS_CONFIG_FILE_NAME -def agent_for_ide(ide_name: Optional[str]) -> str: - ide_name = (ide_name or '').lower() - return _AGENT_BY_IDE_NAME.get(ide_name, ide_name) - - def _default_sensitive_globs() -> list: return list(DEFAULT_SENSITIVE_PATH_GLOBS) @@ -69,8 +61,9 @@ def __post_init__(self) -> None: } def mode_for(self, guardrail_key: str, ide_name: Optional[str]) -> str: + """The platform keys the cells by our --ide names, so the lookup is direct.""" agents = (self._guardrails.get(guardrail_key) or {}).get('agents') or {} - return str(agents.get(agent_for_ide(ide_name), GuardrailCellMode.REPORT.value)).lower() + return str(agents.get((ide_name or '').lower(), GuardrailCellMode.REPORT.value)).lower() def _modes_for_event(self, event_name: str, ide_name: Optional[str]) -> list: return [ diff --git a/cycode/cli/apps/ai_guardrails/scan/handlers.py b/cycode/cli/apps/ai_guardrails/scan/handlers.py index 20254738..54cd92a0 100644 --- a/cycode/cli/apps/ai_guardrails/scan/handlers.py +++ b/cycode/cli/apps/ai_guardrails/scan/handlers.py @@ -13,7 +13,7 @@ from dataclasses import dataclass from multiprocessing.pool import ThreadPool from multiprocessing.pool import TimeoutError as PoolTimeoutError -from typing import TYPE_CHECKING, Callable, Optional +from typing import TYPE_CHECKING, Callable, NamedTuple, Optional import typer @@ -45,12 +45,32 @@ HandlerFn = Callable[[typer.Context, AIHookPayload, dict], HookDecision] +class ScanOutcome(NamedTuple): + """What one guardrail scan came back with; the verdict is the server's, which applied the tenant's floors.""" + + violation_summary: Optional[str] = None + scan_id: Optional[str] = None + verdict: Optional[GuardrailsMode] = None + + +NO_SCAN = ScanOutcome() + + +def _parse_verdict(verdict: Optional[str]) -> Optional[GuardrailsMode]: + """The server spells the verdict "Block"/"Report" and omits it when the scan found nothing to decide on.""" + if not verdict: + return None + try: + return GuardrailsMode(verdict.lower()) + except ValueError: + logger.debug('Ignoring unknown guardrail verdict, %s', {'verdict': verdict}) + return None + + def handle_before_submit_prompt(ctx: typer.Context, payload: AIHookPayload, policy: dict) -> HookDecision: """Scan prompt text for secrets before it's sent to the AI model.""" ai_client = ctx.obj['ai_security_client'] - prompt_config = get_policy_value(policy, 'prompt', default={}) - effective_mode = get_effective_mode(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) @@ -62,20 +82,20 @@ def handle_before_submit_prompt(ctx: typer.Context, payload: AIHookPayload, poli error_message = None try: - violation_summary, scan_id = _scan_text_for_secrets( + scan_outcome = _scan_text_for_secrets( ctx, clipped, timeout_ms, payload=payload, event_type=AiHookEventType.PROMPT, - effective_mode=effective_mode, ) + scan_id = scan_outcome.scan_id - if violation_summary: + if scan_outcome.violation_summary: block_reason = SECRETS_BLOCK_REASON_BY_EVENT_TYPE[AiHookEventType.PROMPT] - if effective_mode == GuardrailsMode.BLOCK: + if scan_outcome.verdict == GuardrailsMode.BLOCK: outcome = AIHookOutcome.BLOCKED - user_message = f'Remove secrets before sending. {violation_summary}' + user_message = f'Remove secrets before sending. {scan_outcome.violation_summary}' return HookDecision.deny(AiHookEventType.PROMPT, user_message) outcome = AIHookOutcome.WARNED return HookDecision.allow(AiHookEventType.PROMPT) @@ -104,7 +124,6 @@ def handle_before_read_file(ctx: typer.Context, payload: AIHookPayload, policy: file_read_config = get_policy_value(policy, 'file_read', default={}) file_path = payload.file_path or '' path_mode = get_effective_mode(file_read_config, action_key='path_action') - content_mode = get_effective_mode(file_read_config) scan_id = None block_reason = None @@ -138,21 +157,20 @@ 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, payload=payload, effective_mode=content_mode - ) - if violation_summary: + scan_outcome = _scan_path_for_secrets(ctx, file_path, policy, payload=payload) + scan_id = scan_outcome.scan_id + if scan_outcome.violation_summary: block_reason = SECRETS_BLOCK_REASON_BY_EVENT_TYPE[AiHookEventType.FILE_READ] - if content_mode == GuardrailsMode.BLOCK: + if scan_outcome.verdict == GuardrailsMode.BLOCK: outcome = AIHookOutcome.BLOCKED - user_message = f'Cycode blocked reading {file_path}. {violation_summary}' + user_message = f'Cycode blocked reading {file_path}. {scan_outcome.violation_summary}' return HookDecision.deny( AiHookEventType.FILE_READ, user_message, 'Secrets detected; do not send this file to the model.', ) outcome = AIHookOutcome.WARNED - user_message = f'Cycode detected secrets in {file_path}. {violation_summary}' + user_message = f'Cycode detected secrets in {file_path}. {scan_outcome.violation_summary}' return HookDecision.ask( AiHookEventType.FILE_READ, user_message, @@ -192,10 +210,9 @@ class _ArgScanFeature: """Configuration for a "scan some text and decide" event. MCP execution and command exec share identical scan-and-decide logic; - only the policy key, event type, and user-facing messages differ. + only the event type and user-facing messages differ. """ - policy_key: str # 'mcp' or 'command_exec' event_type: AiHookEventType deny_message: Callable[[str], str] deny_agent_message: str @@ -213,11 +230,9 @@ def _handle_arg_scan( """Shared scan + decision flow for MCP_EXECUTION and COMMAND_EXEC events.""" ai_client = ctx.obj['ai_security_client'] - feature_config = get_policy_value(policy, feature.policy_key, default={}) 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) - effective_mode = get_effective_mode(feature_config) scan_id = None block_reason = None @@ -225,27 +240,27 @@ def _handle_arg_scan( error_message = None try: - violation_summary, scan_id = _scan_text_for_secrets( + scan_outcome = _scan_text_for_secrets( ctx, clipped, timeout_ms, payload=payload, event_type=feature.event_type, - effective_mode=effective_mode, ) - if violation_summary: + scan_id = scan_outcome.scan_id + if scan_outcome.violation_summary: block_reason = SECRETS_BLOCK_REASON_BY_EVENT_TYPE[feature.event_type] - if effective_mode == GuardrailsMode.BLOCK: + if scan_outcome.verdict == GuardrailsMode.BLOCK: outcome = AIHookOutcome.BLOCKED return HookDecision.deny( feature.event_type, - feature.deny_message(violation_summary), + feature.deny_message(scan_outcome.violation_summary), feature.deny_agent_message, ) outcome = AIHookOutcome.WARNED return HookDecision.ask( feature.event_type, - feature.ask_message(violation_summary), + feature.ask_message(scan_outcome.violation_summary), feature.ask_agent_message, ) @@ -278,7 +293,6 @@ def handle_before_mcp_execution(ctx: typer.Context, payload: AIHookPayload, poli payload, policy, _ArgScanFeature( - policy_key='mcp', event_type=AiHookEventType.MCP_EXECUTION, 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.', @@ -329,11 +343,9 @@ def build_ai_guardrails_scan_parameters( 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(), @@ -360,13 +372,13 @@ def _setup_scan_context(ctx: typer.Context) -> typer.Context: def _perform_scan( ctx: typer.Context, documents: list[Document], scan_parameters: dict, timeout_seconds: float -) -> tuple[Optional[str], Optional[str]]: - """Run a scan on documents, returning (violation_summary, scan_id). +) -> ScanOutcome: + """Run a scan on documents. Raises on scan failure / timeout so the fail-open policy can take over. """ if not documents: - return None, None + return NO_SCAN scan_batch_thread_func = _get_scan_documents_thread_func( ctx, is_git_diff=False, is_commit_range=False, scan_parameters=scan_parameters @@ -377,7 +389,7 @@ def _perform_scan( with ThreadPool(processes=1) as pool: result = pool.apply_async(scan_batch_thread_func, (documents,)) try: - scan_id, error, local_scan_result = result.get(timeout=timeout_seconds) + _, error, local_scan_result = result.get(timeout=timeout_seconds) except PoolTimeoutError: logger.debug('Scan timed out after %s seconds', timeout_seconds) raise RuntimeError(f'Scan timed out after {timeout_seconds} seconds') from None @@ -387,15 +399,14 @@ def _perform_scan( raise RuntimeError(error.message) if not local_scan_result: - return None, None - - scan_id = local_scan_result.scan_id + return NO_SCAN - if local_scan_result.issue_detected: - violation_summary = build_violation_summary([local_scan_result]) - return violation_summary, scan_id - - return None, scan_id + violation_summary = build_violation_summary([local_scan_result]) if local_scan_result.issue_detected else None + return ScanOutcome( + violation_summary=violation_summary, + scan_id=local_scan_result.scan_id, + verdict=_parse_verdict(local_scan_result.verdict), + ) def _scan_text_for_secrets( @@ -404,16 +415,15 @@ def _scan_text_for_secrets( timeout_ms: int, payload: AIHookPayload, event_type: AiHookEventType, - effective_mode: GuardrailsMode, -) -> tuple[Optional[str], Optional[str]]: +) -> ScanOutcome: """Scan text content for secrets using Cycode CLI.""" if not text: - return None, None + return NO_SCAN 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 - scan_parameters = build_ai_guardrails_scan_parameters(scan_ctx, None, payload, event_type, effective_mode) + scan_parameters = build_ai_guardrails_scan_parameters(scan_ctx, None, payload, event_type) return _perform_scan(scan_ctx, [document], scan_parameters, timeout_seconds) @@ -422,15 +432,14 @@ def _scan_path_for_secrets( file_path: str, policy: dict, payload: AIHookPayload, - effective_mode: GuardrailsMode, -) -> tuple[Optional[str], Optional[str]]: +) -> ScanOutcome: """Scan a file path for secrets.""" if not file_path or not os.path.isfile(file_path): - return None, None + return NO_SCAN if is_path_configured_in_exclusions(str(ScanTypeOption.SECRET), os.path.abspath(file_path)): logger.debug('Skipping scan; the path is in the ignore paths list, %s', {'file_path': file_path}) - return None, None + return NO_SCAN max_bytes = get_policy_value(policy, 'secrets', 'max_bytes', default=200000) @@ -442,7 +451,5 @@ def _scan_path_for_secrets( document = Document(path=os.path.basename(file_path), content=content, is_git_diff_format=False) scan_ctx = _setup_scan_context(ctx) - scan_parameters = build_ai_guardrails_scan_parameters( - scan_ctx, (file_path,), payload, AiHookEventType.FILE_READ, effective_mode - ) + scan_parameters = build_ai_guardrails_scan_parameters(scan_ctx, (file_path,), payload, AiHookEventType.FILE_READ) return _perform_scan(scan_ctx, [document], scan_parameters, timeout_seconds) diff --git a/cycode/cli/apps/scan/scan_result.py b/cycode/cli/apps/scan/scan_result.py index 9fb1da1d..015d354c 100644 --- a/cycode/cli/apps/scan/scan_result.py +++ b/cycode/cli/apps/scan/scan_result.py @@ -79,6 +79,7 @@ def create_local_scan_result( issue_detected=len(relevant_document_detections_list) > 0, detections_count=detections_count, relevant_detections_count=relevant_detections_count, + verdict=scan_result.verdict, ) @@ -170,6 +171,7 @@ def get_sync_scan_result(scan_type: str, scan_results: 'ScanResultsSyncFlow') -> did_detect=True, detections_per_file=_map_detections_per_file_and_commit_id(scan_type, scan_results.detection_messages), scan_id=scan_results.id, + verdict=scan_results.verdict, ) diff --git a/cycode/cli/models.py b/cycode/cli/models.py index 3c59eeee..ada31137 100644 --- a/cycode/cli/models.py +++ b/cycode/cli/models.py @@ -58,6 +58,7 @@ class LocalScanResult(NamedTuple): issue_detected: bool detections_count: int relevant_detections_count: int + verdict: Optional[str] = None @dataclass diff --git a/cycode/cyclient/models.py b/cycode/cyclient/models.py index 904fe0ef..b4bf4f7a 100644 --- a/cycode/cyclient/models.py +++ b/cycode/cyclient/models.py @@ -76,6 +76,7 @@ def __init__( report_url: Optional[str] = None, scan_id: Optional[str] = None, err: Optional[str] = None, + verdict: Optional[str] = None, ) -> None: super().__init__() self.did_detect = did_detect @@ -83,6 +84,7 @@ def __init__( self.scan_id = scan_id self.report_url = report_url self.err = err + self.verdict = verdict class ScanResult(Schema): @@ -506,6 +508,7 @@ def build_dto(self, data: dict[str, Any], **_) -> DetectionRule: class ScanResultsSyncFlow: id: str detection_messages: list[dict] + verdict: Optional[str] = None class ScanResultsSyncFlowSchema(Schema): @@ -514,6 +517,7 @@ class Meta: id = fields.String() detection_messages = fields.List(fields.Dict()) + verdict = fields.String(allow_none=True, load_default=None) @post_load def build_dto(self, data: dict[str, Any], **_) -> ScanResultsSyncFlow: diff --git a/tests/cli/commands/ai_guardrails/scan/conftest.py b/tests/cli/commands/ai_guardrails/scan/conftest.py index ddef2df8..86c1b2ae 100644 --- a/tests/cli/commands/ai_guardrails/scan/conftest.py +++ b/tests/cli/commands/ai_guardrails/scan/conftest.py @@ -15,7 +15,7 @@ def resolved_guardrails_payload( return { 'ttl_seconds': 900, 'guardrails': [ - {'key': 'secrets_in_prompt', 'event_type': 'Prompt', 'agents': {'cursor': prompt, 'claude': 'Block'}}, + {'key': 'secrets_in_prompt', 'event_type': 'Prompt', 'agents': {'cursor': prompt, 'claude-code': 'Block'}}, {'key': 'secrets_in_file', 'event_type': 'FileRead', 'agents': {'cursor': file_read}}, { 'key': 'sensitive_path', diff --git a/tests/cli/commands/ai_guardrails/scan/test_guardrail_config.py b/tests/cli/commands/ai_guardrails/scan/test_guardrail_config.py index 66502a01..efde1d26 100644 --- a/tests/cli/commands/ai_guardrails/scan/test_guardrail_config.py +++ b/tests/cli/commands/ai_guardrails/scan/test_guardrail_config.py @@ -62,10 +62,11 @@ def test_expired_cache_detected() -> None: # --- lookups --- -def test_agent_mapping_claude_code_reads_claude_column() -> None: +def test_cells_are_read_by_ide_name() -> None: config = _config() - # cursor column is Report; claude column is Block - the claude-code ide maps onto it. + # cursor column is Report; the claude-code column is Block. assert config.can_event_block('Prompt', 'claude-code') is True + assert config.can_event_block('Prompt', 'Claude-Code') is True assert config.can_event_block('Prompt', 'cursor') is False diff --git a/tests/cli/commands/ai_guardrails/scan/test_handlers.py b/tests/cli/commands/ai_guardrails/scan/test_handlers.py index 47d473bc..0e4acd8a 100644 --- a/tests/cli/commands/ai_guardrails/scan/test_handlers.py +++ b/tests/cli/commands/ai_guardrails/scan/test_handlers.py @@ -10,6 +10,7 @@ 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 ( + ScanOutcome, _perform_scan, _scan_path_for_secrets, _scan_text_for_secrets, @@ -73,7 +74,7 @@ def test_handle_before_submit_prompt_no_secrets( mock_scan: MagicMock, mock_ctx: MagicMock, mock_payload: AIHookPayload, default_policy: dict[str, Any] ) -> None: """Test that prompt with no secrets is allowed.""" - mock_scan.return_value = (None, 'scan-id-123') + mock_scan.return_value = ScanOutcome(scan_id='scan-id-123') result = handle_before_submit_prompt(mock_ctx, mock_payload, default_policy) @@ -91,7 +92,7 @@ def test_handle_before_submit_prompt_with_secrets_blocked( mock_scan: MagicMock, mock_ctx: MagicMock, mock_payload: AIHookPayload, default_policy: dict[str, Any] ) -> None: """Test that prompt with secrets is blocked.""" - mock_scan.return_value = ('Found 1 secret: API key', 'scan-id-456') + mock_scan.return_value = ScanOutcome('Found 1 secret: API key', 'scan-id-456', GuardrailsMode.BLOCK) result = handle_before_submit_prompt(mock_ctx, mock_payload, default_policy) @@ -108,9 +109,9 @@ def test_handle_before_submit_prompt_with_secrets_blocked( def test_handle_before_submit_prompt_with_secrets_warned( mock_scan: MagicMock, mock_ctx: MagicMock, mock_payload: AIHookPayload, default_policy: dict[str, Any] ) -> None: - """Test that prompt with secrets in warn mode is allowed.""" + """Test that prompt with secrets under a Report verdict is allowed.""" default_policy['prompt']['action'] = 'warn' - mock_scan.return_value = ('Found 1 secret: API key', 'scan-id-789') + mock_scan.return_value = ScanOutcome('Found 1 secret: API key', 'scan-id-789', GuardrailsMode.REPORT) result = handle_before_submit_prompt(mock_ctx, mock_payload, default_policy) @@ -199,7 +200,7 @@ def test_handle_before_read_file_no_secrets( ) -> None: """Test that file with no secrets is allowed.""" mock_is_denied.return_value = False - mock_scan.return_value = (None, 'scan-id-123') + mock_scan.return_value = ScanOutcome(scan_id='scan-id-123') payload = AIHookPayload( event_name='FileRead', ide_provider='cursor', @@ -219,9 +220,9 @@ def test_handle_before_read_file_no_secrets( def test_handle_before_read_file_with_secrets( mock_scan: MagicMock, mock_is_denied: MagicMock, mock_ctx: MagicMock, default_policy: dict[str, Any] ) -> None: - """Test that file with secrets is blocked.""" + """Test that file with secrets is blocked, and merely reported when the verdict says so.""" mock_is_denied.return_value = False - mock_scan.return_value = ('Found 1 secret: password', 'scan-id-456') + mock_scan.return_value = ScanOutcome('Found 1 secret: password', 'scan-id-456', GuardrailsMode.BLOCK) payload = AIHookPayload( event_name='FileRead', ide_provider='cursor', @@ -238,14 +239,14 @@ def test_handle_before_read_file_with_secrets( assert call_args.kwargs['block_reason'] == BlockReason.SECRETS_IN_FILE assert call_args.kwargs['file_path'] == '/path/to/file.txt' - # A block-mode path guardrail must not make a report-mode content scan block. - default_policy['file_read']['action'] = 'warn' + # A block-mode path guardrail must not make a Report verdict block; the developer is asked instead default_policy['file_read']['path_action'] = 'block' + mock_scan.return_value = ScanOutcome('Found 1 secret: password', 'scan-id-456', GuardrailsMode.REPORT) result = handle_before_read_file(mock_ctx, payload, default_policy) assert result.action == DecisionAction.ASK - assert mock_scan.call_args.kwargs['effective_mode'] == GuardrailsMode.REPORT + assert 'Found 1 secret: password' in result.user_message assert mock_ctx.obj['ai_security_client'].create_event.call_args.args[2] == AIHookOutcome.WARNED @@ -276,7 +277,7 @@ def test_handle_before_read_file_sensitive_path_warn_mode_scans_content( ) -> None: """Test that sensitive path in warn mode still scans file content and emits two events.""" mock_is_denied.return_value = True - mock_scan.return_value = (None, 'scan-id-123') + mock_scan.return_value = ScanOutcome(scan_id='scan-id-123') default_policy['file_read']['path_action'] = 'warn' payload = AIHookPayload( event_name='FileRead', @@ -307,7 +308,7 @@ def test_handle_before_read_file_sensitive_path_warn_mode_with_secrets( ) -> None: """Test that sensitive path in warn mode reports secrets and emits two events.""" mock_is_denied.return_value = True - mock_scan.return_value = ('Found 1 secret: API key', 'scan-id-456') + mock_scan.return_value = ScanOutcome('Found 1 secret: API key', 'scan-id-456', GuardrailsMode.REPORT) default_policy['file_read']['path_action'] = 'warn' default_policy['file_read']['action'] = 'warn' payload = AIHookPayload( @@ -321,7 +322,7 @@ def test_handle_before_read_file_sensitive_path_warn_mode_with_secrets( mock_scan.assert_called_once() assert result.action == DecisionAction.ASK assert result.event_type == AiHookEventType.FILE_READ - assert 'Found 1 secret: API key' in result.user_message + assert '.env' in result.user_message assert mock_ctx.obj['ai_security_client'].create_event.call_count == 2 first_event = mock_ctx.obj['ai_security_client'].create_event.call_args_list[0] @@ -366,11 +367,9 @@ def test_scan_path_for_secrets_directory( """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, payload=mock_payload, effective_mode=GuardrailsMode.BLOCK - ) + result = _scan_path_for_secrets(mock_ctx, '/path/to/some_directory', default_policy, payload=mock_payload) - assert result == (None, None) + assert result == ScanOutcome() @patch('cycode.cli.apps.ai_guardrails.scan.handlers._perform_scan') @@ -386,17 +385,15 @@ def test_scan_path_for_secrets_skips_path_configured_in_exclusions( excluded_dir = os.path.abspath(os.path.join(os.sep, 'project', 'secrets')) file_path = os.path.join(excluded_dir, 'creds.env') fs.create_file(file_path, contents='password=hunter2') - mock_perform_scan.return_value = ('Cycode found 1 violations', 'scan-id-123') + mock_perform_scan.return_value = ScanOutcome('Cycode found 1 violations', 'scan-id-123') with patch( '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, payload=mock_payload, effective_mode=GuardrailsMode.BLOCK - ) + result = _scan_path_for_secrets(mock_ctx, file_path, default_policy, payload=mock_payload) - assert result == (None, None) + assert result == ScanOutcome() mock_perform_scan.assert_not_called() @@ -409,6 +406,7 @@ def test_perform_scan_no_violation_when_all_detections_excluded(mock_ctx: MagicM issue_detected=False, detections_count=1, relevant_detections_count=0, + verdict='Block', ) document = Document(path='prompt-content.txt', content='some content', is_git_diff_format=False) @@ -416,10 +414,11 @@ def test_perform_scan_no_violation_when_all_detections_excluded(mock_ctx: MagicM 'cycode.cli.apps.ai_guardrails.scan.handlers._get_scan_documents_thread_func', return_value=lambda batch: ('scan-id-123', None, local_scan_result), ): - violation_summary, scan_id = _perform_scan(mock_ctx, [document], {}, timeout_seconds=5.0) + scan_outcome = _perform_scan(mock_ctx, [document], {}, timeout_seconds=5.0) - assert violation_summary is None - assert scan_id == 'scan-id-123' + assert scan_outcome.violation_summary is None + assert scan_outcome.scan_id == 'scan-id-123' + assert scan_outcome.verdict == GuardrailsMode.BLOCK def _local_scan_result_with_detections(*shas: str) -> LocalScanResult: @@ -484,7 +483,7 @@ def test_handle_before_mcp_execution_no_secrets( mock_scan: MagicMock, mock_ctx: MagicMock, default_policy: dict[str, Any] ) -> None: """Test that MCP execution with no secrets is allowed.""" - mock_scan.return_value = (None, 'scan-id-123') + mock_scan.return_value = ScanOutcome(scan_id='scan-id-123') payload = AIHookPayload( event_name='McpExecution', ide_provider='cursor', @@ -504,7 +503,7 @@ def test_handle_before_mcp_execution_with_secrets_blocked( mock_scan: MagicMock, mock_ctx: MagicMock, default_policy: dict[str, Any] ) -> None: """Test that MCP execution with secrets is blocked.""" - mock_scan.return_value = ('Found 1 secret: token', 'scan-id-456') + mock_scan.return_value = ScanOutcome('Found 1 secret: token', 'scan-id-456', GuardrailsMode.BLOCK) payload = AIHookPayload( event_name='McpExecution', ide_provider='cursor', @@ -527,7 +526,7 @@ def test_handle_before_mcp_execution_with_secrets_warned( mock_scan: MagicMock, mock_ctx: MagicMock, default_policy: dict[str, Any] ) -> None: """Test that MCP execution with secrets in warn mode asks permission.""" - mock_scan.return_value = ('Found 1 secret: token', 'scan-id-789') + mock_scan.return_value = ScanOutcome('Found 1 secret: token', 'scan-id-789', GuardrailsMode.REPORT) default_policy['mcp']['action'] = 'warn' payload = AIHookPayload( event_name='McpExecution', @@ -563,13 +562,10 @@ def test_build_ai_guardrails_scan_parameters( """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 - ) + params = build_ai_guardrails_scan_parameters(mock_ctx, None, mock_payload, AiHookEventType.PROMPT) 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', @@ -590,7 +586,7 @@ def test_scan_text_for_secrets_injects_ai_guardrails_scan_parameter( ) -> 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') + mock_perform_scan.return_value = ScanOutcome(scan_id='scan-id-123') _scan_text_for_secrets( mock_ctx, @@ -598,11 +594,9 @@ def test_scan_text_for_secrets_injects_ai_guardrails_scan_parameter( 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' diff --git a/tests/test_models_deserialization.py b/tests/test_models_deserialization.py index 4c7dcd72..36d67207 100644 --- a/tests/test_models_deserialization.py +++ b/tests/test_models_deserialization.py @@ -401,6 +401,10 @@ def test_scan_results_sync_flow_schema_load() -> None: assert isinstance(result, ScanResultsSyncFlow) assert result.id == 'sync-123' assert len(result.detection_messages) == 2 + assert result.verdict is None + + raw['verdict'] = 'Block' + assert ScanResultsSyncFlowSchema().load(raw).verdict == 'Block' # --- SupportedModulesPreferencesSchema ---