CM-74352: Treat not-reviewed AI entities as unauthorized

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: Altruistus

cycode/cli/apps/ai_guardrails/scan/handlers.py

@@ -331,12 +331,13 @@ def _check_mcp_server_authorization(payload: AIHookPayload, policy: dict) -> Opt
         payload.mcp_server_name = server.alias
 
     status = server.status if server is not None else None
-    if not is_enforced(status):
+    treat_unreviewed = statuses.treat_unreviewed_as_unauthorized
+    if not is_enforced(status, treat_unreviewed):
         return None
 
     logger.debug(
         'MCP server is not authorized, %s',
-        {'mcp_server_name': payload.mcp_server_name, 'status': status},
+        {'mcp_server_name': payload.mcp_server_name, 'status': status, 'treat_unreviewed': treat_unreviewed},
     )
     server_name = payload.mcp_server_name
     return _PreScanFinding(

cycode/cli/apps/ai_guardrails/scan/mcp_server_status.py

@@ -20,6 +20,8 @@
 
 MCP_SERVER_STATUSES_FILE_NAME = 'ai-guardrails-mcp-servers.json'
 
+TREAT_UNREVIEWED_AS_UNAUTHORIZED_KEY = 'should_treat_unreviewed_as_unauthorized'
+
 # Claude Code names plugin servers `plugin_<plugin>_<server>` in tool names; the platform stores `<server>`.
 _PLUGIN_ALIAS_PREFIX = 'plugin_'
 
@@ -56,9 +58,11 @@ def _normalize_alias(alias: str) -> str:
     return re.sub(r'[^a-z0-9_-]', '_', alias.lower())
 
 
-def is_enforced(status: Optional[McpServerAuthorizationStatus]) -> bool:
+def is_enforced(status: Optional[McpServerAuthorizationStatus], treat_unreviewed: bool) -> bool:
     """``status`` is None when the platform never saw the server."""
-    return status == McpServerAuthorizationStatus.UNAUTHORIZED
+    if status == McpServerAuthorizationStatus.UNAUTHORIZED:
+        return True
+    return treat_unreviewed and status != McpServerAuthorizationStatus.AUTHORIZED
 
 
 class McpServerMatch(NamedTuple):
@@ -72,6 +76,7 @@ class McpServerStatuses:
     fetched_at: float
     tenant_id: Optional[str] = None
     ttl_seconds: float = DEFAULT_TTL_SECONDS
+    treat_unreviewed_as_unauthorized: bool = False
     _by_alias: dict = field(init=False, repr=False)
 
     def __post_init__(self) -> None:
@@ -110,12 +115,18 @@ def needs_refresh(self, tenant_id: Optional[str]) -> bool:
         return self.is_expired() or self.tenant_id != tenant_id
 
 
-def save_mcp_server_statuses(servers: list, tenant_id: Optional[str], ttl_seconds: float = DEFAULT_TTL_SECONDS) -> None:
+def save_mcp_server_statuses(
+    servers: list,
+    tenant_id: Optional[str],
+    ttl_seconds: float = DEFAULT_TTL_SECONDS,
+    treat_unreviewed_as_unauthorized: bool = False,
+) -> None:
     path = get_mcp_server_statuses_cache_path()
     content = {
         'fetched_at': time.time(),
         'tenant_id': tenant_id,
         'ttl_seconds': ttl_seconds,
+        TREAT_UNREVIEWED_AS_UNAUTHORIZED_KEY: treat_unreviewed_as_unauthorized,
         'servers': servers,
     }
     try:
@@ -141,6 +152,7 @@ def load_mcp_server_statuses() -> Optional[McpServerStatuses]:
             fetched_at=float(content['fetched_at']),
             tenant_id=content.get('tenant_id'),
             ttl_seconds=float(content.get('ttl_seconds') or DEFAULT_TTL_SECONDS),
+            treat_unreviewed_as_unauthorized=content.get(TREAT_UNREVIEWED_AS_UNAUTHORIZED_KEY) is True,
         )
     except Exception as e:
         logger.warning('MCP server statuses cache is corrupt and will be moved aside', exc_info=e)

