cycodehq · Altruistus · Sep 25, 2026
@@ -34,6 +34,13 @@ class GuardrailCellMode(str, Enum):
     BLOCK = GuardrailsMode.BLOCK.value
 
 
+class McpServerEnforceOn(str, Enum):
+    """Which MCP servers the unauthorized MCP server guardrail enforces on (its `enforce_on` setting)."""
+
+    UNAUTHORIZED = 'unauthorized'  # only servers explicitly marked Unauthorized
+    NOT_AUTHORIZED = 'not_authorized'  # strict: anything that isn't Authorized, servers ASM hasn't seen included
+
+
 # Base CLI commands invoked from installed hooks. IDE classes append --ide flags
 # (and any other suffix) on top of these.
 CYCODE_SCAN_PROMPT_COMMAND = 'cycode ai-guardrails scan'

@@ -64,6 +64,50 @@ def _load_cursor_mcp_config(config_path: Optional[Path] = None) -> Optional[dict
         return None
 
 
+def _cursor_mcp_config_paths(workspace_roots: object) -> list[Path]:
+    """The MCP configs a Cursor session may load servers from: each project's, then the user's."""
+    roots = workspace_roots if isinstance(workspace_roots, list) else []
+    paths = [Path(root) / _REPO_SUBDIR / _MCP_CONFIG_FILENAME for root in roots if isinstance(root, str) and root]
+    paths.append(_cursor_mcp_config_path())
+    return paths
+
+
+def _server_command_line(server: dict) -> Optional[str]:
+    command = server.get('command')
+    if not isinstance(command, str) or not command:
+        return None
+    args = server.get('args') if isinstance(server.get('args'), list) else []
+    return ' '.join([command, *(str(arg) for arg in args)])
+
+
+def _resolve_mcp_server_name(raw_payload: dict) -> Optional[str]:
+    """The ``mcp.json`` entry name of the server a beforeMCPExecution payload called.
+
+    Cursor identifies the server by its ``url`` (remote servers) or ``command`` (stdio servers)
+    rather than by the entry name the platform stores servers under, so look the entry up in the
+    MCP configs: a url matches exactly; a command matches the entry's command line (command + args),
+    or the entry name itself. Falls back to the raw ``command`` when no entry matches.
+    """
+    url = raw_payload.get('url')
+    command = raw_payload.get('command')
+    if not url and not command:
+        return None
+
+    for config_path in _cursor_mcp_config_paths(raw_payload.get('workspace_roots')):
+        servers = (_load_cursor_mcp_config(config_path) or {}).get('mcpServers')
+        if not isinstance(servers, dict):
+            continue
+        for name, server in servers.items():
+            if not isinstance(server, dict):
+                continue
+            if url and server.get('url') == url:
+                return name
+            if command and (command == _server_command_line(server) or command.lower() == str(name).lower()):
+                return name
+
+    return command
+
+
 class Cursor(IDE):
     name: ClassVar[str] = 'cursor'
     display_name: ClassVar[str] = 'Cursor'
@@ -95,7 +139,9 @@ def parse_hook_payload(self, raw_payload: dict) -> AIHookPayload:
             ide_version=raw_payload.get('cursor_version'),
             prompt=raw_payload.get('prompt', ''),
             file_path=raw_payload.get('file_path') or raw_payload.get('path'),
-            mcp_server_name=raw_payload.get('command'),
+            mcp_server_name=(
+                _resolve_mcp_server_name(raw_payload) if canonical_event == AiHookEventType.MCP_EXECUTION else None
+            ),
             mcp_tool_name=raw_payload.get('tool_name') or raw_payload.get('tool'),
             mcp_arguments=(raw_payload.get('arguments') or raw_payload.get('tool_input') or raw_payload.get('input')),
         )

@@ -3,7 +3,7 @@
 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.
+everywhere + the default globs, with the unauthorized MCP server guardrail Off), always synchronous.
 """
 
 import json
@@ -12,7 +12,7 @@
 from pathlib import Path
 from typing import Optional
 
-from cycode.cli.apps.ai_guardrails.consts import GuardrailCellMode, PolicyMode
+from cycode.cli.apps.ai_guardrails.consts import GuardrailCellMode, McpServerEnforceOn, PolicyMode
 from cycode.cli.apps.ai_guardrails.scan.consts import DEFAULT_SENSITIVE_PATH_GLOBS
 from cycode.cli.apps.ai_guardrails.scan.types import BlockReason
 from cycode.cli.consts import CYCODE_CONFIGURATION_DIRECTORY
@@ -23,7 +23,7 @@
 
 GUARDRAILS_CONFIG_FILE_NAME = 'ai-guardrails-config.json'
 
-_DEFAULT_TTL_SECONDS = 900
+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.
@@ -34,14 +34,26 @@
         BlockReason.SECRETS_IN_FILE,
         BlockReason.SENSITIVE_PATH,
         BlockReason.SECRETS_IN_MCP_ARGS,
+        BlockReason.UNAUTHORIZED_MCP_SERVER,
     )
 )
 
+# Guardrails that are Off unless the platform switches them on for an agent. The rest default to
+# Report; these would otherwise start enforcing on a tenant that never looked at them.
+_DEFAULT_OFF_GUARDRAIL_KEYS = frozenset((BlockReason.UNAUTHORIZED_MCP_SERVER.value,))
+
 
 def get_config_cache_path() -> Path:
     return Path.home() / CYCODE_CONFIGURATION_DIRECTORY / GUARDRAILS_CONFIG_FILE_NAME
 
 
+def default_mode_for(guardrail_key: str) -> str:
+    """A guardrail's mode for an agent the platform sent no cell for (or with no cache at all)."""
+    if guardrail_key in _DEFAULT_OFF_GUARDRAIL_KEYS:
+        return GuardrailCellMode.OFF.value
+    return GuardrailCellMode.REPORT.value
+
+
 def _default_sensitive_globs() -> list:
     return list(DEFAULT_SENSITIVE_PATH_GLOBS)
 
