diff --git a/cycode/cli/apps/ai_guardrails/hooks_manager.py b/cycode/cli/apps/ai_guardrails/hooks_manager.py index 73867475..13ab3e71 100644 --- a/cycode/cli/apps/ai_guardrails/hooks_manager.py +++ b/cycode/cli/apps/ai_guardrails/hooks_manager.py @@ -12,9 +12,9 @@ import yaml -from cycode.cli.apps.ai_guardrails.consts import PolicyMode from cycode.cli.apps.ai_guardrails.ides.base import IDE from cycode.cli.apps.ai_guardrails.scan.consts import DEFAULT_POLICY, POLICY_FILE_NAME +from cycode.cli.apps.ai_guardrails.scan.policy import strip_platform_managed_keys from cycode.logger import get_logger logger = get_logger('AI Guardrails Hooks') @@ -102,22 +102,23 @@ def _load_policy_dict(policy_path: Path) -> dict: return {**copy.deepcopy(DEFAULT_POLICY), **existing} -def create_policy_file(scope: str, mode: PolicyMode, repo_path: Optional[Path] = None) -> tuple[bool, str]: - """Create or update the ai-guardrails.yaml policy file. +def create_policy_file(scope: str, repo_path: Optional[Path] = None) -> tuple[bool, str]: + """Create or update the ai-guardrails.yaml policy file (operational knobs only). - If the file already exists, only the mode field is updated; otherwise a new - file is created from the default policy. + Enforcement mode and sensitive-path globs are platform-managed; those keys are stripped + (including ones an older CLI wrote), everything else the user customized is preserved. """ config_dir = repo_path / '.cycode' if scope == 'repo' and repo_path else Path.home() / '.cycode' policy_path = config_dir / POLICY_FILE_NAME - policy = _load_policy_dict(policy_path) - policy['mode'] = mode.value + policy = strip_platform_managed_keys(_load_policy_dict(policy_path), str(policy_path)) + # Sections left empty by the stripping have nothing for the user to edit. + policy = {key: value for key, value in policy.items() if value != {}} try: config_dir.mkdir(parents=True, exist_ok=True) policy_path.write_text(yaml.dump(policy, default_flow_style=False, sort_keys=False), encoding='utf-8') - return True, f'AI guardrails policy ({mode.value} mode) set: {policy_path}' + return True, f'AI guardrails policy file set: {policy_path}' except Exception as e: logger.error('Failed to create policy file', exc_info=e) return False, f'Failed to create policy file: {policy_path}' diff --git a/cycode/cli/apps/ai_guardrails/install_command.py b/cycode/cli/apps/ai_guardrails/install_command.py index 48fb090c..c1407905 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 GuardrailsMode, PolicyMode +from cycode.cli.apps.ai_guardrails.consts import GuardrailsMode 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 @@ -44,8 +44,7 @@ def install_command( typer.Option( '--mode', '-m', - help='Installation mode: "report" for async non-blocking hooks with warn policy, ' - '"block" for sync blocking hooks.', + help='[Deprecated] Enforcement mode is platform-managed; configure guardrails in the Cycode platform.', ), ] = GuardrailsMode.REPORT, ) -> None: @@ -80,32 +79,34 @@ def install_command( console.print(f'[red]✗[/] {message}', style='bold red') all_success = False + if mode == GuardrailsMode.BLOCK: + console.print( + '[yellow]--mode is deprecated:[/] enforcement mode is platform-managed; ' + 'configure guardrails in the Cycode platform.' + ) + if any_success: - policy_mode = PolicyMode.WARN if mode == GuardrailsMode.REPORT else PolicyMode.BLOCK - _install_policy(scope, repo_path, policy_mode) - _print_next_steps(results, mode) + _install_policy(scope, repo_path) + _print_next_steps(results) if not all_success: raise typer.Exit(1) -def _install_policy(scope: str, repo_path: Optional[Path], policy_mode: PolicyMode) -> None: - policy_success, policy_message = create_policy_file(scope, policy_mode, repo_path) +def _install_policy(scope: str, repo_path: Optional[Path]) -> None: + policy_success, policy_message = create_policy_file(scope, repo_path) if policy_success: console.print(f'[green]✓[/] {policy_message}') else: console.print(f'[red]✗[/] {policy_message}', style='bold red') -def _print_next_steps(results: list[tuple[str, bool, str]], mode: GuardrailsMode) -> None: +def _print_next_steps(results: list[tuple[str, bool, str]]) -> None: console.print() console.print('[bold]Next steps:[/]') successful_ides = [name for name, success, _ in results if success] ide_list = ', '.join(successful_ides) console.print(f'1. Restart {ide_list} to activate the hooks') - console.print('2. (Optional) Customize policy in ~/.cycode/ai-guardrails.yaml') + console.print('2. Configure guardrail enforcement in the Cycode platform') console.print() - if mode == GuardrailsMode.REPORT: - console.print('[dim]Report mode: policy is set to warn.[/]') - else: - console.print('[dim]The hooks will scan prompts, file reads, and MCP tool calls for secrets.[/]') + 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/consts.py b/cycode/cli/apps/ai_guardrails/scan/consts.py index 007892a8..d17d5bda 100644 --- a/cycode/cli/apps/ai_guardrails/scan/consts.py +++ b/cycode/cli/apps/ai_guardrails/scan/consts.py @@ -12,7 +12,6 @@ # Default policy configuration DEFAULT_POLICY = { 'version': 1, - 'mode': 'block', # block | warn 'fail_open': True, # allow if scan fails/timeouts 'secrets': { 'scan_type': 'secret', diff --git a/cycode/cli/apps/ai_guardrails/scan/guardrail_config.py b/cycode/cli/apps/ai_guardrails/scan/guardrail_config.py new file mode 100644 index 00000000..d9f92c98 --- /dev/null +++ b/cycode/cli/apps/ai_guardrails/scan/guardrail_config.py @@ -0,0 +1,170 @@ +"""Platform guardrail config cache. + +session-start fetches the tenant's resolved guardrail config from the platform and writes it +here; scans only read. Per-agent modes and sensitive-path globs are platform-owned - local +policy files never carry them. An absent or corrupt cache means built-in defaults (Report +everywhere + the default globs), always synchronous. +""" + +import json +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + +from cycode.cli.apps.ai_guardrails.consts import GuardrailsMode, PolicyMode +from cycode.cli.apps.ai_guardrails.scan.consts import DEFAULT_POLICY +from cycode.cli.apps.ai_guardrails.scan.types import BlockReason +from cycode.cli.consts import CYCODE_CONFIGURATION_DIRECTORY +from cycode.cli.utils.path_utils import atomic_write_text, quarantine_corrupt_file +from cycode.logger import get_logger + +logger = get_logger('AI Guardrails') + +GUARDRAILS_CONFIG_FILE_NAME = 'ai-guardrails-config.json' + +# The matrix cell values; Report and Block share GuardrailsMode's spelling. +MODE_OFF = 'off' +MODE_REPORT = GuardrailsMode.REPORT.value +MODE_BLOCK = GuardrailsMode.BLOCK.value + +_DEFAULT_TTL_SECONDS = 900 + +# Guardrail keys are the CLI's block-reason vocabulary. Anything else in the payload (a future +# guardrail this CLI doesn't implement) is ignored - unknown config must never fail closed. +_KNOWN_GUARDRAIL_KEYS = frozenset( + reason.value + for reason in ( + BlockReason.SECRETS_IN_PROMPT, + BlockReason.SECRETS_IN_FILE, + BlockReason.SENSITIVE_PATH, + BlockReason.SECRETS_IN_MCP_ARGS, + ) +) + +# 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_POLICY['file_read']['deny_globs']) + + +@dataclass +class GuardrailConfig: + payload: dict + fetched_at: float + tenant_id: Optional[str] = None + _guardrails: dict = field(init=False, repr=False) + + def __post_init__(self) -> None: + self._guardrails = { + guardrail.get('key'): guardrail + for guardrail in self.payload.get('guardrails') or [] + if guardrail.get('key') in _KNOWN_GUARDRAIL_KEYS + } + + def mode_for(self, guardrail_key: str, ide_name: Optional[str]) -> str: + agents = (self._guardrails.get(guardrail_key) or {}).get('agents') or {} + return str(agents.get(agent_for_ide(ide_name), MODE_REPORT)).lower() + + def _modes_for_event(self, event_name: str, ide_name: Optional[str]) -> list: + return [ + self.mode_for(key, ide_name) + for key, guardrail in self._guardrails.items() + if str(guardrail.get('event_type', '')).lower() == str(event_name).lower() + ] + + def is_event_off(self, event_name: str, ide_name: Optional[str]) -> bool: + """Every guardrail for this event is Off - skip the scan entirely.""" + modes = self._modes_for_event(event_name, ide_name) + return bool(modes) and all(mode == MODE_OFF for mode in modes) + + def can_event_block(self, event_name: str, ide_name: Optional[str]) -> bool: + """At least one guardrail for this event is in Block mode - the scan must stay synchronous.""" + return MODE_BLOCK in self._modes_for_event(event_name, ide_name) + + def sensitive_globs(self) -> list: + settings = (self._guardrails.get(BlockReason.SENSITIVE_PATH) or {}).get('settings') or {} + globs = settings.get('globs') + return globs if isinstance(globs, list) and globs else _default_sensitive_globs() + + def is_expired(self) -> bool: + ttl = self.payload.get('ttl_seconds') or _DEFAULT_TTL_SECONDS + return time.time() - self.fetched_at > ttl + + def needs_refresh(self, tenant_id: Optional[str]) -> bool: + """Expired, or fetched for another tenant (the user switched tenants since).""" + return self.is_expired() or self.tenant_id != tenant_id + + +def apply_platform_config(policy: dict, config: Optional[GuardrailConfig], ide_name: Optional[str]) -> None: + """Overlay the platform-owned enforcement config onto the local knobs-only policy. + + The platform is the only mode source: no cache (cold start) means the built-in defaults - + Report everywhere with the default globs - which equal an unconfigured tenant's platform + config, so behaviour is uniform either way. Each matrix cell lands on its own per-feature + action, so the two FileRead guardrails (content scan vs. sensitive path) keep independent modes. + An all-Off event never reaches here (scan_command skips it), so `enabled` stays untouched. + """ + + def cell(guardrail_key: str) -> str: + return config.mode_for(guardrail_key, ide_name) if config is not None else MODE_REPORT + + def action(guardrail_key: str) -> str: + return PolicyMode.BLOCK.value if cell(guardrail_key) == MODE_BLOCK else PolicyMode.WARN.value + + policy.setdefault('prompt', {})['action'] = action(BlockReason.SECRETS_IN_PROMPT) + + file_read = policy.setdefault('file_read', {}) + file_read['scan_content'] = cell(BlockReason.SECRETS_IN_FILE) != MODE_OFF + file_read['action'] = action(BlockReason.SECRETS_IN_FILE) + file_read['deny_globs'] = ( + (config.sensitive_globs() if config is not None else _default_sensitive_globs()) + if cell(BlockReason.SENSITIVE_PATH) != MODE_OFF + else [] + ) + file_read['path_action'] = action(BlockReason.SENSITIVE_PATH) + + policy.setdefault('mcp', {})['action'] = action(BlockReason.SECRETS_IN_MCP_ARGS) + + +def save_guardrail_config(payload: dict, tenant_id: Optional[str]) -> None: + """Persist a fetched resolved config; a failed write just leaves the previous cache in place.""" + path = get_config_cache_path() + content = {'fetched_at': time.time(), 'tenant_id': tenant_id, 'payload': payload} + try: + path.parent.mkdir(parents=True, exist_ok=True) + atomic_write_text(str(path), json.dumps(content)) + except Exception as e: + logger.debug('Failed to save guardrail config cache', exc_info=e) + + +def load_guardrail_config() -> Optional[GuardrailConfig]: + """The cached platform config, or None when it is absent or corrupt (quarantined).""" + path = get_config_cache_path() + if not path.exists(): + return None + + try: + with open(path, encoding='UTF-8') as file: + content = json.load(file) + payload = content['payload'] + if not isinstance(payload, dict): + raise ValueError('payload is not an object') + return GuardrailConfig( + payload=payload, fetched_at=float(content['fetched_at']), tenant_id=content.get('tenant_id') + ) + except Exception as e: + logger.warning('Guardrail config cache is corrupt and will be moved aside', exc_info=e) + quarantine_corrupt_file(str(path)) + return None diff --git a/cycode/cli/apps/ai_guardrails/scan/handlers.py b/cycode/cli/apps/ai_guardrails/scan/handlers.py index cbe50637..b9c7f606 100644 --- a/cycode/cli/apps/ai_guardrails/scan/handlers.py +++ b/cycode/cli/apps/ai_guardrails/scan/handlers.py @@ -13,10 +13,13 @@ from dataclasses import dataclass from multiprocessing.pool import ThreadPool from multiprocessing.pool import TimeoutError as PoolTimeoutError -from typing import Callable, Optional +from typing import TYPE_CHECKING, Callable, Optional import typer +if TYPE_CHECKING: + from cycode.cli.apps.ai_guardrails.scan.guardrail_config import GuardrailConfig + 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 @@ -52,7 +55,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) - effective_mode = get_effective_mode(policy, prompt_config) + 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) @@ -109,7 +112,9 @@ def handle_before_read_file(ctx: typer.Context, payload: AIHookPayload, policy: return HookDecision.allow(AiHookEventType.FILE_READ) file_path = payload.file_path or '' - effective_mode = get_effective_mode(policy, file_read_config) + # Two guardrails share this event, each with its own mode: the path match and the content scan. + 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 @@ -120,7 +125,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 effective_mode == GuardrailsMode.BLOCK: + if path_mode == GuardrailsMode.BLOCK: outcome = AIHookOutcome.BLOCKED user_message = f'Cycode blocked sending {file_path} to the AI (sensitive path policy).' return HookDecision.deny( @@ -144,11 +149,11 @@ def handle_before_read_file(ctx: typer.Context, payload: AIHookPayload, policy: 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=effective_mode + ctx, file_path, policy, payload=payload, effective_mode=content_mode ) if violation_summary: block_reason = SECRETS_BLOCK_REASON_BY_EVENT_TYPE[AiHookEventType.FILE_READ] - if effective_mode == GuardrailsMode.BLOCK: + if content_mode == GuardrailsMode.BLOCK: outcome = AIHookOutcome.BLOCKED user_message = f'Cycode blocked reading {file_path}. {violation_summary}' return HookDecision.deny( @@ -227,7 +232,7 @@ def _handle_arg_scan( 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(policy, feature_config) + effective_mode = get_effective_mode(feature_config) scan_id = None block_reason = None @@ -311,35 +316,29 @@ 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 get_effective_mode(feature_config: dict, action_key: str = 'action') -> GuardrailsMode: + """A guardrail's action is its matrix cell: block, or warn (report) for everything else.""" + action = get_policy_value(feature_config, action_key, default=PolicyMode.BLOCK) + return GuardrailsMode.BLOCK if action == PolicyMode.BLOCK else GuardrailsMode.REPORT -# The policy section each event's handler reads its feature config from. -_FEATURE_KEY_BY_EVENT_TYPE: dict[str, str] = { - AiHookEventType.PROMPT.value: 'prompt', - AiHookEventType.FILE_READ.value: 'file_read', - AiHookEventType.MCP_EXECUTION.value: 'mcp', -} - - -def should_detach_scan(policy: dict, event_name: str) -> bool: +def should_detach_scan( + config: Optional['GuardrailConfig'], + policy: dict, + event_name: str, + ide_name: Optional[str], +) -> bool: """Whether this event's scan is safe to run detached. - Report mode never blocks, so nobody consumes the verdict. Fail-closed - configs stay synchronous even in report mode: their deny on scan failure - must reach the IDE. Unknown events stay synchronous - they exit fast anyway. + Report mode never blocks, so nobody consumes the verdict. The platform config is the only + mode source: without a cache the scan stays synchronous (never detach on an assumption). + Fail-closed configs also stay synchronous: their deny on scan failure must reach the IDE. """ - feature_key = _FEATURE_KEY_BY_EVENT_TYPE.get(event_name) - if feature_key is None: + if config is None: return False if not get_policy_value(policy, 'fail_open', default=True): return False - feature_config = get_policy_value(policy, feature_key, default={}) - return get_effective_mode(policy, feature_config) == GuardrailsMode.REPORT + return not config.can_event_block(event_name, ide_name) def build_ai_guardrails_scan_parameters( diff --git a/cycode/cli/apps/ai_guardrails/scan/policy.py b/cycode/cli/apps/ai_guardrails/scan/policy.py index 96c45574..51824fd5 100644 --- a/cycode/cli/apps/ai_guardrails/scan/policy.py +++ b/cycode/cli/apps/ai_guardrails/scan/policy.py @@ -17,6 +17,38 @@ import yaml from cycode.cli.apps.ai_guardrails.scan.consts import DEFAULT_POLICY, POLICY_FILE_NAME +from cycode.logger import get_logger + +logger = get_logger('AI Guardrails') + +# Enforcement policy is platform-owned (fetched at session-start and cached): local files may +# only carry operational knobs (secrets.timeout_ms/max_bytes, fail_open). Every key here is +# either overwritten from the platform config on each scan or no longer read at all (`mode`), +# so a local value would be silently dead. +_PLATFORM_MANAGED_TOP_KEYS = ('mode',) +_PLATFORM_MANAGED_FEATURE_KEYS = { + 'prompt': ('enabled', 'action'), + 'file_read': ('enabled', 'action', 'path_action', 'scan_content', 'deny_globs'), + 'mcp': ('enabled', 'action'), +} + + +def strip_platform_managed_keys(config: dict, filename: str) -> dict: + """Drop enforcement keys a local file may carry (older CLIs wrote them).""" + stripped = [key for key in _PLATFORM_MANAGED_TOP_KEYS if config.pop(key, None) is not None] + + for feature, keys in _PLATFORM_MANAGED_FEATURE_KEYS.items(): + feature_config = config.get(feature) + if not isinstance(feature_config, dict): + continue + stripped.extend(f'{feature}.{key}' for key in keys if feature_config.pop(key, None) is not None) + + if stripped: + logger.debug( + 'Ignoring platform-managed keys in local policy file, %s', + {'filename': filename, 'keys': stripped}, + ) + return config def get_machine_policy_path() -> Path: @@ -83,21 +115,22 @@ def load_policy(workspace_root: Optional[str] = None) -> dict: policy = load_defaults() # Merge machine-wide config (admin/MDM-provisioned) - overrides defaults, below user/repo. - machine_config = load_yaml_file(get_machine_policy_path()) + machine_policy_path = get_machine_policy_path() + machine_config = load_yaml_file(machine_policy_path) if machine_config: - policy = deep_merge(policy, machine_config) + policy = deep_merge(policy, strip_platform_managed_keys(machine_config, str(machine_policy_path))) # Merge user-level config (if exists) user_policy_path = Path.home() / '.cycode' / POLICY_FILE_NAME user_config = load_yaml_file(user_policy_path) if user_config: - policy = deep_merge(policy, user_config) + policy = deep_merge(policy, strip_platform_managed_keys(user_config, str(user_policy_path))) # Merge repo-level config (if exists) - highest precedence if workspace_root: repo_policy_path = Path(workspace_root) / '.cycode' / POLICY_FILE_NAME repo_config = load_yaml_file(repo_policy_path) if repo_config: - policy = deep_merge(policy, repo_config) + policy = deep_merge(policy, strip_platform_managed_keys(repo_config, str(repo_policy_path))) return policy diff --git a/cycode/cli/apps/ai_guardrails/scan/scan_command.py b/cycode/cli/apps/ai_guardrails/scan/scan_command.py index 5c3b2513..e81d1ae3 100644 --- a/cycode/cli/apps/ai_guardrails/scan/scan_command.py +++ b/cycode/cli/apps/ai_guardrails/scan/scan_command.py @@ -14,8 +14,9 @@ import typer from cycode.cli.apps.ai_guardrails.ides import DEFAULT_IDE_NAME, get_ide -from cycode.cli.apps.ai_guardrails.ides.base import HookDecision +from cycode.cli.apps.ai_guardrails.ides.base import IDE, HookDecision from cycode.cli.apps.ai_guardrails.scan.detach import is_detached_child, respawn_detached +from cycode.cli.apps.ai_guardrails.scan.guardrail_config import apply_platform_config, load_guardrail_config from cycode.cli.apps.ai_guardrails.scan.handlers import get_handler_for_event, should_detach_scan from cycode.cli.apps.ai_guardrails.scan.policy import load_policy from cycode.cli.apps.ai_guardrails.scan.types import AiHookEventType @@ -61,6 +62,31 @@ def _deny_for_event( return HookDecision.deny(target, user_message, agent_message) +def _should_skip_payload(ide_integration: IDE, payload: Optional[dict]) -> bool: + """Fast exits that never scan: empty/foreign/synthetic payloads all answer a plain allow.""" + if not payload: + logger.debug('Empty or invalid JSON payload received') + return True + + # Prevent cross-IDE processing (e.g. Cursor reading Claude Code hooks + # from ~/.claude/settings.json). + if not ide_integration.matches_payload(payload): + logger.debug( + 'Payload event does not match expected IDE, skipping', + extra={'hook_event_name': payload.get('hook_event_name'), 'expected_ide': ide_integration.name}, + ) + return True + + # Fork/subagent completions arrive as synthetic user turns (e.g. Claude Code's + # ); they are agent-generated, not user prompts - skip before + # parse_hook_payload, which reads the transcript and IDE config from disk. + if ide_integration.is_synthetic_prompt(payload): + logger.debug('Synthetic prompt detected, skipping scan') + return True + + return False + + def _initialize_clients(ctx: typer.Context) -> None: """Initialize API clients. @@ -95,26 +121,7 @@ def scan_command( stdin_data = read_stdin_text().strip() payload = safe_json_parse(stdin_data) - if not payload: - logger.debug('Empty or invalid JSON payload received') - output_json(ide_integration.build_hook_response(HookDecision.allow(AiHookEventType.PROMPT))) - return - - # Prevent cross-IDE processing (e.g. Cursor reading Claude Code hooks - # from ~/.claude/settings.json). - if not ide_integration.matches_payload(payload): - logger.debug( - 'Payload event does not match expected IDE, skipping', - extra={'hook_event_name': payload.get('hook_event_name'), 'expected_ide': ide_integration.name}, - ) - output_json(ide_integration.build_hook_response(HookDecision.allow(AiHookEventType.PROMPT))) - return - - # Fork/subagent completions arrive as synthetic user turns (e.g. Claude Code's - # ); they are agent-generated, not user prompts - skip before - # parse_hook_payload, which reads the transcript and IDE config from disk. - if ide_integration.is_synthetic_prompt(payload): - logger.debug('Synthetic prompt detected, skipping scan') + if _should_skip_payload(ide_integration, payload): output_json(ide_integration.build_hook_response(HookDecision.allow(AiHookEventType.PROMPT))) return @@ -136,14 +143,27 @@ def scan_command( output_json(ide_integration.build_hook_response(HookDecision.allow(AiHookEventType.PROMPT))) return + # Scans only read the cache; session-start refreshes it, so the hot path never waits on the + # network. Every guardrail for this event Off: skip entirely - no scan, no event, no auth. + config = load_guardrail_config() + if config is not None and config.is_event_off(event_name, ide_integration.name): + logger.debug('Guardrails are off for this event, allowing', extra={'event_name': event_name}) + output_json(ide_integration.build_hook_response(HookDecision.allow(AiHookEventType.PROMPT))) + return + # `or` (not a .get default) - Cursor sends workspace_roots=[] when no folder is open. workspace_roots = payload.get('workspace_roots') or ['.'] policy = load_policy(workspace_roots[0]) + apply_platform_config(policy, config, ide_integration.name) # Report mode: nobody consumes the verdict, so hand the scan to a detached # child and release the IDE immediately. Runs before any client or network # work. A failed respawn falls through to the synchronous path. - if not is_detached_child() and should_detach_scan(policy, event_name) and respawn_detached(stdin_data): + if ( + not is_detached_child() + and should_detach_scan(config, policy, event_name, ide_integration.name) + and respawn_detached(stdin_data) + ): return try: diff --git a/cycode/cli/apps/ai_guardrails/session_start_command.py b/cycode/cli/apps/ai_guardrails/session_start_command.py index bebd421d..131c449e 100644 --- a/cycode/cli/apps/ai_guardrails/session_start_command.py +++ b/cycode/cli/apps/ai_guardrails/session_start_command.py @@ -10,6 +10,7 @@ import typer from cycode.cli.apps.ai_guardrails.ides import DEFAULT_IDE_NAME, collect_all_session_contexts, get_ide +from cycode.cli.apps.ai_guardrails.scan.guardrail_config import load_guardrail_config, save_guardrail_config from cycode.cli.apps.ai_guardrails.scan.utils import read_stdin_text, safe_json_parse from cycode.cli.apps.auth.auth_common import get_authorization_info from cycode.cli.apps.auth.auth_manager import AuthManager @@ -158,3 +159,23 @@ def session_start_command( # Report session context (device + cross-IDE MCP servers and plugins) _report_session_context(ai_client, session_payload.ide_user_email, auth_info.tenant_id) + + # SessionStart precedes the first prompt hook in every IDE, so scans normally find a cache. + _sync_guardrail_config(ai_client, auth_info.tenant_id) + + +def _sync_guardrail_config(ai_client: 'AISecurityManagerClient', tenant_id: Optional[str]) -> None: + """Refresh the guardrail config cache when it is expired or belongs to another tenant. + + Every step here swallows its own failures - a broken cache or an unreachable platform + must never fail the session. + """ + cached = load_guardrail_config() + if cached is not None and not cached.needs_refresh(tenant_id): + logger.debug('Guardrail config cache is fresh, skipping fetch') + return + + resolved = ai_client.get_resolved_guardrails() + if resolved: + save_guardrail_config(resolved, tenant_id) + logger.debug('Guardrail config cache updated') diff --git a/cycode/cli/utils/path_utils.py b/cycode/cli/utils/path_utils.py index c2d59805..ea6d2f38 100644 --- a/cycode/cli/utils/path_utils.py +++ b/cycode/cli/utils/path_utils.py @@ -1,5 +1,6 @@ import json import os +import tempfile from functools import cache from typing import TYPE_CHECKING, AnyStr, Optional, Union @@ -86,6 +87,32 @@ def get_file_content(file_path: Union[str, 'PathLike']) -> Optional[AnyStr]: logger.warn('Permission denied to read the file: %s', file_path) +def atomic_write_text(filename: str, content: str) -> None: + """Write via a temp file + rename so concurrent CLI processes never read a torn file.""" + directory = os.path.dirname(filename) + file_descriptor, temp_filename = tempfile.mkstemp(dir=directory, prefix=f'.{os.path.basename(filename)}.') + try: + with os.fdopen(file_descriptor, 'w', encoding='UTF-8') as file: + file.write(content) + file.flush() + os.fsync(file.fileno()) + + os.replace(temp_filename, filename) + except Exception: + if os.path.exists(temp_filename): + os.remove(temp_filename) + raise + + +def quarantine_corrupt_file(filename: str) -> None: + # Renamed rather than deleted: the file may hold the only copy of the user's credentials, + # and keeping it around leaves something to look at in the next bug report. + try: + os.replace(filename, f'{filename}.corrupt') + except OSError as e: + logger.warning('Failed to quarantine corrupt file, %s', {'filename': filename}, exc_info=e) + + def load_json(txt: str) -> Optional[dict]: try: return json.loads(txt) diff --git a/cycode/cli/utils/yaml_utils.py b/cycode/cli/utils/yaml_utils.py index dba4719e..73a61673 100644 --- a/cycode/cli/utils/yaml_utils.py +++ b/cycode/cli/utils/yaml_utils.py @@ -1,10 +1,10 @@ import os -import tempfile from collections.abc import Hashable from typing import Any, TextIO import yaml +from cycode.cli.utils.path_utils import atomic_write_text, quarantine_corrupt_file from cycode.logger import get_logger logger = get_logger('YAML Utils') @@ -35,15 +35,6 @@ def _yaml_object_safe_load(file: TextIO) -> dict[Hashable, Any]: return loaded_file -def _quarantine_corrupt_file(filename: str) -> None: - # Renamed rather than deleted: the file may hold the only copy of the user's credentials, - # and keeping it around leaves something to look at in the next bug report. - try: - os.replace(filename, f'{filename}.corrupt') - except OSError as e: - logger.warning('Failed to quarantine corrupt file, %s', {'filename': filename}, exc_info=e) - - def read_yaml_file(filename: str) -> dict[Hashable, Any]: if not os.access(filename, os.R_OK) or not os.path.exists(filename): logger.debug('Config file is not accessible or does not exist: %s', {'filename': filename}) @@ -54,7 +45,7 @@ def read_yaml_file(filename: str) -> dict[Hashable, Any]: return _yaml_object_safe_load(file) except yaml.YAMLError as e: logger.warning('Config file is corrupt and will be moved aside, %s', {'filename': filename}, exc_info=e) - _quarantine_corrupt_file(filename) + quarantine_corrupt_file(filename) return {} @@ -64,19 +55,7 @@ def write_yaml_file(filename: str, content: dict[Hashable, Any]) -> None: logger.warning('No write permission for file. Cannot save config, %s', {'filename': filename}) return - # Atomic write to avoid race conditions between concurrent CLI processes - file_descriptor, temp_filename = tempfile.mkstemp(dir=directory, prefix=f'.{os.path.basename(filename)}.') - try: - with os.fdopen(file_descriptor, 'w', encoding='UTF-8') as file: - yaml.safe_dump(content, file) - file.flush() - os.fsync(file.fileno()) - - os.replace(temp_filename, filename) - except Exception: - if os.path.exists(temp_filename): - os.remove(temp_filename) - raise + atomic_write_text(filename, yaml.safe_dump(content)) def update_yaml_file(filename: str, content: dict[Hashable, Any]) -> None: diff --git a/cycode/cyclient/ai_security_manager_client.py b/cycode/cyclient/ai_security_manager_client.py index 62f5618b..a1ef7206 100644 --- a/cycode/cyclient/ai_security_manager_client.py +++ b/cycode/cyclient/ai_security_manager_client.py @@ -18,6 +18,7 @@ class AISecurityManagerClient: _CONVERSATIONS_PATH = 'v4/ai-security/interactions/conversations' _EVENTS_PATH = 'v4/ai-security/interactions/events' _SESSION_CONTEXT_PATH = 'v4/ai-security/interactions/session-context' + _RESOLVED_GUARDRAILS_PATH = 'v4/ai-security/guardrails/resolved' def __init__(self, client: CycodeClientBase, service_config: 'AISecurityManagerServiceConfigBase') -> None: self.client = client @@ -92,6 +93,15 @@ def create_event( logger.debug('Failed to create AI hook event', exc_info=e) # Don't fail the hook if tracking fails + def get_resolved_guardrails(self) -> Optional[dict]: + """Fetch the tenant's resolved guardrail config (per-agent modes + sensitive-path globs).""" + try: + response = self.client.get(self._build_endpoint_path(self._RESOLVED_GUARDRAILS_PATH)) + return response.json() + except Exception as e: + logger.debug('Failed to fetch resolved guardrail config', exc_info=e) + return None + def report_session_context( self, hostname: Optional[str] = None, diff --git a/tests/cli/commands/ai_guardrails/scan/conftest.py b/tests/cli/commands/ai_guardrails/scan/conftest.py new file mode 100644 index 00000000..ddef2df8 --- /dev/null +++ b/tests/cli/commands/ai_guardrails/scan/conftest.py @@ -0,0 +1,36 @@ +import time +from typing import Optional + +from cycode.cli.apps.ai_guardrails.scan.guardrail_config import GuardrailConfig + + +def resolved_guardrails_payload( + prompt: str = 'Report', + file_read: str = 'Report', + sensitive_path: str = 'Report', + mcp: str = 'Report', + globs: Optional[list] = None, +) -> dict: + """A platform resolved-config payload with the given per-guardrail modes for the cursor agent.""" + return { + 'ttl_seconds': 900, + 'guardrails': [ + {'key': 'secrets_in_prompt', 'event_type': 'Prompt', 'agents': {'cursor': prompt, 'claude': 'Block'}}, + {'key': 'secrets_in_file', 'event_type': 'FileRead', 'agents': {'cursor': file_read}}, + { + 'key': 'sensitive_path', + 'event_type': 'FileRead', + 'agents': {'cursor': sensitive_path}, + 'settings': {'globs': globs if globs is not None else ['.env', 'secrets/**']}, + }, + {'key': 'secrets_in_mcp_args', 'event_type': 'McpExecution', 'agents': {'cursor': mcp}}, + ], + } + + +def platform_config(fetched_at: Optional[float] = None, **modes: str) -> GuardrailConfig: + """A cached platform config; keyword args are the per-guardrail modes (see resolved_guardrails_payload).""" + return GuardrailConfig( + payload=resolved_guardrails_payload(**modes), + fetched_at=time.time() if fetched_at is None else fetched_at, + ) diff --git a/tests/cli/commands/ai_guardrails/scan/test_guardrail_config.py b/tests/cli/commands/ai_guardrails/scan/test_guardrail_config.py new file mode 100644 index 00000000..7a3cbecd --- /dev/null +++ b/tests/cli/commands/ai_guardrails/scan/test_guardrail_config.py @@ -0,0 +1,152 @@ +"""Tests for the platform guardrail config cache.""" + +import time +from pathlib import Path + +import pytest +from pyfakefs.fake_filesystem import FakeFilesystem + +from cycode.cli.apps.ai_guardrails.scan.consts import DEFAULT_POLICY +from cycode.cli.apps.ai_guardrails.scan.guardrail_config import ( + GuardrailConfig, + apply_platform_config, + get_config_cache_path, + load_guardrail_config, + save_guardrail_config, +) +from tests.cli.commands.ai_guardrails.scan.conftest import platform_config as _config +from tests.cli.commands.ai_guardrails.scan.conftest import resolved_guardrails_payload as _payload + + +@pytest.fixture(autouse=True) +def _fake_home(fs: FakeFilesystem, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv('HOME', '/home/testuser') + fs.create_dir('/home/testuser') + + +# --- cache file --- + + +def test_save_and_load_round_trip() -> None: + save_guardrail_config(_payload(), 'tenant-a') + + config = load_guardrail_config() + + assert config is not None + assert config.is_expired() is False + assert config.tenant_id == 'tenant-a' + assert config.needs_refresh('tenant-a') is False + # Switching tenants invalidates the cache even while it is still fresh. + assert config.needs_refresh('tenant-b') is True + + +def test_load_missing_cache_returns_none() -> None: + assert load_guardrail_config() is None + + +def test_corrupt_cache_is_quarantined(fs: FakeFilesystem) -> None: + path = get_config_cache_path() + fs.create_file(str(path), contents='not json {') + + assert load_guardrail_config() is None + assert not path.exists() + assert Path(f'{path}.corrupt').exists() + + +def test_expired_cache_detected() -> None: + config = GuardrailConfig(payload=_payload(), fetched_at=time.time() - 10_000, tenant_id='tenant-a') + assert config.is_expired() is True + assert config.needs_refresh('tenant-a') is True + + +# --- lookups --- + + +def test_agent_mapping_claude_code_reads_claude_column() -> None: + config = _config() + # cursor column is Report; claude column is Block - the claude-code ide maps onto it. + assert config.can_event_block('Prompt', 'claude-code') is True + assert config.can_event_block('Prompt', 'cursor') is False + + +def test_event_off_requires_every_guardrail_off() -> None: + partially_off = _config(file_read='Off', sensitive_path='Report') + assert partially_off.is_event_off('FileRead', 'cursor') is False + + fully_off = _config(file_read='Off', sensitive_path='Off') + assert fully_off.is_event_off('FileRead', 'cursor') is True + + +def test_unknown_guardrail_keys_are_ignored() -> None: + payload = _payload() + # A future guardrail this CLI doesn't implement must never affect decisions (fail-open). + payload['guardrails'].append( + {'key': 'unauthorized_mcp_server', 'event_type': 'McpExecution', 'agents': {'cursor': 'Block'}} + ) + config = GuardrailConfig(payload=payload, fetched_at=time.time()) + + assert config.can_event_block('McpExecution', 'cursor') is False + + +def test_missing_agent_defaults_to_report() -> None: + config = _config() + assert config.can_event_block('Prompt', 'codex') is False + assert config.is_event_off('Prompt', 'codex') is False + + +# --- platform overlay --- + + +def test_apply_platform_config_without_cache_uses_report_defaults() -> None: + policy: dict = {'fail_open': True} + + apply_platform_config(policy, None, 'cursor') + + assert policy['prompt']['action'] == 'warn' + assert policy['file_read']['action'] == 'warn' + assert policy['file_read']['path_action'] == 'warn' + assert policy['mcp']['action'] == 'warn' + assert policy['file_read']['deny_globs'] == DEFAULT_POLICY['file_read']['deny_globs'] + + +def test_apply_platform_config_block_cell_sets_block_action() -> None: + policy: dict = {} + + apply_platform_config(policy, _config(prompt='Block'), 'cursor') + + assert policy['prompt']['action'] == 'block' + assert policy['mcp']['action'] == 'warn' + + +def test_apply_platform_config_file_read_cells_keep_independent_actions() -> None: + # Content scan and sensitive path share the FileRead event but are separate matrix cells. + policy: dict = {} + apply_platform_config(policy, _config(file_read='Block', sensitive_path='Report'), 'cursor') + assert policy['file_read']['action'] == 'block' + assert policy['file_read']['path_action'] == 'warn' + + policy = {} + apply_platform_config(policy, _config(file_read='Report', sensitive_path='Block'), 'cursor') + assert policy['file_read']['action'] == 'warn' + assert policy['file_read']['path_action'] == 'block' + + +def test_apply_platform_config_off_cells_disable_subfeatures() -> None: + policy: dict = {} + config = _config(file_read='Off', sensitive_path='Report') + + apply_platform_config(policy, config, 'cursor') + + # Content scan off, path matching still on with the platform globs. + assert policy['file_read']['scan_content'] is False + assert policy['file_read']['deny_globs'] == ['.env', 'secrets/**'] + + +def test_apply_platform_config_sensitive_path_off_clears_globs() -> None: + policy: dict = {} + config = _config(sensitive_path='Off') + + apply_platform_config(policy, config, 'cursor') + + assert policy['file_read']['deny_globs'] == [] + assert policy['file_read']['scan_content'] is True diff --git a/tests/cli/commands/ai_guardrails/scan/test_handlers.py b/tests/cli/commands/ai_guardrails/scan/test_handlers.py index 7c008bfc..81ef0d27 100644 --- a/tests/cli/commands/ai_guardrails/scan/test_handlers.py +++ b/tests/cli/commands/ai_guardrails/scan/test_handlers.py @@ -205,6 +205,17 @@ def test_handle_before_read_file_sensitive_path( assert call_args.kwargs['block_reason'] == BlockReason.SENSITIVE_PATH assert call_args.kwargs['file_path'] == '/path/to/.env' + # The path guardrail has its own action: content scan in block mode must not make a + # report-mode path match block. + mock_ctx.obj['ai_security_client'].create_event.reset_mock() + default_policy['file_read']['path_action'] = 'warn' + default_policy['file_read']['scan_content'] = False + + result = handle_before_read_file(mock_ctx, payload, default_policy) + + assert result.action == DecisionAction.ASK + assert mock_ctx.obj['ai_security_client'].create_event.call_args.args[2] == AIHookOutcome.WARNED + @patch('cycode.cli.apps.ai_guardrails.scan.handlers.is_denied_path') @patch('cycode.cli.apps.ai_guardrails.scan.handlers._scan_path_for_secrets') @@ -252,6 +263,16 @@ 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' + default_policy['file_read']['path_action'] = 'block' + + 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 mock_ctx.obj['ai_security_client'].create_event.call_args.args[2] == AIHookOutcome.WARNED + @patch('cycode.cli.apps.ai_guardrails.scan.handlers.is_denied_path') @patch('cycode.cli.apps.ai_guardrails.scan.handlers._scan_path_for_secrets') @@ -281,7 +302,7 @@ def test_handle_before_read_file_sensitive_path_warn_mode_scans_content( """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') - default_policy['mode'] = 'warn' + default_policy['file_read']['path_action'] = 'warn' payload = AIHookPayload( event_name='FileRead', ide_provider='cursor', @@ -312,7 +333,8 @@ def test_handle_before_read_file_sensitive_path_warn_mode_with_secrets( """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') - default_policy['mode'] = 'warn' + default_policy['file_read']['path_action'] = 'warn' + default_policy['file_read']['action'] = 'warn' payload = AIHookPayload( event_name='FileRead', ide_provider='cursor', @@ -342,7 +364,7 @@ def test_handle_before_read_file_sensitive_path_scan_disabled_warns( ) -> None: """Test that sensitive path in warn mode with scan disabled emits a single event.""" mock_is_denied.return_value = True - default_policy['mode'] = 'warn' + default_policy['file_read']['path_action'] = 'warn' default_policy['file_read']['scan_content'] = False payload = AIHookPayload( event_name='FileRead', @@ -528,12 +550,14 @@ def test_handle_before_mcp_execution_scan_disabled( 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 +def test_get_effective_mode_reads_the_guardrails_action() -> None: + assert get_effective_mode({'action': 'block'}) == GuardrailsMode.BLOCK + assert get_effective_mode({'action': 'warn'}) == GuardrailsMode.REPORT + assert get_effective_mode({}) == GuardrailsMode.BLOCK + # A feature may carry more than one action; the caller picks which cell to read. + feature = {'action': 'warn', 'path_action': 'block'} + assert get_effective_mode(feature) == GuardrailsMode.REPORT + assert get_effective_mode(feature, action_key='path_action') == GuardrailsMode.BLOCK @patch('cycode.cli.apps.ai_guardrails.scan.handlers.get_serial_number', return_value='SER-123') diff --git a/tests/cli/commands/ai_guardrails/scan/test_policy.py b/tests/cli/commands/ai_guardrails/scan/test_policy.py index a378ad1c..ee4f606a 100644 --- a/tests/cli/commands/ai_guardrails/scan/test_policy.py +++ b/tests/cli/commands/ai_guardrails/scan/test_policy.py @@ -7,6 +7,7 @@ import pytest from pyfakefs.fake_filesystem import FakeFilesystem +from cycode.cli.apps.ai_guardrails.scan.consts import DEFAULT_POLICY from cycode.cli.apps.ai_guardrails.scan.policy import ( deep_merge, get_machine_policy_path, @@ -79,7 +80,6 @@ def test_load_defaults() -> None: defaults = load_defaults() assert isinstance(defaults, dict) - assert 'mode' in defaults assert 'fail_open' in defaults assert 'prompt' in defaults assert 'file_read' in defaults @@ -132,8 +132,8 @@ def test_load_policy_defaults_only(mock_load: MagicMock) -> None: policy = load_policy() - assert 'mode' in policy assert 'fail_open' in policy + assert 'secrets' in policy @patch('pathlib.Path.home') @@ -146,8 +146,8 @@ def test_load_policy_with_user_config(mock_home: MagicMock, fs: FakeFilesystem) policy = load_policy() - # User config should override defaults - assert policy['mode'] == 'warn' + # Knobs merge; `mode` is platform-managed and stripped from local files. + assert 'mode' not in policy assert policy['fail_open'] is False @@ -159,16 +159,24 @@ def test_load_policy_with_repo_config(mock_load: MagicMock) -> None: def side_effect(path: Path) -> Optional[dict]: if path == repo_config: - return {'mode': 'block', 'prompt': {'enabled': False}} + return { + 'mode': 'block', + 'fail_open': False, + 'prompt': {'enabled': False}, + 'file_read': {'deny_globs': ['*.bak'], 'scan_content': False}, + } return None mock_load.side_effect = side_effect policy = load_policy(str(repo_path)) - # Repo config should have highest precedence - assert policy['mode'] == 'block' - assert policy['prompt']['enabled'] is False + # Knobs merge from the repo file; enforcement keys are platform-managed and stripped. + assert policy['fail_open'] is False + assert 'mode' not in policy + assert policy['prompt']['enabled'] is True + assert policy['file_read']['deny_globs'] == DEFAULT_POLICY['file_read']['deny_globs'] + assert policy['file_read']['scan_content'] is True @patch('pathlib.Path.home') @@ -177,15 +185,17 @@ def test_load_policy_precedence(mock_home: MagicMock, fs: FakeFilesystem) -> Non mock_home.return_value = Path('/home/testuser') # Create user config - fs.create_file('/home/testuser/.cycode/ai-guardrails.yaml', contents='mode: warn\nfail_open: false\n') + fs.create_file( + '/home/testuser/.cycode/ai-guardrails.yaml', contents='fail_open: false\nsecrets:\n max_bytes: 100\n' + ) # Create repo config - fs.create_file('/fake/repo/.cycode/ai-guardrails.yaml', contents='mode: block\n') + fs.create_file('/fake/repo/.cycode/ai-guardrails.yaml', contents='secrets:\n max_bytes: 200\n') policy = load_policy('/fake/repo') - # mode should come from repo (highest precedence) - assert policy['mode'] == 'block' + # max_bytes should come from repo (highest precedence) + assert policy['secrets']['max_bytes'] == 200 # fail_open should come from user config (repo doesn't override it) assert policy['fail_open'] is False @@ -198,7 +208,7 @@ def test_load_policy_none_workspace_root(mock_load: MagicMock) -> None: policy = load_policy(None) # Should only load defaults (no repo config) - assert 'mode' in policy + assert policy == DEFAULT_POLICY def test_get_machine_policy_path_per_os(monkeypatch: pytest.MonkeyPatch) -> None: @@ -223,13 +233,13 @@ def test_load_policy_with_machine_config( mock_home.return_value = Path('/home/testuser') machine_path = Path('/machine/ai-guardrails.yaml') mock_machine_path.return_value = machine_path - fs.create_file(str(machine_path), contents='mode: warn\n') + fs.create_file(str(machine_path), contents='mode: warn\nsecrets:\n timeout_ms: 5000\n') policy = load_policy() - # Machine config overrides the built-in default (block); other keys inherit from defaults. - assert policy['mode'] == 'warn' - assert policy['fail_open'] is True + # Machine knobs merge; `mode` is platform-managed and stripped from local files. + assert 'mode' not in policy + assert policy['secrets']['timeout_ms'] == 5000 @patch('pathlib.Path.home') @@ -241,12 +251,12 @@ def test_load_policy_precedence_defaults_machine_user_repo( mock_home.return_value = Path('/home/testuser') machine_path = Path('/machine/ai-guardrails.yaml') mock_machine_path.return_value = machine_path - fs.create_file(str(machine_path), contents='mode: warn\nfail_open: false\n') + fs.create_file(str(machine_path), contents='fail_open: false\nsecrets:\n timeout_ms: 1000\n') fs.create_file('/home/testuser/.cycode/ai-guardrails.yaml', contents='fail_open: true\n') - fs.create_file('/fake/repo/.cycode/ai-guardrails.yaml', contents='mode: block\n') + fs.create_file('/fake/repo/.cycode/ai-guardrails.yaml', contents='secrets:\n timeout_ms: 3000\n') policy = load_policy('/fake/repo') - # repo overrides machine's mode; user overrides machine's fail_open. - assert policy['mode'] == 'block' + # repo overrides machine's timeout; user overrides machine's fail_open. + assert policy['secrets']['timeout_ms'] == 3000 assert policy['fail_open'] is True diff --git a/tests/cli/commands/ai_guardrails/scan/test_scan_command.py b/tests/cli/commands/ai_guardrails/scan/test_scan_command.py index c7351b67..e60e6047 100644 --- a/tests/cli/commands/ai_guardrails/scan/test_scan_command.py +++ b/tests/cli/commands/ai_guardrails/scan/test_scan_command.py @@ -1,7 +1,9 @@ """Tests for AI guardrails scan command.""" import json +import time from io import StringIO +from typing import Optional from unittest.mock import MagicMock import pytest @@ -10,8 +12,16 @@ from cycode.cli.apps.ai_guardrails import app as ai_guardrails_app from cycode.cli.apps.ai_guardrails.ides.base import HookDecision +from cycode.cli.apps.ai_guardrails.scan.guardrail_config import GuardrailConfig from cycode.cli.apps.ai_guardrails.scan.scan_command import scan_command from cycode.cli.apps.ai_guardrails.scan.types import AiHookEventType +from tests.cli.commands.ai_guardrails.scan.conftest import platform_config + + +@pytest.fixture(autouse=True) +def no_guardrail_cache(mocker: MockerFixture) -> None: + """Keep tests hermetic: never read a real ~/.cycode cache. Tests re-patch with their own config.""" + mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.load_guardrail_config', return_value=None) @pytest.fixture @@ -266,14 +276,24 @@ def mock_respawn(self, mocker: MockerFixture) -> MagicMock: def not_detached(self, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv('_CYCODE_DETACHED', raising=False) - def _run_prompt_scan(self, mock_ctx: MagicMock, mocker: MockerFixture, policy: dict) -> str: + def _run_prompt_scan( + self, + mock_ctx: MagicMock, + mocker: MockerFixture, + config: Optional[GuardrailConfig], + policy: Optional[dict] = None, + ) -> str: payload = json.dumps({'hook_event_name': 'beforeSubmitPrompt', 'conversation_id': 'c-1', 'prompt': 'test'}) mocker.patch('sys.stdin', StringIO(payload)) - mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.load_policy', return_value=policy) + mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.load_guardrail_config', return_value=config) + mocker.patch( + 'cycode.cli.apps.ai_guardrails.scan.scan_command.load_policy', + return_value=policy if policy is not None else {'fail_open': True}, + ) scan_command(mock_ctx, ide='cursor') return payload - def test_warn_mode_detaches_before_any_client_work( + def test_report_mode_detaches_before_any_client_work( self, mock_ctx: MagicMock, mocker: MockerFixture, @@ -285,7 +305,7 @@ def test_warn_mode_detaches_before_any_client_work( handler = MagicMock() mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.get_handler_for_event', return_value=handler) - payload = self._run_prompt_scan(mock_ctx, mocker, {'mode': 'warn', 'fail_open': True}) + payload = self._run_prompt_scan(mock_ctx, mocker, platform_config(prompt='report')) # The parent hands the raw payload to the child and exits with no output. mock_respawn.assert_called_once_with(payload) @@ -293,18 +313,80 @@ def test_warn_mode_detaches_before_any_client_work( handler.assert_not_called() assert capsys.readouterr().out == '' - def test_block_mode_stays_synchronous( + def test_block_cell_stays_synchronous( + self, + mock_ctx: MagicMock, + mocker: MockerFixture, + mock_respawn: MagicMock, + not_detached: None, + ) -> None: + mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command._initialize_clients') + handler = MagicMock(return_value=HookDecision.allow(AiHookEventType.PROMPT)) + mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.get_handler_for_event', return_value=handler) + + self._run_prompt_scan(mock_ctx, mocker, platform_config(prompt='block')) + + mock_respawn.assert_not_called() + handler.assert_called_once() + + def test_cold_start_without_cache_stays_synchronous( + self, + mock_ctx: MagicMock, + mocker: MockerFixture, + mock_respawn: MagicMock, + not_detached: None, + ) -> None: + """No cache means built-in Report defaults, but never a detach on an assumption.""" + mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command._initialize_clients') + handler = MagicMock(return_value=HookDecision.allow(AiHookEventType.PROMPT)) + mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.get_handler_for_event', return_value=handler) + passed_policies = [] + handler.side_effect = lambda _ctx, _payload, policy: ( + passed_policies.append(policy), + HookDecision.allow(AiHookEventType.PROMPT), + )[1] + + self._run_prompt_scan(mock_ctx, mocker, config=None) + + mock_respawn.assert_not_called() + handler.assert_called_once() + # Cold-start defaults equal an unconfigured tenant's platform config: Report (warn). + assert passed_policies[0]['prompt']['action'] == 'warn' + + def test_off_cell_skips_scan_entirely( + self, + mock_ctx: MagicMock, + mocker: MockerFixture, + capsys: pytest.CaptureFixture[str], + mock_respawn: MagicMock, + not_detached: None, + ) -> None: + """Every guardrail for the event Off: allow immediately - no scan, no event, no auth.""" + initialize_clients = mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command._initialize_clients') + handler = MagicMock() + mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.get_handler_for_event', return_value=handler) + + self._run_prompt_scan(mock_ctx, mocker, platform_config(prompt='off')) + + mock_respawn.assert_not_called() + initialize_clients.assert_not_called() + handler.assert_not_called() + assert json.loads(capsys.readouterr().out).get('continue') is True + + def test_expired_cache_is_still_applied( self, mock_ctx: MagicMock, mocker: MockerFixture, mock_respawn: MagicMock, not_detached: None, ) -> None: + """Scans never fetch: a stale cache is used as-is; session-start refreshes it (TTL-gated).""" mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command._initialize_clients') handler = MagicMock(return_value=HookDecision.allow(AiHookEventType.PROMPT)) mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.get_handler_for_event', return_value=handler) - self._run_prompt_scan(mock_ctx, mocker, {'mode': 'block', 'fail_open': True}) + expired = platform_config(prompt='block', fetched_at=time.time() - 10_000) + self._run_prompt_scan(mock_ctx, mocker, expired) mock_respawn.assert_not_called() handler.assert_called_once() @@ -321,12 +403,12 @@ def test_detached_child_never_respawns_again( handler = MagicMock(return_value=HookDecision.allow(AiHookEventType.PROMPT)) mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.get_handler_for_event', return_value=handler) - self._run_prompt_scan(mock_ctx, mocker, {'mode': 'warn', 'fail_open': True}) + self._run_prompt_scan(mock_ctx, mocker, platform_config(prompt='report')) mock_respawn.assert_not_called() handler.assert_called_once() - def test_fail_closed_policy_stays_synchronous_even_in_warn_mode( + def test_fail_closed_policy_stays_synchronous_even_in_report_mode( self, mock_ctx: MagicMock, mocker: MockerFixture, @@ -338,7 +420,7 @@ def test_fail_closed_policy_stays_synchronous_even_in_warn_mode( handler = MagicMock(return_value=HookDecision.allow(AiHookEventType.PROMPT)) mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.get_handler_for_event', return_value=handler) - self._run_prompt_scan(mock_ctx, mocker, {'mode': 'warn', 'fail_open': False}) + self._run_prompt_scan(mock_ctx, mocker, platform_config(prompt='report'), policy={'fail_open': False}) mock_respawn.assert_not_called() handler.assert_called_once() @@ -351,11 +433,12 @@ def test_detach_decision_is_per_event( not_detached: None, ) -> None: """Prompt blocks while file-read reports: only the file-read event detaches.""" - policy = {'mode': 'block', 'fail_open': True, 'file_read': {'action': 'warn'}} + config = platform_config(prompt='block', file_read='report', sensitive_path='report') mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command._initialize_clients') handler = MagicMock(return_value=HookDecision.allow(AiHookEventType.PROMPT)) mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.get_handler_for_event', return_value=handler) - mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.load_policy', return_value=policy) + mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.load_guardrail_config', return_value=config) + mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.load_policy', return_value={'fail_open': True}) mocker.patch('sys.stdin', StringIO(json.dumps({'hook_event_name': 'beforeSubmitPrompt', 'prompt': 'x'}))) scan_command(mock_ctx, ide='cursor') @@ -378,7 +461,7 @@ def test_failed_respawn_falls_back_to_synchronous_scan( handler = MagicMock(return_value=HookDecision.allow(AiHookEventType.PROMPT)) mocker.patch('cycode.cli.apps.ai_guardrails.scan.scan_command.get_handler_for_event', return_value=handler) - self._run_prompt_scan(mock_ctx, mocker, {'mode': 'warn', 'fail_open': True}) + self._run_prompt_scan(mock_ctx, mocker, platform_config(prompt='report')) mock_respawn.assert_called_once() handler.assert_called_once() diff --git a/tests/cli/commands/ai_guardrails/test_hooks_manager.py b/tests/cli/commands/ai_guardrails/test_hooks_manager.py index 2212e798..2f0d2b72 100644 --- a/tests/cli/commands/ai_guardrails/test_hooks_manager.py +++ b/tests/cli/commands/ai_guardrails/test_hooks_manager.py @@ -13,7 +13,6 @@ from cycode.cli.apps.ai_guardrails.consts import ( CYCODE_SCAN_PROMPT_COMMAND, CYCODE_SESSION_START_COMMAND, - PolicyMode, ) from cycode.cli.apps.ai_guardrails.hooks_manager import ( create_policy_file, @@ -114,44 +113,48 @@ def test_claude_code_render_hooks_session_start() -> None: # Policy file tests -def test_create_policy_file_warn(fs: FakeFilesystem) -> None: - """Create a warn-mode policy file.""" +def test_create_policy_file_writes_knobs_only(fs: FakeFilesystem) -> None: + """The policy file carries operational knobs only; enforcement is platform-managed.""" fs.create_dir(Path.home()) - success, message = create_policy_file('user', PolicyMode.WARN) + success, _ = create_policy_file('user') assert success is True - assert 'warn mode' in message policy_path = Path.home() / '.cycode' / 'ai-guardrails.yaml' assert policy_path.exists() - assert yaml.safe_load(policy_path.read_text())['mode'] == 'warn' - - -def test_create_policy_file_block(fs: FakeFilesystem) -> None: - """Create a block-mode policy file.""" - fs.create_dir(Path.home()) - success, message = create_policy_file('user', PolicyMode.BLOCK) - - assert success is True - assert 'block mode' in message - - policy_path = Path.home() / '.cycode' / 'ai-guardrails.yaml' - assert yaml.safe_load(policy_path.read_text())['mode'] == 'block' + policy = yaml.safe_load(policy_path.read_text()) + assert 'mode' not in policy + assert policy['fail_open'] is True + assert policy['secrets']['timeout_ms'] > 0 -def test_create_policy_file_updates_existing(fs: FakeFilesystem) -> None: - """Re-running updates only the mode field and preserves customizations.""" +def test_create_policy_file_updates_existing_and_strips_platform_keys(fs: FakeFilesystem) -> None: + """Re-running preserves customizations but strips enforcement keys older CLIs wrote.""" policy_dir = Path.home() / '.cycode' fs.create_dir(policy_dir) policy_path = policy_dir / 'ai-guardrails.yaml' - policy_path.write_text(yaml.dump({'version': 1, 'mode': 'warn', 'custom_field': 'keep_me'})) + policy_path.write_text( + yaml.dump( + { + 'version': 1, + 'mode': 'warn', + 'custom_field': 'keep_me', + 'secrets': {'timeout_ms': 5000}, + 'file_read': {'deny_globs': ['*.secret'], 'scan_content': False}, + } + ) + ) - success, _ = create_policy_file('user', PolicyMode.BLOCK) + success, _ = create_policy_file('user') assert success is True policy = yaml.safe_load(policy_path.read_text()) - assert policy['mode'] == 'block' + assert 'mode' not in policy assert policy['custom_field'] == 'keep_me' + assert policy['secrets']['timeout_ms'] == 5000 + # Nothing platform-managed is left, and an emptied section is dropped rather than written as {}. + assert 'file_read' not in policy + assert 'prompt' not in policy def test_install_preserves_user_hook_colocated_with_cycode( @@ -316,9 +319,9 @@ def test_create_policy_file_repo_scope(fs: FakeFilesystem) -> None: repo_path = Path('/my-repo') fs.create_dir(repo_path) - success, _ = create_policy_file('repo', PolicyMode.WARN, repo_path=repo_path) + success, _ = create_policy_file('repo', repo_path=repo_path) assert success is True policy_path = repo_path / '.cycode' / 'ai-guardrails.yaml' assert policy_path.exists() - assert yaml.safe_load(policy_path.read_text())['mode'] == 'warn' + assert 'mode' not in yaml.safe_load(policy_path.read_text()) diff --git a/tests/cli/commands/ai_guardrails/test_session_start_command.py b/tests/cli/commands/ai_guardrails/test_session_start_command.py index eaec8531..85c5116a 100644 --- a/tests/cli/commands/ai_guardrails/test_session_start_command.py +++ b/tests/cli/commands/ai_guardrails/test_session_start_command.py @@ -1,6 +1,7 @@ """Tests for session-start command.""" import json +import time from io import StringIO from pathlib import Path from unittest.mock import ANY, MagicMock, patch @@ -14,6 +15,7 @@ from cycode.cli.apps.ai_guardrails.ides import codex as _codex_mod from cycode.cli.apps.ai_guardrails.ides import copilot as _copilot_mod from cycode.cli.apps.ai_guardrails.ides import cursor as _cursor_mod +from cycode.cli.apps.ai_guardrails.scan.guardrail_config import GuardrailConfig from cycode.cli.apps.ai_guardrails.session_start_command import session_start_command @@ -25,6 +27,15 @@ def mock_ctx() -> MagicMock: return ctx +@pytest.fixture(autouse=True) +def mock_save_guardrail_config(monkeypatch: pytest.MonkeyPatch) -> MagicMock: + """Keep tests hermetic: never read or write the real guardrail config cache.""" + save_mock = MagicMock(return_value=True) + monkeypatch.setattr(_session_start_mod, 'save_guardrail_config', save_mock) + monkeypatch.setattr(_session_start_mod, 'load_guardrail_config', MagicMock(return_value=None)) + return save_mock + + @pytest.fixture(autouse=True) def _isolated_session_context_cache(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: """Keep the dedup cache away from the real ~/.cycode in every test.""" @@ -590,3 +601,75 @@ def test_unauthenticated_skips_session_init( session_start_command(mock_ctx, ide='claude-code') mock_get_client.assert_not_called() + + +@patch.object(_session_start_mod, 'get_ai_security_manager_client') +@patch.object(_session_start_mod, 'get_authorization_info') +def test_session_start_fetches_guardrail_config( + mock_get_auth: MagicMock, + mock_get_client: MagicMock, + mock_ctx: MagicMock, + mock_save_guardrail_config: MagicMock, +) -> None: + """session-start fetches the platform guardrail config into the local cache, keyed by tenant.""" + mock_get_auth.return_value = MagicMock(tenant_id='tenant-a') + mock_ai_client = MagicMock() + mock_ai_client.get_resolved_guardrails.return_value = {'ttl_seconds': 900, 'guardrails': []} + mock_get_client.return_value = mock_ai_client + + with patch('sys.stdin', new=StringIO(json.dumps({'conversation_id': 'conv-1'}))): + session_start_command(mock_ctx, ide='cursor') + + mock_save_guardrail_config.assert_called_once_with({'ttl_seconds': 900, 'guardrails': []}, 'tenant-a') + + +@patch.object(_session_start_mod, 'get_ai_security_manager_client') +@patch.object(_session_start_mod, 'get_authorization_info') +def test_guardrail_config_fetch_failure_is_non_fatal( + mock_get_auth: MagicMock, + mock_get_client: MagicMock, + mock_ctx: MagicMock, + mock_save_guardrail_config: MagicMock, +) -> None: + """The client reports a failed fetch as None (it never raises); nothing is written over the cache.""" + mock_get_auth.return_value = MagicMock() + mock_ai_client = MagicMock() + mock_ai_client.get_resolved_guardrails.return_value = None + mock_get_client.return_value = mock_ai_client + + with patch('sys.stdin', new=StringIO(json.dumps({'conversation_id': 'conv-1'}))): + session_start_command(mock_ctx, ide='cursor') + + mock_save_guardrail_config.assert_not_called() + + +@patch.object(_session_start_mod, 'get_ai_security_manager_client') +@patch.object(_session_start_mod, 'get_authorization_info') +def test_fresh_guardrail_config_cache_skips_fetch( + mock_get_auth: MagicMock, + mock_get_client: MagicMock, + mock_ctx: MagicMock, + mock_save_guardrail_config: MagicMock, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The fetch is gated on the cache: fresh and same-tenant skips the network call entirely.""" + mock_get_auth.return_value = MagicMock(tenant_id='tenant-a') + mock_ai_client = MagicMock() + mock_get_client.return_value = mock_ai_client + cached = GuardrailConfig(payload={'ttl_seconds': 900}, fetched_at=time.time(), tenant_id='tenant-a') + monkeypatch.setattr(_session_start_mod, 'load_guardrail_config', MagicMock(return_value=cached)) + + with patch('sys.stdin', new=StringIO(json.dumps({'conversation_id': 'conv-1'}))): + session_start_command(mock_ctx, ide='cursor') + + mock_ai_client.get_resolved_guardrails.assert_not_called() + mock_save_guardrail_config.assert_not_called() + + # Same fresh cache, but the user switched tenants: it must be refetched. + mock_get_auth.return_value = MagicMock(tenant_id='tenant-b') + + with patch('sys.stdin', new=StringIO(json.dumps({'conversation_id': 'conv-2'}))): + session_start_command(mock_ctx, ide='cursor') + + mock_ai_client.get_resolved_guardrails.assert_called_once() + mock_save_guardrail_config.assert_called_once_with(mock_ai_client.get_resolved_guardrails.return_value, 'tenant-b')