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
10 changes: 10 additions & 0 deletions cycode/cli/apps/ai_guardrails/ides/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,16 @@ class HookDecision:
user_message: str | None = None
agent_message: str | None = None

@property
def model_message(self) -> str:
"""What a tool deny tells the model: the findings the user sees, then the agent's instructions."""
return '\n'.join(message for message in (self.user_message, self.agent_message) if message)

@property
def permission_reason(self) -> str:
"""A PreToolUse permissionDecisionReason: a deny reason goes to the model, an ask reason to the user."""
return self.model_message if self.action == DecisionAction.DENY else self.user_message or ''

@classmethod
def allow(cls, event_type: AiHookEventType) -> 'HookDecision':
return cls(action=DecisionAction.ALLOW, event_type=event_type)
Expand Down
11 changes: 8 additions & 3 deletions cycode/cli/apps/ai_guardrails/ides/claude_code.py
Original file line number Diff line number Diff line change
Expand Up @@ -397,8 +397,13 @@ def build_hook_response(self, decision: HookDecision) -> dict:
if decision.event_type == AiHookEventType.PROMPT:
if decision.action == DecisionAction.ALLOW:
return {}
# Both DENY and (unexpected) ASK on prompts collapse to a block.
return {'decision': 'block', 'reason': decision.user_message or ''}
# Both DENY and (unexpected) ASK on prompts collapse to a block. Claude Code echoes the blocked
# prompt (secret included) under the reason unless suppressed; the reason already locates the finding.
return {
'decision': 'block',
'reason': decision.user_message or '',
'hookSpecificOutput': {'hookEventName': 'UserPromptSubmit', 'suppressOriginalPrompt': True},
}

# FILE_READ / MCP_EXECUTION → hookSpecificOutput shape.
if decision.action == DecisionAction.ALLOW:
Expand All @@ -412,7 +417,7 @@ def build_hook_response(self, decision: HookDecision) -> dict:
'hookSpecificOutput': {
'hookEventName': 'PreToolUse',
'permissionDecision': decision.action.value, # 'deny' or 'ask'
'permissionDecisionReason': decision.user_message or '',
'permissionDecisionReason': decision.permission_reason,
}
}

Expand Down
2 changes: 1 addition & 1 deletion cycode/cli/apps/ai_guardrails/ides/codex.py
Original file line number Diff line number Diff line change
Expand Up @@ -319,7 +319,7 @@ def build_hook_response(self, decision: HookDecision) -> dict:
'hookSpecificOutput': {
'hookEventName': 'PreToolUse',
'permissionDecision': decision.action.value, # 'deny' or 'ask'
'permissionDecisionReason': decision.user_message or '',
'permissionDecisionReason': decision.permission_reason,
}
}

Expand Down
3 changes: 2 additions & 1 deletion cycode/cli/apps/ai_guardrails/ides/copilot.py
Original file line number Diff line number Diff line change
Expand Up @@ -474,7 +474,8 @@ def build_hook_response(self, decision: HookDecision) -> dict:
'hookSpecificOutput': {
'hookEventName': 'PreToolUse',
'permissionDecision': decision.action.value, # 'deny' or 'ask'
'permissionDecisionReason': decision.user_message or '',
# Copilot VS Code shows a deny reason to the user rather than the model, which is harmless
'permissionDecisionReason': decision.permission_reason,
}
}

Expand Down
7 changes: 5 additions & 2 deletions cycode/cli/apps/ai_guardrails/ides/cursor.py
Original file line number Diff line number Diff line change
Expand Up @@ -149,10 +149,13 @@ def build_hook_response(self, decision: HookDecision) -> dict:
# FILE_READ / MCP_EXECUTION → permission shape
if decision.action == DecisionAction.ALLOW:
return {'permission': 'allow'}
# beforeReadFile has no agent_message; its deny user_message is what reaches the model
file_read_deny = decision.event_type == AiHookEventType.FILE_READ and decision.action == DecisionAction.DENY
return {
'permission': decision.action.value, # 'deny' or 'ask'
'user_message': decision.user_message or '',
'agent_message': decision.agent_message or '',
'user_message': decision.model_message if file_read_deny else decision.user_message or '',
# The agent does not see user_message, and needs the findings (their sha) to act on its instructions
'agent_message': decision.model_message,
}

def build_session_payload(self, raw_payload: dict) -> AIHookPayload:
Expand Down
72 changes: 62 additions & 10 deletions cycode/cli/apps/ai_guardrails/scan/handlers.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,9 +10,11 @@

import json
import os
import sys
import threading
from collections.abc import Callable
from dataclasses import dataclass
from gettext import ngettext
from typing import TYPE_CHECKING, Any, NamedTuple