cycode/cli/apps/ai_guardrails/session_start_command.py

@@ -222,5 +222,10 @@ def _sync_mcp_server_statuses(
 
     response = ai_client.get_mcp_server_statuses()
     if response is not None:
-        save_mcp_server_statuses(response.servers, tenant_id, config.ttl_seconds)
+        save_mcp_server_statuses(
+            response.servers,
+            tenant_id,
+            config.ttl_seconds,
+            response.should_treat_unreviewed_as_unauthorized,
+        )
         logger.debug('MCP server statuses cache updated')

cycode/cyclient/ai_security_manager_client.py

@@ -15,6 +15,8 @@
 
 class McpServerStatusesResponse(NamedTuple):
     servers: list  # [{alias, normalized_id, status}]
+    # Strict mode: Unreviewed servers count as Unauthorized.
+    should_treat_unreviewed_as_unauthorized: bool
 
 
 class AISecurityManagerClient:
@@ -118,7 +120,10 @@ def get_mcp_server_statuses(self) -> Optional[McpServerStatusesResponse]:
             servers = body.get('servers')
             if not isinstance(servers, list):
                 raise ValueError('servers is not a list')
-            return McpServerStatusesResponse(servers=servers)
+            return McpServerStatusesResponse(
+                servers=servers,
+                should_treat_unreviewed_as_unauthorized=body.get('should_treat_unreviewed_as_unauthorized') is True,
+            )
         except Exception as e:
             logger.debug('Failed to fetch MCP server statuses', exc_info=e)
             return None

tests/cli/commands/ai_guardrails/scan/test_handlers.py

@@ -603,8 +603,12 @@ def _mcp_payload(server: Optional[str] = 'github') -> AIHookPayload:
     )
 
 
-def _cached_statuses(*rows: tuple[str, str]) -> McpServerStatuses:
-    return McpServerStatuses(servers=[{'alias': a, 'status': s} for a, s in rows], fetched_at=time.time())
+def _cached_statuses(*rows: tuple[str, str], treat_unreviewed: bool = False) -> McpServerStatuses:
+    return McpServerStatuses(
+        servers=[{'alias': a, 'status': s} for a, s in rows],
+        fetched_at=time.time(),
+        treat_unreviewed_as_unauthorized=treat_unreviewed,
+    )
 
 
 def _reported_event(mock_ctx: MagicMock) -> tuple[AIHookOutcome, Optional[BlockReason]]:
@@ -662,16 +666,20 @@ def test_unauthorized_mcp_server_report_then_secret_block_takes_over(
 
 
 @pytest.mark.parametrize(
-    ('rows', 'server', 'expected_action'),
+    ('rows', 'treat_unreviewed', 'server', 'expected_action'),
     [
-        # Only explicitly Unauthorized servers are enforced.
-        ((('github', 'Unreviewed'),), 'github', DecisionAction.ALLOW),
-        ((('github', 'Authorized'),), 'github', DecisionAction.ALLOW),
-        ((), 'github', DecisionAction.ALLOW),
+        # Default policy: only explicitly Unauthorized servers.
+        ((('github', 'Unreviewed'),), False, 'github', DecisionAction.ALLOW),
+        ((('github', 'Authorized'),), False, 'github', DecisionAction.ALLOW),
+        ((), False, 'github', DecisionAction.ALLOW),
+        # Strict: anything not Authorized, a server the platform never saw included.
+        ((('github', 'Unreviewed'),), True, 'github', DecisionAction.DENY),
+        ((), True, 'github', DecisionAction.DENY),
+        ((('github', 'Authorized'),), True, 'github', DecisionAction.ALLOW),
         # One alias, several servers: the most restrictive status wins.
-        ((('github', 'Authorized'), ('GitHub', 'Unauthorized')), 'github', DecisionAction.DENY),
-        # No server name to check: fail open.
-        ((('github', 'Unauthorized'),), None, DecisionAction.ALLOW),
+        ((('github', 'Authorized'), ('GitHub', 'Unauthorized')), False, 'github', DecisionAction.DENY),
+        # No server name to check: fail open, even in strict mode.
+        ((), True, None, DecisionAction.ALLOW),
     ],
 )
 @patch('cycode.cli.apps.ai_guardrails.scan.handlers._scan_text_for_secrets', return_value=ScanOutcome())
@@ -682,10 +690,11 @@ def test_unauthorized_mcp_server_enforcement(
     mock_ctx: MagicMock,
     default_policy: dict[str, Any],
     rows: tuple,
+    treat_unreviewed: bool,
     server: Optional[str],
     expected_action: DecisionAction,
 ) -> None:
-    mock_statuses.return_value = _cached_statuses(*rows)
+    mock_statuses.return_value = _cached_statuses(*rows, treat_unreviewed=treat_unreviewed)
     policy = _server_check_policy(default_policy)
 
     result = handle_before_mcp_execution(mock_ctx, _mcp_payload(server), policy)

tests/cli/commands/ai_guardrails/scan/test_mcp_server_status.py

@@ -1,5 +1,6 @@
 """Tests for the MCP server authorization status cache."""
 
+import json
 import time
 from pathlib import Path
 from typing import Optional
@@ -56,16 +57,20 @@ def test_parse_status_is_case_insensitive_and_unknown_reads_unreviewed(
 
 
 @pytest.mark.parametrize(
-    ('status', 'expected'),
+    ('status', 'treat_unreviewed', 'expected'),
     [
-        (_UNAUTHORIZED, True),
-        (_UNREVIEWED, False),
-        (_AUTHORIZED, False),
-        (None, False),
+        (_UNAUTHORIZED, False, True),
+        (_UNREVIEWED, False, False),
+        (_AUTHORIZED, False, False),
+        (None, False, False),
+        (_UNAUTHORIZED, True, True),
+        (_UNREVIEWED, True, True),
+        (None, True, True),
+        (_AUTHORIZED, True, False),
     ],
 )
-def test_is_enforced(status: Optional[McpServerAuthorizationStatus], expected: bool) -> None:
-    assert is_enforced(status) is expected
+def test_is_enforced(status: Optional[McpServerAuthorizationStatus], treat_unreviewed: bool, expected: bool) -> None:
+    assert is_enforced(status, treat_unreviewed) is expected
 
 
 def test_match_is_case_insensitive_and_returns_the_stored_alias() -> None:
@@ -125,6 +130,28 @@ def test_save_and_load_round_trip() -> None:
     assert statuses.ttl_seconds == 60
     assert statuses.needs_refresh('tenant-a') is False
     assert statuses.needs_refresh('tenant-b') is True
+    assert statuses.treat_unreviewed_as_unauthorized is False
+
+
+def test_save_and_load_round_trip_keeps_the_strict_mode_flag() -> None:
+    save_mcp_server_statuses([], 'tenant-a', ttl_seconds=60, treat_unreviewed_as_unauthorized=True)
+
+    statuses = load_mcp_server_statuses()
+
+    assert statuses is not None
+    assert statuses.treat_unreviewed_as_unauthorized is True
+    content = json.loads(get_mcp_server_statuses_cache_path().read_text())
+    assert content['should_treat_unreviewed_as_unauthorized'] is True
+
+
+def test_cache_written_before_the_strict_mode_flag_loads_as_not_strict(fs: FakeFilesystem) -> None:
+    content = {'fetched_at': time.time(), 'tenant_id': 'tenant-a', 'ttl_seconds': 60, 'servers': []}
+    fs.create_file(str(get_mcp_server_statuses_cache_path()), contents=json.dumps(content))
+
+    statuses = load_mcp_server_statuses()
+
+    assert statuses is not None
+    assert statuses.treat_unreviewed_as_unauthorized is False
 
 
 def test_expired_cache_needs_refresh() -> None:

tests/cli/commands/ai_guardrails/test_session_start_command.py

@@ -732,7 +732,7 @@ def _guardrail_config_with_mcp_server(agents: dict) -> GuardrailConfig:
 
 
 _SERVERS = [{'alias': 'github', 'normalized_id': 'pkg:gh', 'status': 'Unauthorized'}]
-_STATUSES_RESPONSE = McpServerStatusesResponse(servers=_SERVERS)
+_STATUSES_RESPONSE = McpServerStatusesResponse(servers=_SERVERS, should_treat_unreviewed_as_unauthorized=True)
 
 
 @pytest.mark.parametrize(
@@ -761,8 +761,8 @@ def test_mcp_server_statuses_fetched_when_on_for_some_agent(mock_save_mcp_server
 
     _session_start_mod._sync_mcp_server_statuses(ai_client, config, 'tenant-a', force=False)
 
-    # Cached with the guardrail config's TTL.
-    mock_save_mcp_server_statuses.assert_called_once_with(_SERVERS, 'tenant-a', 600)
+    # Cached with the guardrail config's TTL and the policy's strict-mode flag.
+    mock_save_mcp_server_statuses.assert_called_once_with(_SERVERS, 'tenant-a', 600, True)
 
 
 def test_fresh_mcp_server_statuses_skip_the_fetch_unless_forced(
@@ -824,7 +824,7 @@ def test_session_start_fetches_mcp_server_statuses_after_reporting_the_context(
         session_start_command(mock_ctx, ide='cursor')
 
     assert calls == ['report', 'statuses']
-    mock_save_mcp_server_statuses.assert_called_once_with(_SERVERS, 'tenant-a', 600)
+    mock_save_mcp_server_statuses.assert_called_once_with(_SERVERS, 'tenant-a', 600, True)
 
     # The unchanged context is not re-reported, so the fresh cache is kept.
     with patch('sys.stdin', new=StringIO(json.dumps({'conversation_id': 'conv-2'}))):

tests/cyclient/test_ai_security_manager_client.py

@@ -7,6 +7,8 @@
 from cycode.cli.apps.ai_guardrails.scan.types import AiHookEventType, AIHookOutcome
 from cycode.cyclient.ai_security_manager_client import AISecurityManagerClient
 
+_MISSING = object()
+
 
 def _build_client() -> tuple[AISecurityManagerClient, MagicMock]:
     http_client = MagicMock()
@@ -113,15 +115,29 @@ def test_create_event_without_a_conversation_posts_nothing() -> None:
     http_client.post.assert_not_called()
 
 
-def test_get_mcp_server_statuses_returns_the_rows() -> None:
+@pytest.mark.parametrize(
+    ('flag', 'expected'),
+    [
+        (True, True),
+        (False, False),
+        (None, False),
+        ('true', False),
+        (_MISSING, False),
+    ],
+)
+def test_get_mcp_server_statuses_returns_the_rows_and_the_strict_mode_flag(flag: object, expected: bool) -> None:
     client, http_client = _build_client()
     servers = [{'alias': 'github', 'normalized_id': 'pkg:gh', 'status': 'Unauthorized'}]
-    http_client.get.return_value.json.return_value = {'servers': servers}
+    body: dict = {'servers': servers}
+    if flag is not _MISSING:
+        body['should_treat_unreviewed_as_unauthorized'] = flag
+    http_client.get.return_value.json.return_value = body
 
     response = client.get_mcp_server_statuses()
 
     assert response is not None
     assert response.servers == servers
+    assert response.should_treat_unreviewed_as_unauthorized is expected
     http_client.get.assert_called_once_with('v4/ai-security/authorization/mcp/servers')