@@ -63,7 +75,12 @@ def __post_init__(self) -> None:
     def mode_for(self, guardrail_key: str, ide_name: Optional[str]) -> str:
         """The platform keys the cells by our --ide names, so the lookup is direct."""
         agents = (self._guardrails.get(guardrail_key) or {}).get('agents') or {}
-        return str(agents.get((ide_name or '').lower(), GuardrailCellMode.REPORT.value)).lower()
+        return str(agents.get((ide_name or '').lower(), default_mode_for(guardrail_key))).lower()
+
+    def is_off_for_every_agent(self, guardrail_key: str) -> bool:
+        """No agent has this guardrail on - nothing it needs has to be fetched."""
+        agents = (self._guardrails.get(guardrail_key) or {}).get('agents') or {}
+        return all(str(mode).lower() == GuardrailCellMode.OFF for mode in agents.values())
 
     def _modes_for_event(self, event_name: str, ide_name: Optional[str]) -> list:
         return [
@@ -86,8 +103,16 @@ def sensitive_globs(self) -> list:
         globs = settings.get('globs')
         return globs if isinstance(globs, list) and globs else _default_sensitive_globs()
 
+    def mcp_server_enforce_on(self) -> str:
+        """Unknown values read as the default, which enforces on fewer servers - config must never fail closed."""
+        settings = (self._guardrails.get(BlockReason.UNAUTHORIZED_MCP_SERVER) or {}).get('settings') or {}
+        enforce_on = str(settings.get('enforce_on') or '').lower()
+        if enforce_on == McpServerEnforceOn.NOT_AUTHORIZED:
+            return McpServerEnforceOn.NOT_AUTHORIZED.value
+        return McpServerEnforceOn.UNAUTHORIZED.value
+
     def is_expired(self) -> bool:
-        ttl = self.payload.get('ttl_seconds') or _DEFAULT_TTL_SECONDS
+        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:
@@ -99,14 +124,15 @@ def apply_platform_config(policy: dict, config: Optional[GuardrailConfig], ide_n
     """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.
+    Report everywhere with the default globs, the unauthorized MCP server guardrail Off - 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 guardrails sharing an event (FileRead: content
+    scan vs. sensitive path; McpExecution: argument scan vs. server authorization) keep independent modes.
     An all-Off event never reaches here at all: scan_command skips it.
     """
 
     def cell(guardrail_key: str) -> str:
-        return config.mode_for(guardrail_key, ide_name) if config is not None else GuardrailCellMode.REPORT.value
+        return config.mode_for(guardrail_key, ide_name) if config is not None else default_mode_for(guardrail_key)
 
     def action(guardrail_key: str) -> str:
         return PolicyMode.BLOCK.value if cell(guardrail_key) == GuardrailCellMode.BLOCK else PolicyMode.WARN.value
@@ -123,7 +149,14 @@ def action(guardrail_key: str) -> str:
     )
     file_read['path_action'] = action(BlockReason.SENSITIVE_PATH)
 
-    policy.setdefault('mcp', {})['action'] = action(BlockReason.SECRETS_IN_MCP_ARGS)
+    mcp = policy.setdefault('mcp', {})
+    mcp['scan_args'] = cell(BlockReason.SECRETS_IN_MCP_ARGS) != GuardrailCellMode.OFF
+    mcp['action'] = action(BlockReason.SECRETS_IN_MCP_ARGS)
+    mcp['check_server'] = cell(BlockReason.UNAUTHORIZED_MCP_SERVER) != GuardrailCellMode.OFF
+    mcp['server_action'] = action(BlockReason.UNAUTHORIZED_MCP_SERVER)
+    mcp['server_enforce_on'] = (
+        config.mcp_server_enforce_on() if config is not None else McpServerEnforceOn.UNAUTHORIZED.value
+    )
 
 
 def save_guardrail_config(payload: dict, tenant_id: Optional[str]) -> None:

@@ -20,8 +20,9 @@
 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.consts import GuardrailsMode, McpServerEnforceOn, PolicyMode
 from cycode.cli.apps.ai_guardrails.ides.base import HookDecision
+from cycode.cli.apps.ai_guardrails.scan.mcp_server_status import is_enforced, load_mcp_server_statuses
 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 (
@@ -218,6 +219,16 @@ class _ArgScanFeature:
     deny_agent_message: str
     ask_message: Callable[[str], str]
     ask_agent_message: str
+    scan_enabled: bool = True
+
+
+class _PreScanFinding(NamedTuple):
+    """A guardrail that fired on the call itself, before its text is scanned (e.g. an unauthorized MCP server)."""
+
+    block_reason: BlockReason
+    mode: GuardrailsMode
+    deny_message: str
+    deny_agent_message: str
 
 
 def _handle_arg_scan(
@@ -226,8 +237,14 @@ def _handle_arg_scan(
     policy: dict,
     feature: _ArgScanFeature,
     scan_text: str,
+    pre_scan_finding: Optional[_PreScanFinding] = None,
 ) -> HookDecision:
-    """Shared scan + decision flow for MCP_EXECUTION and COMMAND_EXEC events."""
+    """Shared scan + decision flow for MCP_EXECUTION and COMMAND_EXEC events.
+
+    A pre-scan finding in Block mode denies without scanning. In Report mode it marks the event
+    warned and the scan still runs; a secret the scan then finds takes over the response and the
+    event's block reason and outcome, since that is what the user is shown.
+    """
     ai_client = ctx.obj['ai_security_client']
 
     max_bytes = get_policy_value(policy, 'secrets', 'max_bytes', default=200000)
@@ -240,6 +257,18 @@ def _handle_arg_scan(
     error_message = None
 
     try:
+        if pre_scan_finding is not None:
+            block_reason = pre_scan_finding.block_reason
+            if pre_scan_finding.mode == GuardrailsMode.BLOCK:
+                outcome = AIHookOutcome.BLOCKED
+                return HookDecision.deny(
+                    feature.event_type, pre_scan_finding.deny_message, pre_scan_finding.deny_agent_message
+                )
+            outcome = AIHookOutcome.WARNED
+
+        if not feature.scan_enabled:
+            return HookDecision.allow(feature.event_type)
+
         scan_outcome = _scan_text_for_secrets(
             ctx,
             clipped,
@@ -283,8 +312,56 @@ def _handle_arg_scan(
         )
 
 
+def _check_mcp_server_authorization(payload: AIHookPayload, policy: dict) -> Optional[_PreScanFinding]:
+    """The unauthorized MCP server guardrail: a finding when the called server is enforced, else None.
+
+    Fails open - no server name, or no cached statuses, means the server is let through.
+    """
+    mcp_config = get_policy_value(policy, 'mcp', default={})
+    if not get_policy_value(mcp_config, 'check_server', default=False):
+        return None
+
+    alias = payload.mcp_server_name
+    if not alias:
+        logger.debug('No MCP server name in the payload; skipping the server authorization check')
+        return None
+
+    statuses = load_mcp_server_statuses()
+    if statuses is None:
+        logger.debug('No cached MCP server statuses; skipping the server authorization check')
+        return None
+
+    server = statuses.match(alias)
+    if server is not None and server.alias.lower() != alias.lower():
+        # A normalized or plugin-namespaced name: report the alias the platform resolves servers by.
+        payload.mcp_server_name = server.alias
+
+    enforce_on = get_policy_value(mcp_config, 'server_enforce_on', default=McpServerEnforceOn.UNAUTHORIZED.value)
+    status = server.status if server is not None else None
+    if not is_enforced(status, enforce_on):
+        return None
+
+    logger.debug(
+        'MCP server is not authorized, %s',
+        {'mcp_server_name': payload.mcp_server_name, 'status': status, 'enforce_on': enforce_on},
+    )
+    server_name = payload.mcp_server_name
+    return _PreScanFinding(
+        block_reason=BlockReason.UNAUTHORIZED_MCP_SERVER,
+        mode=get_effective_mode(mcp_config, action_key='server_action'),
+        deny_message=(
+            f"Cycode blocked MCP server '{server_name}': it is not authorized in your organization. "
+            'Contact your admin to authorize it.'
+        ),
+        deny_agent_message=(
+            f"The MCP server '{server_name}' is not authorized in this organization. "
+            'Do not retry its tools or reach it another way.'
+        ),
+    )
+
+
 def handle_before_mcp_execution(ctx: typer.Context, payload: AIHookPayload, policy: dict) -> HookDecision:
-    """Scan MCP tool arguments for secrets before execution."""
+    """Check the MCP server is authorized, then scan the tool arguments for secrets before execution."""
     tool = payload.mcp_tool_name or 'unknown'
     args = payload.mcp_arguments or {}
     args_text = args if isinstance(args, str) else json.dumps(args)
@@ -298,8 +375,10 @@ def handle_before_mcp_execution(ctx: typer.Context, payload: AIHookPayload, poli
             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_agent_message='Possible secrets detected in tool arguments; proceed with caution.',
+            scan_enabled=get_policy_value(policy, 'mcp', 'scan_args', default=True),
         ),
         scan_text=args_text,
+        pre_scan_finding=_check_mcp_server_authorization(payload, policy),
     )