import typer
Expand Down Expand Up @@ -46,12 +48,53 @@
HandlerFn = Callable[[typer.Context, AIHookPayload, dict], HookDecision]


def _ignore_commands(shas: tuple[str, ...]) -> list[str]:
"""Ready-to-run ignore commands, one per finding, in the order the findings are listed.

Names the binary running this hook, which may not be on PATH (e.g. MDM installs). Repo scope: it lands in
the cwd's .cycode/config.yaml, which matches where the hook reads it as long as the agent runs it.
"""
executable = sys.executable if getattr(sys, 'frozen', False) else sys.argv[0]
if ' ' in executable:
executable = f'"{executable}"'
return [f'`{executable} ignore --by-sha {sha}`' for sha in shas]


def _with_ignore_commands(shas: tuple[str, ...], one: str, many: str) -> str:
"""`one` formatted with the single command, or `many` followed by the commands as a list; '' without shas."""
commands = _ignore_commands(shas)
if len(commands) == 1:
return one.format(command=commands[0])
if commands:
return many + ''.join(f'\n {command}' for command in commands)
return ''


def _false_positive_user_hint(shas: tuple[str, ...]) -> str:
# The agent never saw the blocked prompt, so the command has to be complete for the developer to pass on
return _with_ignore_commands(
shas,
one='Not a real secret? Ask your code agent to run {command}, then resend.',
many='Not real secrets? Ask your code agent to run the command for each one, then resend:',
)


def _false_positive_agent_hint(shas: tuple[str, ...]) -> str:
# Ignoring is the developer's call; the agent may run it for them, but only once they confirm
return 'Tell the user what was blocked.' + _with_ignore_commands(
shas,
one=' Only if the user confirms it is a false positive, run {command} and retry.',
many=' Only for a finding the user confirms is a false positive, run its command and retry:',
)
Comment thread
omer-roth marked this conversation as resolved.


class ScanOutcome(NamedTuple):
"""What one guardrail scan came back with; the verdict is the server's, which applied the tenant's floors."""

violation_summary: str | None = None
scan_id: str | None = None
verdict: GuardrailsMode | None = None
shas: tuple[str, ...] = ()


