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
12 changes: 8 additions & 4 deletions cycode/cli/apps/ai_guardrails/hooks_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,16 +127,20 @@ def install_hooks(
ide: IDE,
scope: str = 'user',
repo_path: Optional[Path] = None,
report_mode: bool = False,
) -> tuple[bool, str]:
"""Install Cycode AI guardrails hooks for ``ide``."""
hooks_path = ide.settings_path(scope, repo_path)

existing = _load_hooks_file(hooks_path) or {'version': 1, 'hooks': {}}
existing.setdefault('version', 1)
existing = _load_hooks_file(hooks_path) or {'hooks': {}}
existing.setdefault('hooks', {})

rendered = ide.render_hooks_config(async_mode=report_mode)
rendered = ide.render_hooks_config()

# Top-level fields come from the IDE's render only - Codex rejects a hooks file
# with unknown top-level fields, so no `version` may be injected here.
for key, value in rendered.items():
if key != 'hooks':
existing[key] = value

for event, entries in rendered['hooks'].items():
existing['hooks'].setdefault(event, [])
Expand Down
21 changes: 1 addition & 20 deletions cycode/cli/apps/ai_guardrails/ides/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@
JSON response shape that the IDE expects on stdout.
"""

import platform
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
Expand All @@ -25,24 +24,6 @@
from cycode.cli.apps.ai_guardrails.scan.types import AiHookEventType


def shell_background_suffix(async_mode: bool) -> str:
"""`' &'` when backgrounding is requested and the platform's shell supports it.

