Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 29 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,9 @@ This guide walks you through both installation and usage.
1. [Discovering Commands](#discovering-commands)
2. [Examples](#platform-examples)
3. [Notes & Limitations](#platform-notes--limitations)
6. [Scan Command](#scan-command)
6. [AI Guardrails](#ai-guardrails-beta)
1. [Data Collected by AI Guardrails](#data-collected-by-ai-guardrails)
7. [Scan Command](#scan-command)
1. [Running a Scan](#running-a-scan)
1. [Options](#options)
1. [Severity Threshold](#severity-option)
Expand Down Expand Up @@ -704,6 +706,32 @@ cycode platform projects list --page-size 100 | jq '.items[].name'
- **Override the cache TTL** with `CYCODE_SPEC_CACHE_TTL=<seconds>`.


# AI Guardrails \[BETA\]

AI Guardrails installs hooks into supported AI coding agents (Claude Code, Cursor, Copilot, Codex) so that
prompts, files the agent reads, and MCP tool arguments are scanned for secrets before they reach the model.

## Data Collected by AI Guardrails

Scanning happens server-side, so the scanned content leaves the machine: the prompt text, the contents of
files the agent reads, and MCP tool arguments are sent to your Cycode tenant to be checked for secrets.

Each event is also reported with context about the developer and the machine, so a finding can be attributed
to the device and user it came from. Some of this is personal data:

- **Device identifiers** — the machine's hostname and hardware serial number.
- **User identifiers** — the email address of the user signed in to the AI coding agent, and the local
operating-system username.
- **Environment details** — operating system and version, the AI agent, its version and the model in use,
the contents of the agent's MCP configuration files, and its enabled plugins.

The hardware serial number is cached in a local temporary file, readable only by the user who ran the
command, so repeated hook invocations don't re-query the hardware.

If collecting this data is not acceptable in your environment, do not install the guardrails hooks
(`cycode ai-guardrails uninstall` removes hooks that are already installed).


# Scan Command

## Running a Scan
Expand Down
9 changes: 7 additions & 2 deletions cycode/cli/apps/ai_guardrails/consts.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,13 @@ class PolicyMode(str, Enum):
WARN = 'warn'


class InstallMode(str, Enum):
"""Installation mode for ai-guardrails install command."""
class GuardrailsMode(str, Enum):
"""Guardrails enforcement mode.

Used both as the ai-guardrails install-command mode and as the per-event
effective mode reported to the server (the ai_guardrails scan parameter's
`mode` field)
"""

REPORT = 'report'
BLOCK = 'block'
Expand Down
4 changes: 1 addition & 3 deletions cycode/cli/apps/ai_guardrails/ides/copilot.py
Original file line number Diff line number Diff line change
Expand Up @@ -333,9 +333,7 @@ def entry(command: str) -> dict:
return {
'version': 1,
'hooks': {
'sessionStart': [
{'type': 'command', 'command': _SESSION_START_COMMAND, 'timeoutSec': _HOOK_TIMEOUT_SEC}
],
'sessionStart': [{'type': 'command', 'command': _SESSION_START_COMMAND}],
Comment thread
omer-roth marked this conversation as resolved.
'userPromptSubmitted': [entry(_SCAN_PROMPT_COMMAND)],
'preToolUse': [entry(_SCAN_TOOL_COMMAND)],
},
Expand Down
14 changes: 7 additions & 7 deletions cycode/cli/apps/ai_guardrails/install_command.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
import typer

from cycode.cli.apps.ai_guardrails.command_utils import console, resolve_repo_path, validate_scope
from cycode.cli.apps.ai_guardrails.consts import InstallMode, PolicyMode
from cycode.cli.apps.ai_guardrails.consts import GuardrailsMode, PolicyMode
from cycode.cli.apps.ai_guardrails.hooks_manager import create_policy_file, install_hooks
from cycode.cli.apps.ai_guardrails.ides import DEFAULT_IDE_NAME, IDES, resolve_ides

Expand Down Expand Up @@ -40,14 +40,14 @@ def install_command(
),
] = None,
mode: Annotated[
InstallMode,
GuardrailsMode,
typer.Option(
'--mode',
'-m',
help='Installation mode: "report" for async non-blocking hooks with warn policy, '
'"block" for sync blocking hooks.',
),
] = InstallMode.REPORT,
] = GuardrailsMode.REPORT,
) -> None:
"""Install AI guardrails hooks for supported IDEs.

Expand All @@ -65,7 +65,7 @@ def install_command(
repo_path = resolve_repo_path(scope, repo_path)
ides_to_install = resolve_ides(ide)

report_mode = mode == InstallMode.REPORT
report_mode = mode == GuardrailsMode.REPORT

results: list[tuple[str, bool, str]] = []
for current_ide in ides_to_install:
Expand All @@ -83,7 +83,7 @@ def install_command(
all_success = False

if any_success:
policy_mode = PolicyMode.WARN if mode == InstallMode.REPORT else PolicyMode.BLOCK
policy_mode = PolicyMode.WARN if mode == GuardrailsMode.REPORT else PolicyMode.BLOCK
_install_policy(scope, repo_path, policy_mode)
_print_next_steps(results, mode)

Expand All @@ -99,15 +99,15 @@ def _install_policy(scope: str, repo_path: Optional[Path], policy_mode: PolicyMo
console.print(f'[red]✗[/] {policy_message}', style='bold red')


def _print_next_steps(results: list[tuple[str, bool, str]], mode: InstallMode) -> None:
def _print_next_steps(results: list[tuple[str, bool, str]], mode: GuardrailsMode) -> None:
console.print()
console.print('[bold]Next steps:[/]')
successful_ides = [name for name, success, _ in results if success]
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()
if mode == InstallMode.REPORT:
if mode == GuardrailsMode.REPORT:
console.print('[dim]Report mode: hooks run async (non-blocking) and policy is set to warn.[/]')
else:
console.print('[dim]The hooks will scan prompts, file reads, and MCP tool calls for secrets.[/]')
113 changes: 88 additions & 25 deletions cycode/cli/apps/ai_guardrails/scan/handlers.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,24 +17,29 @@

import typer

from cycode.cli.apps.ai_guardrails.consts import PolicyMode
from cycode.cli.apps.ai_guardrails.consts import GuardrailsMode, PolicyMode
from cycode.cli.apps.ai_guardrails.ides.base import HookDecision
from cycode.cli.apps.ai_guardrails.scan.payload import AIHookPayload
from cycode.cli.apps.ai_guardrails.scan.policy import get_policy_value
from cycode.cli.apps.ai_guardrails.scan.types import AiHookEventType, AIHookOutcome, BlockReason
from cycode.cli.apps.ai_guardrails.scan.types import (
SECRETS_BLOCK_REASON_BY_EVENT_TYPE,
AiHookEventType,
AIHookOutcome,
BlockReason,
)
from cycode.cli.apps.ai_guardrails.scan.utils import is_denied_path, truncate_utf8
from cycode.cli.apps.scan.code_scanner import _get_scan_documents_thread_func
from cycode.cli.apps.scan.scan_parameters import get_scan_parameters
from cycode.cli.cli_types import ScanTypeOption, SeverityOption
from cycode.cli.files_collector.file_excluder import is_path_configured_in_exclusions
from cycode.cli.models import Document
from cycode.cli.utils.host_info import get_hostname, get_serial_number
from cycode.cli.utils.progress_bar import DummyProgressBar, ScanProgressBarSection
from cycode.cli.utils.scan_utils import build_violation_summary
from cycode.logger import get_logger

logger = get_logger('AI Guardrails')


HandlerFn = Callable[[typer.Context, AIHookPayload, dict], HookDecision]


Expand All @@ -47,7 +52,7 @@ def handle_before_submit_prompt(ctx: typer.Context, payload: AIHookPayload, poli
ai_client.create_event(payload, AiHookEventType.PROMPT, AIHookOutcome.ALLOWED)
return HookDecision.allow(AiHookEventType.PROMPT)

mode = get_policy_value(policy, 'mode', default=PolicyMode.BLOCK)
effective_mode = get_effective_mode(policy, prompt_config)
prompt = payload.prompt or ''
max_bytes = get_policy_value(policy, 'secrets', 'max_bytes', default=200000)
timeout_ms = get_policy_value(policy, 'secrets', 'timeout_ms', default=30000)
Expand All @@ -59,12 +64,18 @@ def handle_before_submit_prompt(ctx: typer.Context, payload: AIHookPayload, poli
error_message = None

try:
violation_summary, scan_id = _scan_text_for_secrets(ctx, clipped, timeout_ms)
violation_summary, scan_id = _scan_text_for_secrets(
ctx,
clipped,
timeout_ms,
payload=payload,
event_type=AiHookEventType.PROMPT,
effective_mode=effective_mode,
)

if violation_summary:
block_reason = BlockReason.SECRETS_IN_PROMPT
action = get_policy_value(prompt_config, 'action', default=PolicyMode.BLOCK)
if action == PolicyMode.BLOCK and mode == PolicyMode.BLOCK:
block_reason = SECRETS_BLOCK_REASON_BY_EVENT_TYPE[AiHookEventType.PROMPT]
if effective_mode == GuardrailsMode.BLOCK:
outcome = AIHookOutcome.BLOCKED
user_message = f'{violation_summary}. Remove secrets before sending.'
return HookDecision.deny(AiHookEventType.PROMPT, user_message)
Expand Down Expand Up @@ -97,9 +108,8 @@ def handle_before_read_file(ctx: typer.Context, payload: AIHookPayload, policy:
ai_client.create_event(payload, AiHookEventType.FILE_READ, AIHookOutcome.ALLOWED)
return HookDecision.allow(AiHookEventType.FILE_READ)

mode = get_policy_value(policy, 'mode', default=PolicyMode.BLOCK)
file_path = payload.file_path or ''
action = get_policy_value(file_read_config, 'action', default=PolicyMode.BLOCK)
effective_mode = get_effective_mode(policy, file_read_config)

scan_id = None
block_reason = None
Expand All @@ -110,7 +120,7 @@ def handle_before_read_file(ctx: typer.Context, payload: AIHookPayload, policy:
is_sensitive_path = is_denied_path(file_path, policy)
if is_sensitive_path:
block_reason = BlockReason.SENSITIVE_PATH
if mode == PolicyMode.BLOCK and action == PolicyMode.BLOCK:
if effective_mode == GuardrailsMode.BLOCK:
outcome = AIHookOutcome.BLOCKED
user_message = f'Cycode blocked sending {file_path} to the AI (sensitive path policy).'
return HookDecision.deny(
Expand All @@ -133,10 +143,12 @@ def handle_before_read_file(ctx: typer.Context, payload: AIHookPayload, policy:
outcome = AIHookOutcome.ALLOWED

if get_policy_value(file_read_config, 'scan_content', default=True):
violation_summary, scan_id = _scan_path_for_secrets(ctx, file_path, policy)
violation_summary, scan_id = _scan_path_for_secrets(
ctx, file_path, policy, payload=payload, effective_mode=effective_mode
)
if violation_summary:
block_reason = BlockReason.SECRETS_IN_FILE
if mode == PolicyMode.BLOCK and action == PolicyMode.BLOCK:
block_reason = SECRETS_BLOCK_REASON_BY_EVENT_TYPE[AiHookEventType.FILE_READ]
if effective_mode == GuardrailsMode.BLOCK:
outcome = AIHookOutcome.BLOCKED
user_message = f'Cycode blocked reading {file_path}. {violation_summary}'
return HookDecision.deny(
Expand Down Expand Up @@ -191,7 +203,6 @@ class _ArgScanFeature:
policy_key: str # 'mcp' or 'command_exec'
scan_key: str # 'scan_arguments' or 'scan_command'
event_type: AiHookEventType
block_reason: BlockReason
deny_message: Callable[[str], str]
deny_agent_message: str
ask_message: Callable[[str], str]
Expand All @@ -213,11 +224,10 @@ def _handle_arg_scan(
ai_client.create_event(payload, feature.event_type, AIHookOutcome.ALLOWED)
return HookDecision.allow(feature.event_type)

mode = get_policy_value(policy, 'mode', default=PolicyMode.BLOCK)
max_bytes = get_policy_value(policy, 'secrets', 'max_bytes', default=200000)
timeout_ms = get_policy_value(policy, 'secrets', 'timeout_ms', default=30000)
clipped = truncate_utf8(scan_text, max_bytes)
action = get_policy_value(feature_config, 'action', default=PolicyMode.BLOCK)
effective_mode = get_effective_mode(policy, feature_config)

scan_id = None
block_reason = None
Expand All @@ -226,10 +236,17 @@ def _handle_arg_scan(

try:
if get_policy_value(feature_config, feature.scan_key, default=True):
violation_summary, scan_id = _scan_text_for_secrets(ctx, clipped, timeout_ms)
violation_summary, scan_id = _scan_text_for_secrets(
ctx,
clipped,
timeout_ms,
payload=payload,
event_type=feature.event_type,
effective_mode=effective_mode,
)
if violation_summary:
block_reason = feature.block_reason
if mode == PolicyMode.BLOCK and action == PolicyMode.BLOCK:
block_reason = SECRETS_BLOCK_REASON_BY_EVENT_TYPE[feature.event_type]
if effective_mode == GuardrailsMode.BLOCK:
outcome = AIHookOutcome.BLOCKED
return HookDecision.deny(
feature.event_type,
Expand Down Expand Up @@ -275,7 +292,6 @@ def handle_before_mcp_execution(ctx: typer.Context, payload: AIHookPayload, poli
policy_key='mcp',
scan_key='scan_arguments',
event_type=AiHookEventType.MCP_EXECUTION,
block_reason=BlockReason.SECRETS_IN_MCP_ARGS,
deny_message=lambda v: f'Cycode blocked MCP tool call "{tool}". {v}',
deny_agent_message='Do not pass secrets to tools. Use secret references (name/id) instead.',
ask_message=lambda v: f'{v} in MCP tool call "{tool}". Allow execution?',
Expand All @@ -295,6 +311,36 @@ def get_handler_for_event(event_type: str) -> Optional[HandlerFn]:
return handlers.get(event_type)


def get_effective_mode(policy: dict, feature_config: dict) -> GuardrailsMode:
"""The event only blocks when both the global mode and the per-guardrail action are block."""
mode = get_policy_value(policy, 'mode', default=PolicyMode.BLOCK)
action = get_policy_value(feature_config, 'action', default=PolicyMode.BLOCK)
return GuardrailsMode.BLOCK if (mode == PolicyMode.BLOCK and action == PolicyMode.BLOCK) else GuardrailsMode.REPORT


def build_ai_guardrails_scan_parameters(
Comment thread
omer-roth marked this conversation as resolved.
ctx: typer.Context,
paths: Optional[tuple[str, ...]],
payload: AIHookPayload,
event_type: AiHookEventType,
effective_mode: GuardrailsMode,
) -> dict:
scan_parameters = get_scan_parameters(ctx, paths)
scan_parameters.setdefault('metadata', {})['ai_guardrails'] = {
'mode': effective_mode.value,
'ide_provider': payload.ide_provider,
'detection_source': SECRETS_BLOCK_REASON_BY_EVENT_TYPE[event_type].value,
'device_id': get_serial_number(),
'device_hostname': get_hostname(),
'conversation_id': payload.conversation_id,
'generation_id': payload.generation_id,
'ide_user_email': payload.ide_user_email,
Comment thread
omer-roth marked this conversation as resolved.
'mcp_server_name': payload.mcp_server_name,
'mcp_tool_name': payload.mcp_tool_name,
}
return scan_parameters


def _setup_scan_context(ctx: typer.Context) -> typer.Context:
"""Set up minimal context for scan_documents without progress bars or printing."""
ctx.obj['progress_bar'] = DummyProgressBar([ScanProgressBarSection])
Expand Down Expand Up @@ -345,18 +391,32 @@ def _perform_scan(
return None, scan_id


def _scan_text_for_secrets(ctx: typer.Context, text: str, timeout_ms: int) -> tuple[Optional[str], Optional[str]]:
def _scan_text_for_secrets(
ctx: typer.Context,
text: str,
timeout_ms: int,
payload: AIHookPayload,
event_type: AiHookEventType,
effective_mode: GuardrailsMode,
) -> tuple[Optional[str], Optional[str]]:
"""Scan text content for secrets using Cycode CLI."""
if not text:
return None, None

document = Document(path='prompt-content.txt', content=text, is_git_diff_format=False)
scan_ctx = _setup_scan_context(ctx)
timeout_seconds = timeout_ms / 1000.0
return _perform_scan(scan_ctx, [document], get_scan_parameters(scan_ctx, None), timeout_seconds)
scan_parameters = build_ai_guardrails_scan_parameters(scan_ctx, None, payload, event_type, effective_mode)
return _perform_scan(scan_ctx, [document], scan_parameters, timeout_seconds)


def _scan_path_for_secrets(ctx: typer.Context, file_path: str, policy: dict) -> tuple[Optional[str], Optional[str]]:
def _scan_path_for_secrets(
ctx: typer.Context,
file_path: str,
policy: dict,
payload: AIHookPayload,
effective_mode: GuardrailsMode,
) -> tuple[Optional[str], Optional[str]]:
"""Scan a file path for secrets."""
if not file_path or not os.path.isfile(file_path):
return None, None
Expand All @@ -375,4 +435,7 @@ def _scan_path_for_secrets(ctx: typer.Context, file_path: str, policy: dict) ->

document = Document(path=os.path.basename(file_path), content=content, is_git_diff_format=False)
scan_ctx = _setup_scan_context(ctx)
return _perform_scan(scan_ctx, [document], get_scan_parameters(scan_ctx, (file_path,)), timeout_seconds)
scan_parameters = build_ai_guardrails_scan_parameters(
scan_ctx, (file_path,), payload, AiHookEventType.FILE_READ, effective_mode
)
return _perform_scan(scan_ctx, [document], scan_parameters, timeout_seconds)
4 changes: 4 additions & 0 deletions cycode/cli/apps/ai_guardrails/scan/scan_command.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
"""

from typing import Annotated, Optional, Union
from uuid import uuid4

import click
import typer
Expand Down Expand Up @@ -125,6 +126,9 @@ def scan_command(
return

unified_payload = ide_integration.parse_hook_payload(payload)
if not unified_payload.generation_id:
# Not every IDE dialect provides a generation id (e.g. Copilot)
unified_payload.generation_id = str(uuid4())
event_name = unified_payload.event_name
logger.debug(
'Processing AI guardrails hook',
Expand Down
9 changes: 9 additions & 0 deletions cycode/cli/apps/ai_guardrails/scan/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,3 +41,12 @@ class BlockReason(StrEnum):
SECRETS_IN_MCP_ARGS = 'secrets_in_mcp_args'
SENSITIVE_PATH = 'sensitive_path'
SCAN_FAILURE = 'scan_failure'


# The reason each event type yields when a secret is found in it. Also travels with the scan as
# `detection_source`, so the violation and the hook event are labelled from the same vocabulary.
SECRETS_BLOCK_REASON_BY_EVENT_TYPE: dict[AiHookEventType, BlockReason] = {
AiHookEventType.PROMPT: BlockReason.SECRETS_IN_PROMPT,
AiHookEventType.FILE_READ: BlockReason.SECRETS_IN_FILE,
AiHookEventType.MCP_EXECUTION: BlockReason.SECRETS_IN_MCP_ARGS,
}
Loading
Loading