NO_SCAN = ScanOutcome()
Expand Down Expand Up @@ -96,7 +139,10 @@ def handle_before_submit_prompt(ctx: typer.Context, payload: AIHookPayload, poli
block_reason = SECRETS_BLOCK_REASON_BY_EVENT_TYPE[AiHookEventType.PROMPT]
if scan_outcome.verdict == GuardrailsMode.BLOCK:
outcome = AIHookOutcome.BLOCKED
user_message = f'Remove secrets before sending. {scan_outcome.violation_summary}'
secrets = ngettext('secret', 'secrets', len(scan_outcome.shas))
user_message = f'{scan_outcome.violation_summary}\nRemove the {secrets} and resend.'
if scan_outcome.shas:
user_message += f'\n{_false_positive_user_hint(scan_outcome.shas)}'
return HookDecision.deny(AiHookEventType.PROMPT, user_message)
outcome = AIHookOutcome.WARNED
return HookDecision.allow(AiHookEventType.PROMPT)
Expand Down Expand Up @@ -164,14 +210,15 @@ def handle_before_read_file(ctx: typer.Context, payload: AIHookPayload, policy:
block_reason = SECRETS_BLOCK_REASON_BY_EVENT_TYPE[AiHookEventType.FILE_READ]
if scan_outcome.verdict == GuardrailsMode.BLOCK:
outcome = AIHookOutcome.BLOCKED
user_message = f'Cycode blocked reading {file_path}. {scan_outcome.violation_summary}'
user_message = (
f'Cycode blocked reading {file_path} because it contains a secret.\n'
f'{scan_outcome.violation_summary}'
)
return HookDecision.deny(
AiHookEventType.FILE_READ,
user_message,
'Secrets detected; do not send this file to the model.',
AiHookEventType.FILE_READ, user_message, _false_positive_agent_hint(scan_outcome.shas)
)
outcome = AIHookOutcome.WARNED
user_message = f'Cycode detected secrets in {file_path}. {scan_outcome.violation_summary}'
user_message = f'Cycode detected secrets in {file_path}.\n{scan_outcome.violation_summary}'
return HookDecision.ask(
AiHookEventType.FILE_READ,
user_message,
Expand Down Expand Up @@ -282,7 +329,7 @@ def _handle_arg_scan(
return HookDecision.deny(
feature.event_type,
feature.deny_message(scan_outcome.violation_summary),
feature.deny_agent_message,
f'{feature.deny_agent_message} {_false_positive_agent_hint(scan_outcome.shas)}',
)
outcome = AIHookOutcome.WARNED
return HookDecision.ask(
Expand Down Expand Up @@ -360,9 +407,11 @@ def handle_before_mcp_execution(ctx: typer.Context, payload: AIHookPayload, poli
policy,
_ArgScanFeature(
event_type=AiHookEventType.MCP_EXECUTION,
deny_message=lambda v: f'Cycode blocked MCP tool call "{tool}". {v}',
deny_message=lambda v: (
f'Cycode blocked MCP tool call "{tool}" because its arguments contain a secret.\n{v}'
),
deny_agent_message='Do not pass secrets to tools. Use secret references (name/id) instead.',
ask_message=lambda v: f'Allow MCP tool call "{tool}"? {v}',
ask_message=lambda v: f'Allow MCP tool call "{tool}"?\n{v}',
ask_agent_message='Possible secrets detected in tool arguments; proceed with caution.',
scan_enabled=get_policy_value(policy, 'mcp', 'scan_args', default=True),
),
Expand Down Expand Up @@ -478,11 +527,14 @@ def _perform_scan(
if not local_scan_result:
return NO_SCAN

violation_summary = build_violation_summary([local_scan_result]) if local_scan_result.issue_detected else None
violation_summary, shas = (
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),
shas=shas,
)


Expand Down
52 changes: 35 additions & 17 deletions cycode/cli/apps/ai_guardrails/scan/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,17 @@
import os
import sys
from collections import defaultdict
from gettext import ngettext
from pathlib import Path
from typing import TYPE_CHECKING

from cycode.cli.apps.ai_guardrails.scan.policy import get_policy_value
from cycode.cli.cli_types import SeverityOption
from cycode.cli.utils.string_utils import obfuscate_text

if TYPE_CHECKING:
from cycode.cli.models import LocalScanResult
from cycode.cyclient.models import Detection

# Keeps the hook message readable when a single file trips dozens of detections
MAX_VIOLATION_DETAIL_LINES = 5
Expand Down Expand Up @@ -98,35 +101,50 @@ def output_json(obj: dict) -> None:
print(json.dumps(obj), end='') # noqa: T201


def _build_detection_lines(
def _describe_detection(detection: 'Detection', content: str) -> str:
details = detection.detection_details
description = detection.type or detection.message

line = details.get('line')
if isinstance(line, int) and line >= 0:
description += f' (line {line + 1})' # the scanner reports 0-based lines

start, length = details.get('start_position'), details.get('length')
if isinstance(start, int) and isinstance(length, int) and start >= 0 and length > 0:
value = content[start : start + length]
if value:
description += f': {obfuscate_text(value)}'

return description


def build_violation_summary(
local_scan_results: list['LocalScanResult'], max_lines: int = MAX_VIOLATION_DETAIL_LINES
) -> str:
"""One line per distinct finding: what it is, and the value hash identifying it.
) -> tuple[str, tuple[str, ...]]:
"""One line per distinct finding (what it is, where, the masked value), and the value hashes to ignore them by.

The value hash is safe to display; the value itself is not. Detections excluded by an existing
The masked value and hash are safe to display; the value itself is not. Detections excluded by an existing
ignore rule are already gone from `document_detections`, so only what actually blocked is listed.
"""
type_by_sha = {}
# Keyed by value hash so a repeated value is listed once; detections without one key by their description
finding_by_key = {}
for local_scan_result in local_scan_results:
for document_detections in local_scan_result.document_detections:
for detection in document_detections.detections:
description = _describe_detection(detection, document_detections.document.content)
sha = detection.detection_details.get('sha512')
if sha and sha not in type_by_sha:
type_by_sha[sha] = detection.type or detection.message

if not type_by_sha:
return ''
finding_by_key.setdefault(sha or description, (sha, description))

lines = [f' - {detection_type}: {sha}' for sha, detection_type in list(type_by_sha.items())[:max_lines]]
remaining = len(type_by_sha) - len(lines)
listed = list(finding_by_key.values())[:max_lines]
lines = [_build_severity_line(local_scan_results)] + [f' - {description}' for _, description in listed]
remaining = len(finding_by_key) - len(listed)
if remaining:
lines.append(f' - ...and {remaining} more')

return '\n' + '\n'.join(lines)
return '\n'.join(lines), tuple(sha for sha, _ in listed if sha)


def build_violation_summary(local_scan_results: list['LocalScanResult']) -> str:
"""Build violation summary string with severity breakdown and emojis."""
def _build_severity_line(local_scan_results: list['LocalScanResult']) -> str:
detections_count = 0
severity_counts = defaultdict(int)

Expand All @@ -143,5 +161,5 @@ def build_violation_summary(local_scan_results: list['LocalScanResult']) -> str:
count = severity_counts[severity]
severity_parts.append(f'{emoji} {severity.upper()} - {count}')

summary = f'Cycode found {detections_count} violations: {" | ".join(severity_parts)}'
return summary + _build_detection_lines(local_scan_results)
violations = ngettext('violation', 'violations', detections_count)
return f'Cycode found {detections_count} {violations}: {" | ".join(severity_parts)}'
5 changes: 4 additions & 1 deletion tests/cli/commands/ai_guardrails/ides/test_claude_code.py
Original file line number Diff line number Diff line change
Expand Up @@ -194,9 +194,11 @@ def test_parse_empty_payload_defaults() -> None:
def test_build_prompt_responses() -> None:
claude = ClaudeCode()
assert claude.build_hook_response(HookDecision.allow(AiHookEventType.PROMPT)) == {}
# The blocked prompt carries the secret, so it must not be echoed back under the reason
assert claude.build_hook_response(HookDecision.deny(AiHookEventType.PROMPT, 'no!')) == {
'decision': 'block',
'reason': 'no!',
'hookSpecificOutput': {'hookEventName': 'UserPromptSubmit', 'suppressOriginalPrompt': True},
}


Expand All @@ -210,7 +212,8 @@ def test_build_permission_responses() -> None:
'hookSpecificOutput': {
'hookEventName': 'PreToolUse',
'permissionDecision': 'deny',
'permissionDecisionReason': 'user!',
# A deny reason reaches the model, which needs both the findings and its instructions
'permissionDecisionReason': 'user!\nagent!',
}
}

Expand Down
4 changes: 2 additions & 2 deletions tests/cli/commands/ai_guardrails/ides/test_codex.py
Original file line number Diff line number Diff line change
Expand Up @@ -149,12 +149,12 @@ def test_build_mcp_execution_allow_and_deny() -> None:
# Explicit permissionDecision "allow" is rejected by older Codex runtimes.
assert allow == {}

deny = codex.build_hook_response(HookDecision.deny(AiHookEventType.MCP_EXECUTION, 'secret in args!'))
deny = codex.build_hook_response(HookDecision.deny(AiHookEventType.MCP_EXECUTION, 'secret in args!', 'agent!'))
assert deny == {
'hookSpecificOutput': {
'hookEventName': 'PreToolUse',
'permissionDecision': 'deny',
'permissionDecisionReason': 'secret in args!',
'permissionDecisionReason': 'secret in args!\nagent!',
}
}

Expand Down
4 changes: 2 additions & 2 deletions tests/cli/commands/ai_guardrails/ides/test_copilot.py
Original file line number Diff line number Diff line change
Expand Up @@ -282,12 +282,12 @@ def test_deny_prompt_response_shape() -> None:


def test_deny_tool_response_shape() -> None:
response = Copilot().build_hook_response(HookDecision.deny(AiHookEventType.FILE_READ, 'Sensitive file'))
response = Copilot().build_hook_response(HookDecision.deny(AiHookEventType.FILE_READ, 'Sensitive file', 'agent!'))
assert response == {
'hookSpecificOutput': {
'hookEventName': 'PreToolUse',
'permissionDecision': 'deny',
'permissionDecisionReason': 'Sensitive file',
'permissionDecisionReason': 'Sensitive file\nagent!',
}
}

Expand Down
11 changes: 9 additions & 2 deletions tests/cli/commands/ai_guardrails/ides/test_cursor.py
Original file line number Diff line number Diff line change
Expand Up @@ -175,15 +175,22 @@ def test_build_prompt_responses() -> None:
def test_build_permission_responses() -> None:
cursor = Cursor()
assert cursor.build_hook_response(HookDecision.allow(AiHookEventType.FILE_READ)) == {'permission': 'allow'}
# beforeReadFile drops agent_message and feeds user_message to the model, so it carries both
assert cursor.build_hook_response(HookDecision.deny(AiHookEventType.FILE_READ, 'user!', 'agent!')) == {
'permission': 'deny',
'user_message': 'user!\nagent!',
'agent_message': 'user!\nagent!',
}
assert cursor.build_hook_response(HookDecision.deny(AiHookEventType.MCP_EXECUTION, 'user!', 'agent!')) == {
'permission': 'deny',
'user_message': 'user!',
'agent_message': 'agent!',
# The agent never sees user_message, so it gets the findings too
'agent_message': 'user!\nagent!',
}
assert cursor.build_hook_response(HookDecision.ask(AiHookEventType.MCP_EXECUTION, 'u', 'a')) == {
'permission': 'ask',
'user_message': 'u',
'agent_message': 'a',
'agent_message': 'u\na',
}


Expand Down
Loading
Loading