Only valid for hooks whose runner is stdin-safe under backgrounding (zsh keeps
a backgrounded command's stdin; verified for Cursor/Codex). bash/sh reattach it
to /dev/null, silently emptying the payload — hooks that run under bash (e.g.
Copilot's `bash` field) must add an explicit `<&0` redirect instead.

Windows gets no suffix: depending on the IDE, hooks may run under cmd (where a
trailing `&` is a no-op separator) or Windows PowerShell (where it's a parse
error that would fail the hook). Until the CLI can self-detach in report mode,
Windows hooks run synchronously.
"""
if not async_mode or platform.system() == 'Windows':
return ''
return ' &'


class DecisionAction(str, Enum):
"""Canonical decision action returned by event handlers."""

Expand Down Expand Up @@ -118,7 +99,7 @@ def settings_path(self, scope: str, repo_path: Optional[Path] = None) -> Path:
"""

@abstractmethod
def render_hooks_config(self, async_mode: bool = False) -> dict:
def render_hooks_config(self) -> dict:
"""Return the settings blob to merge into the IDE's settings file.

Shape is IDE-specific (Cursor uses a flat ``{event: [{command}]}`` dict;
Expand Down
6 changes: 1 addition & 5 deletions cycode/cli/apps/ai_guardrails/ides/claude_code.py
Original file line number Diff line number Diff line change
Expand Up @@ -249,12 +249,8 @@ def settings_path(self, scope: str, repo_path: Optional[Path] = None) -> Path:
return repo_path / _REPO_SUBDIR / _HOOKS_FILE_NAME
return _USER_HOOKS_DIR / _HOOKS_FILE_NAME

def render_hooks_config(self, async_mode: bool = False) -> dict:
# Claude Code uses a nested hook structure with optional async/timeout.
def render_hooks_config(self) -> dict:
hook_entry: dict = {'type': 'command', 'command': _SCAN_COMMAND}
if async_mode:
hook_entry['async'] = True
hook_entry['timeout'] = 20

return {
'hooks': {
Expand Down
11 changes: 5 additions & 6 deletions cycode/cli/apps/ai_guardrails/ides/codex.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
resolve_cached_plugin_dir,
walk_enabled_plugins,
)
from cycode.cli.apps.ai_guardrails.ides.base import IDE, DecisionAction, HookDecision, shell_background_suffix
from cycode.cli.apps.ai_guardrails.ides.base import IDE, DecisionAction, HookDecision
from cycode.cli.apps.ai_guardrails.scan.payload import AIHookPayload
from cycode.cli.apps.ai_guardrails.scan.types import AiHookEventType
from cycode.cli.utils.jwt_utils import decode_jwt_unverified
Expand Down Expand Up @@ -189,11 +189,10 @@ def settings_path(self, scope: str, repo_path: Optional[Path] = None) -> Path:
return repo_path / _CONFIG_DIR_NAME / _HOOKS_FILE_NAME
return _codex_home() / _HOOKS_FILE_NAME

def render_hooks_config(self, async_mode: bool = False) -> dict:
# Codex's TOML `async: true` flag is unimplemented; shell-background via
# `&` is the working mechanism (unix only). SessionStart stays sync so
# the conversation context is registered before any scan hook fires.
scan_cmd = f'{_SCAN_COMMAND}{shell_background_suffix(async_mode)}'
def render_hooks_config(self) -> dict:
# SessionStart stays sync so the conversation context is registered
# before any scan hook fires.
scan_cmd = _SCAN_COMMAND
return {
'hooks': {
'SessionStart': [
Expand Down
19 changes: 2 additions & 17 deletions cycode/cli/apps/ai_guardrails/ides/copilot.py
Original file line number Diff line number Diff line change
Expand Up @@ -379,24 +379,9 @@ def settings_path(self, scope: str, repo_path: Optional[Path] = None) -> Path:
return repo_path / _REPO_HOOKS_SUBDIR / _HOOKS_FILE_NAME
return _copilot_home() / 'hooks' / _HOOKS_FILE_NAME

def render_hooks_config(self, async_mode: bool = False) -> dict:
def render_hooks_config(self) -> dict:
# Single cross-platform `command` field, copied to both shells by Copilot.
def entry(command: str) -> dict:
if async_mode:
# Copilot has no async hook flag; background via shell on unix. Both
# redirects are load-bearing. `<&0` keeps the payload flowing: a bare
# `cmd &` gets its stdin reattached to /dev/null by the shell (job
# control is off in hooks), so the scan reads nothing and allows. The
# stdout redirect is what actually makes it async: the backgrounded
# child inherits the hook's stdout and the runner waits on that pipe
# for EOF, so without it the scan blocks the response it was meant to
# run behind. Windows PowerShell has no trailing-&, so it stays sync.
return {
'type': 'command',
'bash': f'{command} <&0 >/dev/null 2>&1 &',
'powershell': command,
'timeoutSec': _HOOK_TIMEOUT_SEC,
}
# Single cross-platform `command` field, copied to both shells by Copilot.
return {'type': 'command', 'command': command, 'timeoutSec': _HOOK_TIMEOUT_SEC}

return {
Expand Down
7 changes: 3 additions & 4 deletions cycode/cli/apps/ai_guardrails/ides/cursor.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

from cycode.cli.apps.ai_guardrails.consts import CYCODE_SCAN_PROMPT_COMMAND, CYCODE_SESSION_START_COMMAND
from cycode.cli.apps.ai_guardrails.ides._plugin_utils import build_global_config_file
from cycode.cli.apps.ai_guardrails.ides.base import IDE, DecisionAction, HookDecision, shell_background_suffix
from cycode.cli.apps.ai_guardrails.ides.base import IDE, DecisionAction, HookDecision
from cycode.cli.apps.ai_guardrails.scan.payload import AIHookPayload
from cycode.cli.apps.ai_guardrails.scan.types import AiHookEventType
from cycode.logger import get_logger
Expand Down Expand Up @@ -68,9 +68,8 @@ def settings_path(self, scope: str, repo_path: Optional[Path] = None) -> Path:
return repo_path / _REPO_SUBDIR / _HOOKS_FILE_NAME
return _user_hooks_dir() / _HOOKS_FILE_NAME

def render_hooks_config(self, async_mode: bool = False) -> dict:
command = f'{_SCAN_COMMAND}{shell_background_suffix(async_mode)}'
hooks = {event: [{'command': command}] for event in self.hook_events}
def render_hooks_config(self) -> dict:
hooks = {event: [{'command': _SCAN_COMMAND}] for event in self.hook_events}
hooks['sessionStart'] = [{'command': _SESSION_START_COMMAND}]
return {'version': 1, 'hooks': hooks}

Expand Down
6 changes: 2 additions & 4 deletions cycode/cli/apps/ai_guardrails/install_command.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,11 +65,9 @@ def install_command(
repo_path = resolve_repo_path(scope, repo_path)
ides_to_install = resolve_ides(ide)

report_mode = mode == GuardrailsMode.REPORT

results: list[tuple[str, bool, str]] = []
for current_ide in ides_to_install:
success, message = install_hooks(current_ide, scope, repo_path, report_mode=report_mode)
success, message = install_hooks(current_ide, scope, repo_path)
results.append((current_ide.display_name, success, message))

any_success = False
Expand Down Expand Up @@ -108,6 +106,6 @@ def _print_next_steps(results: list[tuple[str, bool, str]], mode: GuardrailsMode
console.print('2. (Optional) Customize policy in ~/.cycode/ai-guardrails.yaml')
console.print()
if mode == GuardrailsMode.REPORT:
console.print('[dim]Report mode: hooks run async (non-blocking) and policy is set to warn.[/]')
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.[/]')
71 changes: 71 additions & 0 deletions cycode/cli/apps/ai_guardrails/scan/detach.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
"""Self-detach support for report-mode scans.

In report mode nobody consumes the hook verdict, so the scan respawns itself
detached and the parent exits immediately — the IDE is released after roughly
CLI startup instead of waiting for a full scan. The child's handles are set at
process-creation time (stdout/stderr to devnull, payload piped to stdin), so it
never inherits the IDE's pipes: a backgrounded child that shares the hook's
stdout keeps EOF-waiting runners blocked for the scan's full duration.
"""

import os
import subprocess
import sys

from cycode.logger import get_logger

logger = get_logger('AI Guardrails')

DETACHED_ENV_VAR = '_CYCODE_DETACHED'

# Numeric fallbacks let non-Windows platforms (tests included) build the same flags.
_WINDOWS_CREATIONFLAGS = (
getattr(subprocess, 'DETACHED_PROCESS', 0x00000008)
| getattr(subprocess, 'CREATE_NEW_PROCESS_GROUP', 0x00000200)
| getattr(subprocess, 'CREATE_NO_WINDOW', 0x08000000)
)


def is_detached_child() -> bool:
"""Whether this process is the respawned detached child."""
return os.environ.get(DETACHED_ENV_VAR) == '1'


def build_respawn_command() -> list[str]:
"""The command that re-runs the current invocation.

Under PyInstaller sys.executable is the CLI binary itself; otherwise it is
the Python interpreter and argv[0] is the console script to re-run.
"""
if getattr(sys, 'frozen', False):
return [sys.executable, *sys.argv[1:]]
return [sys.executable, *sys.argv]


def respawn_detached(stdin_payload: str) -> bool:
"""Respawn the current command detached, feeding it ``stdin_payload``.

Returns False when the respawn failed, so the caller can fall back to the
synchronous path instead of dropping the event.
"""
try:
detach_kwargs = (
{'creationflags': _WINDOWS_CREATIONFLAGS} if sys.platform == 'win32' else {'start_new_session': True}
)
process = subprocess.Popen( # noqa: S603
build_respawn_command(),
stdin=subprocess.PIPE,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
env={**os.environ, DETACHED_ENV_VAR: '1'},
**detach_kwargs,
)
# Hand over the payload and close, then return without waiting - the
# whole point is that the parent exits while the child scans.
process.stdin.write(stdin_payload.encode('utf-8'))
process.stdin.close()
logger.debug('Respawned detached scan', extra={'child_pid': process.pid})
return True
except Exception as e:
logger.debug('Failed to respawn detached, falling back to synchronous scan', exc_info=e)
return False
24 changes: 24 additions & 0 deletions cycode/cli/apps/ai_guardrails/scan/handlers.py
Original file line number Diff line number Diff line change
Expand Up @@ -318,6 +318,30 @@ def get_effective_mode(policy: dict, feature_config: dict) -> GuardrailsMode:
return GuardrailsMode.BLOCK if (mode == PolicyMode.BLOCK and 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:
"""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.
"""
feature_key = _FEATURE_KEY_BY_EVENT_TYPE.get(event_name)
if feature_key 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


def build_ai_guardrails_scan_parameters(
ctx: typer.Context,
paths: Optional[tuple[str, ...]],
Expand Down
9 changes: 8 additions & 1 deletion cycode/cli/apps/ai_guardrails/scan/scan_command.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,8 @@

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.scan.handlers import get_handler_for_event
from cycode.cli.apps.ai_guardrails.scan.detach import is_detached_child, respawn_detached
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
from cycode.cli.apps.ai_guardrails.scan.utils import output_json, read_stdin_text, safe_json_parse
Expand Down Expand Up @@ -139,6 +140,12 @@ def scan_command(
workspace_roots = payload.get('workspace_roots') or ['.']
policy = load_policy(workspace_roots[0])

# 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):
return

try:
_initialize_clients(ctx)

Expand Down
40 changes: 11 additions & 29 deletions tests/cli/commands/ai_guardrails/ides/test_codex.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@

import pytest
from pyfakefs.fake_filesystem import FakeFilesystem
from pytest_mock import MockerFixture

from cycode.cli.apps.ai_guardrails.ides.base import HookDecision
from cycode.cli.apps.ai_guardrails.ides.codex import (
Expand Down Expand Up @@ -158,40 +157,23 @@ def test_render_hooks_session_start_matches_all_sources() -> None:

def test_render_hooks_never_emits_async_toml_flags() -> None:
"""Codex's TOML `async: true` / `timeout` flags are unimplemented; we must not emit them."""
for mode in (False, True):
rendered = Codex().render_hooks_config(async_mode=mode)
for entry in rendered['hooks']['PreToolUse']:
for hook in entry['hooks']:
assert 'async' not in hook
assert 'timeout' not in hook


def test_render_hooks_async_backgrounds_scan_hooks(mocker: MockerFixture) -> None:
"""In async mode, UserPromptSubmit + PreToolUse scan hooks shell-background (unix)."""
mocker.patch('platform.system', return_value='Linux')
rendered = Codex().render_hooks_config(async_mode=True)
prompt_cmd = rendered['hooks']['UserPromptSubmit'][0]['hooks'][0]['command']
pretool_cmd = rendered['hooks']['PreToolUse'][0]['hooks'][0]['command']
assert prompt_cmd.endswith(' &')
assert pretool_cmd.endswith(' &')
rendered = Codex().render_hooks_config()
for entry in rendered['hooks']['PreToolUse']:
for hook in entry['hooks']:
assert 'async' not in hook
assert 'timeout' not in hook


def test_render_hooks_async_windows_stays_sync(mocker: MockerFixture) -> None:
"""No '&' on Windows - nothing there detaches safely, so hooks run sync."""
mocker.patch('platform.system', return_value='Windows')
rendered = Codex().render_hooks_config(async_mode=True)
def test_render_hooks_commands_always_plain_sync() -> None:
"""Hooks are always installed plain sync: warn-vs-block synchronicity is
decided at scan time by the CLI itself (self-detach)."""
rendered = Codex().render_hooks_config()
prompt_cmd = rendered['hooks']['UserPromptSubmit'][0]['hooks'][0]['command']
pretool_cmd = rendered['hooks']['PreToolUse'][0]['hooks'][0]['command']
session_cmd = rendered['hooks']['SessionStart'][0]['hooks'][0]['command']
assert '&' not in prompt_cmd
assert '&' not in pretool_cmd


def test_render_hooks_session_start_always_synchronous() -> None:
"""SessionStart registers the conversation context — never backgrounded."""
for mode in (False, True):
rendered = Codex().render_hooks_config(async_mode=mode)
session_cmd = rendered['hooks']['SessionStart'][0]['hooks'][0]['command']
assert '&' not in session_cmd
assert '&' not in session_cmd


def test_render_hooks_pretooluse_matchers_are_mcp_only() -> None:
Expand Down
Loading
Loading