diff --git a/README.md b/README.md index 8e71188..2ce8adc 100644 --- a/README.md +++ b/README.md @@ -48,7 +48,7 @@ When you run `scc` or `scc start`: - **Shows Quick Resume** if you have recent sessions for this workspace - **Prints brief context** (workspace root, entry directory, team) before launching - **Applies personal profile** (if saved) after team config, before workspace overrides -- **Bypass mode enabled**: Permission prompts are skipped by default since the Docker sandbox already provides isolation. Press `Shift+Tab` inside Claude to toggle permissions back on if needed +- **Bypass mode enabled**: Permission prompts are skipped by default since the Docker sandbox already provides isolation. This does not prevent access to files inside the mounted workspace. Press `Shift+Tab` inside Claude to toggle permissions back on if needed - **Safety guard**: Won't auto-launch from suspicious directories (home, `/tmp`). Explicit paths like `scc start ~/` prompt for confirmation **Keyboard shortcuts in dashboard:** @@ -110,6 +110,13 @@ Organization security blocks cannot be overridden by teams or developers. *"Approves" = teams can only select from org-allowed marketplaces; blocks always apply. "Extends" = can add plugins/settings, cannot remove org defaults.* +### Enforcement Scope (v1) + +- SCC enforces org-managed plugins and MCP servers at runtime. +- MCP servers in repo `.mcp.json` or plugin bundles are outside SCC enforcement scope (block the plugin to restrict). +- `network_policy` is partially enforced (proxy env injection + MCP suppression under isolated), not a full egress firewall. +- `session.auto_resume` is advisory only in v1. + --- ### Organization Setup diff --git a/examples/04-org-stdio-hardened.json b/examples/04-org-stdio-hardened.json index 8699d7b..6f0570e 100644 --- a/examples/04-org-stdio-hardened.json +++ b/examples/04-org-stdio-hardened.json @@ -35,7 +35,7 @@ { "name": "playwright", "type": "stdio", - "command": "npx", + "command": "/usr/local/bin/npx", "args": ["@playwright/mcp@latest"] } ] @@ -46,7 +46,7 @@ { "name": "playwright", "type": "stdio", - "command": "npx", + "command": "/usr/local/bin/npx", "args": ["@playwright/mcp@latest"] }, { diff --git a/examples/99-complete-reference.json b/examples/99-complete-reference.json index 69cae2b..92cba54 100644 --- a/examples/99-complete-reference.json +++ b/examples/99-complete-reference.json @@ -1,36 +1,87 @@ { "$schema": "../src/scc_cli/schemas/org-v1.schema.json", "schema_version": "1.0.0", - "min_cli_version": "0.5.0", + "min_cli_version": "1.7.0", "organization": { "name": "Reference Organization", "id": "reference-org", "contact": "docs@example.com" }, + "marketplaces": { + "official-plugins": { + "source": "github", + "owner": "CCimen", + "repo": "sandboxed-code-plugins", + "branch": "main", + "path": "/" + }, + "internal-gitlab": { + "source": "git", + "url": "https://gitlab.company.com/scc/plugins.git", + "branch": "main", + "path": "/" + }, + "remote-manifest": { + "source": "url", + "url": "https://plugins.company.com/manifest.json", + "headers": { + "Authorization": "Bearer ${PLUGIN_API_TOKEN}" + }, + "materialization_mode": "self_contained" + }, + "local-dev": { + "source": "directory", + "path": "./local-plugins" + } + }, "security": { "blocked_plugins": [ "pattern-*", "exact-plugin-name", - "*-suffix" + "*-suffix", + "*-experimental", + "*-deprecated" ], "blocked_mcp_servers": [ "*.blocked.com", - "specific-server" + "specific-server", + "*.untrusted.org" ], "allow_stdio_mcp": true, "allowed_stdio_prefixes": [ "/usr/local/bin", "/opt/approved-tools" - ] + ], + "safety_net": { + "action": "block", + "block_force_push": true, + "block_reset_hard": true, + "block_branch_force_delete": true, + "block_checkout_restore": true, + "block_clean": true, + "block_stash_destructive": true + } }, "defaults": { + "enabled_plugins": [ + "scc-safety-net@official-plugins" + ], + "disabled_plugins": [ + "legacy-tool@official-plugins" + ], "allowed_plugins": [ "core-*", - "org-approved-*" + "org-approved-*", + "*@official-plugins" ], "allowed_mcp_servers": [ - "https://mcp.example.com/*" + "https://mcp.example.com/*", + "https://*.company.com/*" + ], + "extra_marketplaces": [ + "official-plugins" ], + "cache_ttl_hours": 24, "network_policy": "unrestricted", "session": { "timeout_hours": 12, @@ -41,12 +92,14 @@ "teams": { "allow_additional_plugins": [ "team-*", - "community-*" + "community-*", + "*" ], "allow_additional_mcp_servers": [ "team-a", "team-b", - "special-team" + "special-team", + "federated-team" ] }, "projects": { @@ -57,8 +110,8 @@ "team-a": { "description": "Example team A with HTTP MCP servers (authenticated and public)", "additional_plugins": [ - "team-plugin-1", - "team-plugin-2" + "team-plugin-1@official-plugins", + "team-plugin-2@official-plugins" ], "additional_mcp_servers": [ { @@ -66,7 +119,7 @@ "type": "http", "url": "https://mcp.context7.com/mcp", "headers": { - "CONTEXT7_API_KEY": "${CONTEXT7_API_KEY}" + "X-API-Key": "${CONTEXT7_API_KEY}" } }, { @@ -101,26 +154,51 @@ } }, "special-team": { - "description": "Team with stdio MCP server (requires allow_stdio_mcp: true)", + "description": "Team with stdio MCP server (requires allow_stdio_mcp: true in security)", "additional_mcp_servers": [ { "name": "playwright", "type": "stdio", - "command": "npx", + "command": "/usr/local/bin/npx", "args": [ "@playwright/mcp@latest" ], "env": { - "PLAYWRIGHT_BROWSERS_PATH": "${HOME}/.cache/ms-playwright" + "PLAYWRIGHT_BROWSERS_PATH": "${HOME}/.cache/ms-playwright", + "DEBUG": "pw:api" } } ], "network_policy": "isolated" + }, + "federated-team": { + "description": "Federated team with external config source (team manages own config)", + "config_source": { + "source": "github", + "owner": "company", + "repo": "team-config", + "branch": "main", + "path": "scc/team-config.json", + "headers": { + "Authorization": "Bearer ${GITHUB_TOKEN}" + } + }, + "trust": { + "inherit_org_marketplaces": true, + "allow_additional_marketplaces": true, + "marketplace_source_patterns": [ + "https://github.com/company/*", + "https://gitlab.company.com/*" + ] + }, + "delegation": { + "allow_project_overrides": true + } } }, "stats": { "enabled": true, "user_identity_mode": "hash", - "retention_days": 365 + "retention_days": 90 } } diff --git a/src/scc_cli/adapters/docker_sandbox_runtime.py b/src/scc_cli/adapters/docker_sandbox_runtime.py index fc2f85c..ab3d814 100644 --- a/src/scc_cli/adapters/docker_sandbox_runtime.py +++ b/src/scc_cli/adapters/docker_sandbox_runtime.py @@ -3,6 +3,8 @@ from __future__ import annotations from scc_cli import docker +from scc_cli.core.enums import NetworkPolicy +from scc_cli.core.network_policy import collect_proxy_env from scc_cli.ports.models import SandboxHandle, SandboxSpec, SandboxState, SandboxStatus from scc_cli.ports.sandbox_runtime import SandboxRuntime @@ -26,13 +28,18 @@ def ensure_available(self) -> None: def run(self, spec: SandboxSpec) -> SandboxHandle: docker.prepare_sandbox_volume_for_credentials() + env_vars = dict(spec.env) if spec.env else {} + if spec.network_policy == NetworkPolicy.CORP_PROXY_ONLY.value: + for key, value in collect_proxy_env().items(): + env_vars.setdefault(key, value) + runtime_env = env_vars or None docker_cmd, _is_resume = docker.get_or_create_container( workspace=spec.workspace_mount.source, branch=None, profile=None, force_new=spec.force_new, continue_session=spec.continue_session, - env_vars=spec.env or None, + env_vars=runtime_env, ) container_name = _extract_container_name(docker_cmd) plugin_settings = spec.agent_settings.content if spec.agent_settings else None @@ -41,6 +48,7 @@ def run(self, spec: SandboxSpec) -> SandboxHandle: org_config=spec.org_config, container_workdir=spec.workdir, plugin_settings=plugin_settings, + env_vars=runtime_env, ) return SandboxHandle( sandbox_id=container_name or "sandbox", diff --git a/src/scc_cli/application/compute_effective_config.py b/src/scc_cli/application/compute_effective_config.py index 9380f62..a079eca 100644 --- a/src/scc_cli/application/compute_effective_config.py +++ b/src/scc_cli/application/compute_effective_config.py @@ -9,7 +9,8 @@ from urllib.parse import urlparse from scc_cli import config as config_module -from scc_cli.core.enums import MCPServerType, RequestSource, TargetType +from scc_cli.core.enums import MCPServerType, NetworkPolicy, RequestSource, TargetType +from scc_cli.core.network_policy import is_more_or_equal_restrictive if TYPE_CHECKING: pass @@ -135,13 +136,36 @@ def matches_blocked(item: str, blocked_patterns: list[str]) -> str | None: return None -def is_allowed(item: str, allowed_patterns: list[str] | None) -> bool: - """Check whether item is allowed by an optional allowlist.""" +def matches_plugin_pattern(plugin_ref: str, pattern: str) -> bool: + """Check plugin patterns, allowing bare names to match any marketplace.""" + if not plugin_ref or not pattern: + return False + normalized_ref = plugin_ref.strip().casefold() + normalized_pattern = pattern.strip().casefold() + if "@" not in normalized_pattern and "@" in normalized_ref: + plugin_name = normalized_ref.split("@", 1)[0] + return fnmatch(plugin_name, normalized_pattern) + return fnmatch(normalized_ref, normalized_pattern) + + +def matches_blocked_plugin(plugin_ref: str, blocked_patterns: list[str]) -> str | None: + """Return the matching pattern for a blocked plugin, if any.""" + for pattern in blocked_patterns: + if matches_plugin_pattern(plugin_ref, pattern): + return pattern + return None + + +def is_plugin_allowed(plugin_ref: str, allowed_patterns: list[str] | None) -> bool: + """Check whether plugin is allowed by an optional allowlist.""" if allowed_patterns is None: return True if not allowed_patterns: return False - return matches_blocked(item, allowed_patterns) is not None + for pattern in allowed_patterns: + if matches_plugin_pattern(plugin_ref, pattern): + return True + return False def mcp_candidates(server: dict[str, Any]) -> list[str]: @@ -174,6 +198,39 @@ def is_mcp_allowed(server: dict[str, Any], allowed_patterns: list[str] | None) - return False +def match_blocked_mcp(server: dict[str, Any], blocked_patterns: list[str]) -> str | None: + """Return the matching pattern for a blocked MCP server, if any.""" + for candidate in mcp_candidates(server): + matched = matches_blocked(candidate, blocked_patterns) + if matched: + return matched + return None + + +def is_network_mcp(server: dict[str, Any]) -> bool: + """Return True for MCP transports that require network access.""" + return server.get("type") in {MCPServerType.SSE, MCPServerType.HTTP} + + +def record_network_policy_decision( + result: EffectiveConfig, + *, + policy: str, + reason: str, + source: str, +) -> None: + """Record the active network_policy decision (replace any prior entries).""" + result.decisions = [d for d in result.decisions if d.field != "network_policy"] + result.decisions.append( + ConfigDecision( + field="network_policy", + value=policy, + reason=reason, + source=source, + ) + ) + + def validate_stdio_server( server: dict[str, Any], org_config: dict[str, Any], @@ -340,14 +397,14 @@ def compute_effective_config( default_session = defaults.get("session", {}) for plugin in default_plugins: - blocked_by = matches_blocked(plugin, blocked_plugins) + blocked_by = matches_blocked_plugin(plugin, blocked_plugins) if blocked_by: result.blocked_items.append( BlockedItem(item=plugin, blocked_by=blocked_by, source="org.security") ) continue - if matches_blocked(plugin, disabled_plugins): + if matches_blocked_plugin(plugin, disabled_plugins): continue result.plugins.add(plugin) @@ -360,15 +417,15 @@ def compute_effective_config( ) ) + network_policy_source: str | None = None if default_network_policy: result.network_policy = default_network_policy - result.decisions.append( - ConfigDecision( - field="network_policy", - value=default_network_policy, - reason="Organization default network policy", - source="org.defaults", - ) + network_policy_source = "org.defaults" + record_network_policy_decision( + result, + policy=default_network_policy, + reason="Organization default network policy", + source="org.defaults", ) if default_session.get("timeout_hours") is not None: @@ -387,11 +444,32 @@ def compute_effective_config( profiles = org_config.get("profiles", {}) team_config = profiles.get(team_name, {}) + team_network_policy = team_config.get("network_policy") + if team_network_policy: + if result.network_policy is None: + result.network_policy = team_network_policy + network_policy_source = f"team.{team_name}" + record_network_policy_decision( + result, + policy=team_network_policy, + reason=f"Overridden by team profile '{team_name}'", + source=f"team.{team_name}", + ) + elif is_more_or_equal_restrictive(team_network_policy, result.network_policy): + result.network_policy = team_network_policy + network_policy_source = f"team.{team_name}" + record_network_policy_decision( + result, + policy=team_network_policy, + reason=f"Overridden by team profile '{team_name}'", + source=f"team.{team_name}", + ) + team_plugins = team_config.get("additional_plugins", []) team_delegated_plugins = is_team_delegated_for_plugins(org_config, team_name) for plugin in team_plugins: - blocked_by = matches_blocked(plugin, blocked_plugins) + blocked_by = matches_blocked_plugin(plugin, blocked_plugins) if blocked_by: result.blocked_items.append( BlockedItem(item=plugin, blocked_by=blocked_by, source="org.security") @@ -408,7 +486,7 @@ def compute_effective_config( ) continue - if not is_allowed(plugin, allowed_plugins): + if not is_plugin_allowed(plugin, allowed_plugins): result.denied_additions.append( DelegationDenied( item=plugin, @@ -435,10 +513,7 @@ def compute_effective_config( server_name = server_dict.get("name", "") server_url = server_dict.get("url", "") - blocked_by = matches_blocked(server_name, blocked_mcp_servers) - if not blocked_by and server_url: - domain = _extract_domain(server_url) - blocked_by = matches_blocked(domain, blocked_mcp_servers) + blocked_by = match_blocked_mcp(server_dict, blocked_mcp_servers) if blocked_by: result.blocked_items.append( @@ -473,6 +548,17 @@ def compute_effective_config( ) continue + if result.network_policy == NetworkPolicy.ISOLATED.value and is_network_mcp(server_dict): + result.blocked_items.append( + BlockedItem( + item=server_name or server_url, + blocked_by="network_policy=isolated", + source=network_policy_source or "org.defaults", + target_type=TargetType.MCP_SERVER, + ) + ) + continue + if server_dict.get("type") == MCPServerType.STDIO: stdio_result = validate_stdio_server(server_dict, org_config) if stdio_result.blocked: @@ -492,6 +578,8 @@ def compute_effective_config( url=server_url or None, command=server_dict.get("command"), args=server_dict.get("args"), + env=server_dict.get("env"), + headers=server_dict.get("headers"), ) result.mcp_servers.append(mcp_server) result.decisions.append( @@ -520,7 +608,7 @@ def compute_effective_config( project_plugins = project_config.get("additional_plugins", []) for plugin in project_plugins: - blocked_by = matches_blocked(plugin, blocked_plugins) + blocked_by = matches_blocked_plugin(plugin, blocked_plugins) if blocked_by: result.blocked_items.append( BlockedItem(item=plugin, blocked_by=blocked_by, source="org.security") @@ -537,7 +625,7 @@ def compute_effective_config( ) continue - if not is_allowed(plugin, allowed_plugins): + if not is_plugin_allowed(plugin, allowed_plugins): result.denied_additions.append( DelegationDenied( item=plugin, @@ -562,10 +650,7 @@ def compute_effective_config( server_name = server_dict.get("name", "") server_url = server_dict.get("url", "") - blocked_by = matches_blocked(server_name, blocked_mcp_servers) - if not blocked_by and server_url: - domain = _extract_domain(server_url) - blocked_by = matches_blocked(domain, blocked_mcp_servers) + blocked_by = match_blocked_mcp(server_dict, blocked_mcp_servers) if blocked_by: result.blocked_items.append( @@ -600,6 +685,19 @@ def compute_effective_config( ) continue + if result.network_policy == NetworkPolicy.ISOLATED.value and is_network_mcp( + server_dict + ): + result.blocked_items.append( + BlockedItem( + item=server_name or server_url, + blocked_by="network_policy=isolated", + source=network_policy_source or "org.defaults", + target_type=TargetType.MCP_SERVER, + ) + ) + continue + if server_dict.get("type") == MCPServerType.STDIO: stdio_result = validate_stdio_server(server_dict, org_config) if stdio_result.blocked: @@ -619,6 +717,8 @@ def compute_effective_config( url=server_url or None, command=server_dict.get("command"), args=server_dict.get("args"), + env=server_dict.get("env"), + headers=server_dict.get("headers"), ) result.mcp_servers.append(mcp_server) result.decisions.append( diff --git a/src/scc_cli/application/start_session.py b/src/scc_cli/application/start_session.py index 72d9573..83ffc87 100644 --- a/src/scc_cli/application/start_session.py +++ b/src/scc_cli/application/start_session.py @@ -92,7 +92,11 @@ def prepare_start_session( resolver_result = _resolve_workspace_context(request) effective_config = _compute_effective_config(request) sync_result, sync_error_message = sync_marketplace_settings_for_start(request, dependencies) - agent_settings = _build_agent_settings(sync_result, dependencies.agent_runner) + agent_settings = _build_agent_settings( + sync_result, + dependencies.agent_runner, + effective_config=effective_config, + ) current_branch = _resolve_current_branch(request.workspace_path, dependencies.git_client) sandbox_spec = _build_sandbox_spec( request=request, @@ -196,11 +200,23 @@ def sync_marketplace_settings_for_start( def _build_agent_settings( sync_result: SyncResult | None, agent_runner: AgentRunner, + *, + effective_config: EffectiveConfig | None, ) -> AgentSettings | None: - if not sync_result or not sync_result.rendered_settings: + settings: dict[str, Any] | None = None + if sync_result and sync_result.rendered_settings: + settings = dict(sync_result.rendered_settings) + + if effective_config: + from scc_cli.claude_adapter import merge_mcp_servers + + settings = merge_mcp_servers(settings, effective_config) + + if not settings: return None + settings_path = Path("/home/agent") / AGENT_CONFIG_DIR / "settings.json" - return agent_runner.build_settings(sync_result.rendered_settings, path=settings_path) + return agent_runner.build_settings(settings, path=settings_path) def _resolve_current_branch(workspace_path: Path, git_client: GitClient) -> str | None: diff --git a/src/scc_cli/claude_adapter.py b/src/scc_cli/claude_adapter.py index 6bff40f..a761211 100644 --- a/src/scc_cli/claude_adapter.py +++ b/src/scc_cli/claude_adapter.py @@ -444,6 +444,27 @@ def build_mcp_servers(effective_config: EffectiveConfig) -> dict[str, Any]: return mcp_servers +def merge_mcp_servers( + settings: dict[str, Any] | None, + effective_config: EffectiveConfig | None, +) -> dict[str, Any] | None: + """Merge MCP servers into an existing settings dict.""" + if effective_config is None: + return settings + + mcp_servers = build_mcp_servers(effective_config) + if not mcp_servers: + return settings + + merged: dict[str, Any] = dict(settings) if settings else {} + existing = merged.get("mcpServers") + if isinstance(existing, dict): + merged["mcpServers"] = {**existing, **mcp_servers} + else: + merged["mcpServers"] = mcp_servers + return merged + + # ═══════════════════════════════════════════════════════════════════════════════ # Credential Injection # ═══════════════════════════════════════════════════════════════════════════════ diff --git a/src/scc_cli/commands/config.py b/src/scc_cli/commands/config.py index 2ef568b..d1e8250 100644 --- a/src/scc_cli/commands/config.py +++ b/src/scc_cli/commands/config.py @@ -1,8 +1,9 @@ """Provide CLI commands for managing teams, configuration, and setup.""" import json +from dataclasses import dataclass from pathlib import Path -from typing import Annotated +from typing import Annotated, Any import typer from rich import box @@ -18,7 +19,9 @@ ) from ..cli_common import console, handle_errors from ..core import personal_profiles +from ..core.enums import NetworkPolicy from ..core.exit_codes import EXIT_USAGE +from ..core.network_policy import collect_proxy_env, is_more_or_equal_restrictive from ..maintenance import get_paths, get_total_size from ..panels import create_error_panel, create_info_panel from ..source_resolver import ResolveError, resolve_source @@ -37,6 +40,15 @@ ) +@dataclass(frozen=True) +class EnforcementStatusEntry: + """Describe runtime enforcement status for a config surface.""" + + surface: str + status: str + detail: str + + # ───────────────────────────────────────────────────────────────────────────── # Setup Command # ───────────────────────────────────────────────────────────────────────────── @@ -188,7 +200,17 @@ def config_cmd( return if action == "explain": - _config_explain(field_filter=field, workspace_path=workspace) + if json_output: + from ..output_mode import json_command_mode, json_output_mode + + with json_output_mode(), json_command_mode(): + _config_explain( + field_filter=field, + workspace_path=workspace, + json_output=True, + ) + else: + _config_explain(field_filter=field, workspace_path=workspace, json_output=False) return # Handle --show and --edit flags @@ -263,7 +285,11 @@ def _config_get(key: str) -> None: console.print(str(obj)) -def _config_explain(field_filter: str | None = None, workspace_path: str | None = None) -> None: +def _config_explain( + field_filter: str | None = None, + workspace_path: str | None = None, + json_output: bool = False, +) -> None: """Explain the effective configuration with source attribution. Shows: @@ -293,6 +319,34 @@ def _config_explain(field_filter: str | None = None, workspace_path: str | None workspace_path=ws_path, ) + enforcement_status = _build_enforcement_status_entries() + enforcement_payload = _serialize_enforcement_status_entries(enforcement_status) + warnings = _collect_advisory_warnings( + org_config=org_config, + team_name=team, + workspace_path=ws_path, + effective_network_policy=effective.network_policy, + ) + + if json_output: + from ..output_mode import print_json + from ..presentation.json.config_json import ( + build_config_explain_data, + build_config_explain_envelope, + ) + + data = build_config_explain_data( + org_config=org_config, + team_name=team, + effective=effective, + enforcement_status=enforcement_payload, + warnings=warnings, + workspace_path=ws_path, + ) + envelope = build_config_explain_envelope(data, warnings=warnings) + print_json(envelope) + return + # Build output console.print( create_info_panel( @@ -303,6 +357,8 @@ def _config_explain(field_filter: str | None = None, workspace_path: str | None ) console.print() + _render_enforcement_status(enforcement_status, field_filter) + # Show decisions (config values with source attribution) _render_config_decisions(effective, field_filter) @@ -327,6 +383,138 @@ def _config_explain(field_filter: str | None = None, workspace_path: str | None ) console.print() + _render_advisory_warnings(warnings, field_filter) + + +def _build_enforcement_status_entries() -> list[EnforcementStatusEntry]: + return [ + EnforcementStatusEntry( + surface="Plugins", + status="Enforced", + detail="SCC-managed plugins are injected into runtime settings.", + ), + EnforcementStatusEntry( + surface="Marketplaces", + status="Enforced", + detail="Managed marketplaces are materialized and injected.", + ), + EnforcementStatusEntry( + surface="MCP servers (org/team/project)", + status="Enforced", + detail="SCC-managed MCP servers are injected after policy gates.", + ), + EnforcementStatusEntry( + surface="MCP servers (.mcp.json)", + status="Advisory", + detail="SCC does not modify repo MCP files in v1.", + ), + EnforcementStatusEntry( + surface="MCP servers (plugin-bundled)", + status="Out of scope", + detail="Plugins are the trust unit; block the plugin to restrict.", + ), + EnforcementStatusEntry( + surface="network_policy", + status="Partially enforced", + detail="Proxy env injection and MCP suppression, not full egress control.", + ), + EnforcementStatusEntry( + surface="safety_net policy", + status="Enforced when enabled", + detail="Policy is enforced by the scc-safety-net plugin.", + ), + EnforcementStatusEntry( + surface="session.auto_resume", + status="Advisory", + detail="Accepted in config but not enforced yet.", + ), + ] + + +def _render_enforcement_status( + entries: list[EnforcementStatusEntry], field_filter: str | None +) -> None: + if field_filter and field_filter not in {"enforcement", "enforcement_status"}: + return + + console.print("[bold cyan]Enforcement Status[/bold cyan]") + for entry in entries: + console.print(f" {entry.surface}: {entry.status}") + console.print(f" [dim]{entry.detail}[/dim]") + console.print() + + +def _serialize_enforcement_status_entries( + entries: list[EnforcementStatusEntry], +) -> list[dict[str, str]]: + return [ + {"surface": entry.surface, "status": entry.status, "detail": entry.detail} + for entry in entries + ] + + +def _collect_advisory_warnings( + *, + org_config: dict[str, Any], + team_name: str, + workspace_path: Path, + effective_network_policy: str | None, +) -> list[str]: + warnings: list[str] = [] + + defaults_session = org_config.get("defaults", {}).get("session", {}) + team_session = org_config.get("profiles", {}).get(team_name, {}).get("session", {}) + project_config = config.read_project_config(workspace_path) or {} + project_session = project_config.get("session", {}) + + auto_resume_sources: list[str] = [] + if "auto_resume" in defaults_session: + auto_resume_sources.append("org.defaults") + if "auto_resume" in team_session: + auto_resume_sources.append(f"team.{team_name}") + if "auto_resume" in project_session: + auto_resume_sources.append("project") + + if auto_resume_sources: + sources = ", ".join(auto_resume_sources) + warnings.append( + f"session.auto_resume is advisory only and not enforced (set by {sources})." + ) + + default_network_policy = org_config.get("defaults", {}).get("network_policy") + team_network_policy = org_config.get("profiles", {}).get(team_name, {}).get("network_policy") + if ( + default_network_policy + and team_network_policy + and not is_more_or_equal_restrictive(team_network_policy, default_network_policy) + ): + warnings.append( + "team network_policy is less restrictive than org default and is ignored " + f"({team_network_policy} < {default_network_policy})." + ) + + if effective_network_policy == NetworkPolicy.CORP_PROXY_ONLY.value: + proxy_env = collect_proxy_env() + if not proxy_env: + warnings.append( + "network_policy is corp-proxy-only but no proxy env vars are set " + "(HTTP_PROXY/HTTPS_PROXY/NO_PROXY)." + ) + + return warnings + + +def _render_advisory_warnings(warnings: list[str], field_filter: str | None) -> None: + if not warnings: + return + if field_filter and field_filter not in {"warnings", "enforcement"}: + return + + console.print("[bold yellow]Warnings[/bold yellow]") + for warning in warnings: + console.print(f" [yellow]⚠[/yellow] {warning}") + console.print() + def _render_config_decisions(effective: EffectiveConfig, field_filter: str | None) -> None: """Render config decisions grouped by field.""" diff --git a/src/scc_cli/commands/launch/sandbox.py b/src/scc_cli/commands/launch/sandbox.py index 4f22860..1be8240 100644 --- a/src/scc_cli/commands/launch/sandbox.py +++ b/src/scc_cli/commands/launch/sandbox.py @@ -55,6 +55,22 @@ def launch_sandbox( # Load org config for safety-net policy injection # This is already cached by _configure_team_settings(), so it's a fast read org_config = config.load_cached_org_config() + env_vars = None + + if org_config and team: + from ...application.compute_effective_config import compute_effective_config + from ...claude_adapter import merge_mcp_servers + from ...core.enums import NetworkPolicy + from ...core.network_policy import collect_proxy_env + + effective_config = compute_effective_config( + org_config=org_config, + team_name=team, + workspace_path=workspace_path or mount_path, + ) + plugin_settings = merge_mcp_servers(plugin_settings, effective_config) + if effective_config.network_policy == NetworkPolicy.CORP_PROXY_ONLY.value: + env_vars = collect_proxy_env() # Prepare sandbox volume for credential persistence docker.prepare_sandbox_volume_for_credentials() @@ -138,6 +154,7 @@ def launch_sandbox( org_config=org_config, container_workdir=workspace_path, plugin_settings=plugin_settings, + env_vars=env_vars, ) diff --git a/src/scc_cli/core/network_policy.py b/src/scc_cli/core/network_policy.py new file mode 100644 index 0000000..b910dc0 --- /dev/null +++ b/src/scc_cli/core/network_policy.py @@ -0,0 +1,38 @@ +"""Network policy helpers for ordering and comparison.""" + +from __future__ import annotations + +from collections.abc import Mapping + +from .enums import NetworkPolicy + +_NETWORK_POLICY_ORDER = { + NetworkPolicy.UNRESTRICTED.value: 0, + NetworkPolicy.CORP_PROXY_ONLY.value: 1, + NetworkPolicy.ISOLATED.value: 2, +} + + +def policy_rank(policy: str | None) -> int: + """Return numeric rank for policy (higher = more restrictive).""" + if policy is None: + return -1 + return _NETWORK_POLICY_ORDER.get(policy, -1) + + +def is_more_or_equal_restrictive(candidate: str, baseline: str) -> bool: + """Return True if candidate is as or more restrictive than baseline.""" + return policy_rank(candidate) >= policy_rank(baseline) + + +def collect_proxy_env(env: Mapping[str, str] | None = None) -> dict[str, str]: + """Collect proxy environment variables for container injection.""" + import os + + source = env or os.environ + proxy_env: dict[str, str] = {} + for key in ("HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY"): + value = source.get(key) or source.get(key.lower()) + if value: + proxy_env[key] = value + return proxy_env diff --git a/src/scc_cli/docker/core.py b/src/scc_cli/docker/core.py index a254051..29998a6 100644 --- a/src/scc_cli/docker/core.py +++ b/src/scc_cli/docker/core.py @@ -224,6 +224,7 @@ def build_command( resume: bool = False, detached: bool = False, policy_host_path: Path | None = None, + env_vars: dict[str, str] | None = None, ) -> list[str]: """ Build the docker sandbox run command. @@ -238,6 +239,7 @@ def build_command( policy_host_path: Host path to safety net policy file to bind-mount read-only. If provided, mounts at /mnt/claude-data/effective_policy.json:ro and sets SCC_POLICY_PATH env var for the plugin. + env_vars: Environment variables to inject into the sandbox runtime. Returns: Command as list of strings @@ -275,6 +277,11 @@ def build_command( # Set SCC_POLICY_PATH env var so plugin knows where to read policy cmd.extend(["-e", f"SCC_POLICY_PATH={container_policy_path}"]) + if env_vars: + for key, value in sorted(env_vars.items()): + if value: + cmd.extend(["-e", f"{key}={value}"]) + # Add workspace mount if workspace: cmd.extend(["-w", str(workspace)]) diff --git a/src/scc_cli/docker/launch.py b/src/scc_cli/docker/launch.py index a028022..e1598b7 100644 --- a/src/scc_cli/docker/launch.py +++ b/src/scc_cli/docker/launch.py @@ -247,6 +247,7 @@ def run( org_config: dict[str, Any] | None = None, container_workdir: Path | None = None, plugin_settings: dict[str, Any] | None = None, + env_vars: dict[str, str] | None = None, ) -> int: """ Execute the Docker command with optional org configuration. @@ -267,6 +268,7 @@ def run( plugin_settings: Plugin settings dict to inject into container HOME. Contains extraKnownMarketplaces and enabledPlugins. Injected to /home/agent/.claude/settings.json to prevent host leakage. + env_vars: Environment variables to set for the sandbox runtime. Raises: SandboxLaunchError: If Docker command fails to start @@ -294,6 +296,7 @@ def run( org_config=org_config, container_workdir=container_workdir, plugin_settings=plugin_settings, + env_vars=env_vars, ) @@ -305,6 +308,7 @@ def run_sandbox( org_config: dict[str, Any] | None = None, container_workdir: Path | None = None, plugin_settings: dict[str, Any] | None = None, + env_vars: dict[str, str] | None = None, ) -> int: """ Run Claude in a Docker sandbox with credential persistence. @@ -335,6 +339,7 @@ def run_sandbox( plugin_settings: Plugin settings dict to inject into container HOME. Contains extraKnownMarketplaces and enabledPlugins. Injected to /home/agent/.claude/settings.json to prevent host leakage. + env_vars: Environment variables to set for the sandbox runtime. Returns: Exit code from Docker process @@ -376,6 +381,7 @@ def run_sandbox( workspace=workspace, detached=True, policy_host_path=policy_host_path, + env_vars=env_vars, ) max_retries = 5 @@ -478,6 +484,7 @@ def run_sandbox( resume=resume, detached=False, policy_host_path=policy_host_path, + env_vars=env_vars, ) if os.name != "nt": @@ -774,7 +781,7 @@ def get_or_create_container( profile: Team profile (unused - sandboxes don't support labels) force_new: Force new container (unused - sandboxes are always new) continue_session: Pass -c flag to Claude - env_vars: Environment variables (unused - sandboxes handle auth) + env_vars: Environment variables to set for the sandbox runtime Returns: Tuple of (command_to_run, is_resume) @@ -784,5 +791,6 @@ def get_or_create_container( cmd = build_command( workspace=workspace, continue_session=continue_session, + env_vars=env_vars, ) return cmd, False diff --git a/src/scc_cli/marketplace/normalize.py b/src/scc_cli/marketplace/normalize.py index 32e50e2..83a9944 100644 --- a/src/scc_cli/marketplace/normalize.py +++ b/src/scc_cli/marketplace/normalize.py @@ -255,7 +255,12 @@ def matches_pattern(plugin_ref: str, pattern: str) -> bool: # Use fnmatch for glob-style matching with case-insensitive comparison # Documentation requires Unicode-aware casefolding for security patterns # to prevent bypass attempts using case variations (e.g., "MALICIOUS-*" vs "malicious-*") - return fnmatch.fnmatch(plugin_ref.casefold(), pattern.casefold()) + normalized_ref = plugin_ref.casefold() + normalized_pattern = pattern.casefold() + if "@" not in normalized_pattern and "@" in normalized_ref: + plugin_name = normalized_ref.split("@", 1)[0] + return fnmatch.fnmatch(plugin_name, normalized_pattern) + return fnmatch.fnmatch(normalized_ref, normalized_pattern) def matches_any_pattern(plugin_ref: str, patterns: list[str]) -> str | None: diff --git a/src/scc_cli/presentation/json/config_json.py b/src/scc_cli/presentation/json/config_json.py new file mode 100644 index 0000000..a54d848 --- /dev/null +++ b/src/scc_cli/presentation/json/config_json.py @@ -0,0 +1,119 @@ +"""JSON mapping helpers for config explain output.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from ...application.compute_effective_config import ( + BlockedItem, + ConfigDecision, + DelegationDenied, + EffectiveConfig, + MCPServer, +) +from ...core import personal_profiles +from ...json_output import build_envelope +from ...kinds import Kind + + +def build_config_explain_data( + *, + org_config: dict[str, Any], + team_name: str, + effective: EffectiveConfig, + enforcement_status: list[dict[str, str]], + warnings: list[str], + workspace_path: Path, +) -> dict[str, Any]: + """Build JSON-ready data for config explain output.""" + org_info = org_config.get("organization", {}) + profile = personal_profiles.load_personal_profile(workspace_path) + + return { + "organization": { + "name": org_info.get("name", "Unknown"), + "id": org_info.get("id", ""), + }, + "team": team_name, + "enforcement": enforcement_status, + "warnings": warnings, + "effective": { + "plugins": sorted(effective.plugins), + "mcp_servers": [_serialize_mcp_server(server) for server in effective.mcp_servers], + "network_policy": effective.network_policy, + "session": { + "timeout_hours": effective.session_config.timeout_hours, + "auto_resume": effective.session_config.auto_resume, + }, + }, + "decisions": [_serialize_decision(decision) for decision in effective.decisions], + "blocked_items": [_serialize_blocked_item(item) for item in effective.blocked_items], + "denied_additions": [ + _serialize_denied_addition(denied) for denied in effective.denied_additions + ], + "personal_profile": _serialize_personal_profile(profile), + } + + +def build_config_explain_envelope( + data: dict[str, Any], *, warnings: list[str] | None = None +) -> dict[str, Any]: + """Build the JSON envelope for config explain output.""" + return build_envelope(Kind.CONFIG_EXPLAIN, data=data, warnings=warnings or []) + + +def _serialize_mcp_server(server: MCPServer) -> dict[str, Any]: + payload: dict[str, Any] = {"name": server.name, "type": server.type} + if server.url: + payload["url"] = server.url + if server.command: + payload["command"] = server.command + if server.args: + payload["args"] = server.args + if server.env: + payload["env"] = server.env + if server.headers: + payload["headers"] = server.headers + return payload + + +def _serialize_decision(decision: ConfigDecision) -> dict[str, Any]: + return { + "field": decision.field, + "value": decision.value, + "reason": decision.reason, + "source": decision.source, + } + + +def _serialize_blocked_item(item: BlockedItem) -> dict[str, Any]: + return { + "item": item.item, + "blocked_by": item.blocked_by, + "source": item.source, + "target_type": item.target_type, + } + + +def _serialize_denied_addition(denied: DelegationDenied) -> dict[str, Any]: + return { + "item": denied.item, + "requested_by": denied.requested_by, + "reason": denied.reason, + "target_type": denied.target_type, + } + + +def _serialize_personal_profile( + profile: personal_profiles.PersonalProfile | None, +) -> dict[str, Any]: + if profile is None: + return {} + + plugins = personal_profiles.extract_personal_plugins(profile) + return { + "repo": profile.repo_id, + "plugins": sorted(plugins), + "mcp": bool(profile.mcp), + } diff --git a/src/scc_cli/profiles.py b/src/scc_cli/profiles.py index dbc5be5..61ed511 100644 --- a/src/scc_cli/profiles.py +++ b/src/scc_cli/profiles.py @@ -26,7 +26,7 @@ SessionConfig = compute_effective_config_module.SessionConfig StdioValidationResult = compute_effective_config_module.StdioValidationResult compute_effective_config = compute_effective_config_module.compute_effective_config -is_allowed = compute_effective_config_module.is_allowed +is_allowed = compute_effective_config_module.is_plugin_allowed is_mcp_allowed = compute_effective_config_module.is_mcp_allowed is_project_delegated = compute_effective_config_module.is_project_delegated is_team_delegated_for_mcp = compute_effective_config_module.is_team_delegated_for_mcp diff --git a/tests/test_application_start_session.py b/tests/test_application_start_session.py index 8e899db..c8c7102 100644 --- a/tests/test_application_start_session.py +++ b/tests/test_application_start_session.py @@ -3,6 +3,7 @@ from pathlib import Path from unittest.mock import MagicMock, patch +from scc_cli.application.compute_effective_config import EffectiveConfig, MCPServer from scc_cli.application.start_session import ( StartSessionDependencies, StartSessionPlan, @@ -166,6 +167,57 @@ def test_prepare_start_session_captures_sync_error(tmp_path: Path) -> None: assert plan.sandbox_spec is not None +def test_prepare_start_session_injects_mcp_servers(tmp_path: Path) -> None: + workspace_path = tmp_path / "workspace" + workspace_path.mkdir() + request = StartSessionRequest( + workspace_path=workspace_path, + workspace_arg=str(workspace_path), + entry_dir=workspace_path, + team="alpha", + session_name="session-1", + resume=False, + fresh=False, + offline=False, + standalone=False, + dry_run=False, + allow_suspicious=False, + org_config={ + "defaults": {}, + "profiles": {"alpha": {}}, + }, + ) + sync_result = SyncResult( + success=True, + rendered_settings={"enabledPlugins": {"tool@market": True}}, + ) + resolver_result = _build_resolver_result(workspace_path) + dependencies = _build_dependencies(FakeGitClient(branch="main")) + effective_config = EffectiveConfig( + mcp_servers=[MCPServer(name="gis-internal", type="sse", url="https://gis.example.com/mcp")] + ) + + with ( + patch( + "scc_cli.application.start_session.resolve_workspace", + return_value=WorkspaceContext(resolver_result), + ), + patch( + "scc_cli.application.start_session.compute_effective_config", + return_value=effective_config, + ), + patch( + "scc_cli.application.start_session.sync_marketplace_settings", + return_value=sync_result, + ), + ): + plan = prepare_start_session(request, dependencies=dependencies) + + assert plan.agent_settings is not None + assert "mcpServers" in plan.agent_settings.content + assert "gis-internal" in plan.agent_settings.content["mcpServers"] + + def test_start_session_runs_sandbox_runtime(tmp_path: Path) -> None: workspace_path = tmp_path / "workspace" workspace_path.mkdir() diff --git a/tests/test_claude_adapter.py b/tests/test_claude_adapter.py index 859a022..ffe6fdb 100644 --- a/tests/test_claude_adapter.py +++ b/tests/test_claude_adapter.py @@ -13,6 +13,7 @@ import pytest from scc_cli import claude_adapter +from scc_cli.application.compute_effective_config import EffectiveConfig, MCPServer from scc_cli.claude_adapter import AuthResult # ═══════════════════════════════════════════════════════════════════════════════ @@ -622,6 +623,43 @@ def test_settings_is_opaque_dict(self, sample_profile, gitlab_marketplace): # docker.py just needs to know it's a dict to inject assert isinstance(settings, dict) + +# ═══════════════════════════════════════════════════════════════════════════════ +# Tests for merge_mcp_servers +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestMergeMcpServers: + """Tests for merge_mcp_servers function.""" + + def test_merge_preserves_existing_servers(self): + """Should merge without dropping existing mcpServers or other settings.""" + settings = { + "mcpServers": { + "existing": { + "type": "http", + "url": "https://existing.example.com/mcp", + } + }, + "otherSetting": True, + } + effective_config = EffectiveConfig( + mcp_servers=[ + MCPServer( + name="new", + type="http", + url="https://new.example.com/mcp", + ) + ] + ) + + merged = claude_adapter.merge_mcp_servers(settings, effective_config) + + assert merged is not None + assert merged["otherSetting"] is True + assert "existing" in merged["mcpServers"] + assert "new" in merged["mcpServers"] + def test_env_vars_are_simple_dict(self, gitlab_marketplace): """inject_credentials produces simple str->str dict.""" docker_env = {} diff --git a/tests/test_config_explain.py b/tests/test_config_explain.py index 24a98f6..4cdc8e9 100644 --- a/tests/test_config_explain.py +++ b/tests/test_config_explain.py @@ -11,6 +11,7 @@ from __future__ import annotations +import json from pathlib import Path from unittest.mock import patch @@ -209,6 +210,24 @@ def test_explain_shows_effective_plugins(self, effective_config_basic, mock_org_ assert "plugin-a" in result.output assert "plugin-b" in result.output + def test_explain_shows_enforcement_status(self, effective_config_basic, mock_org_config): + """Should show enforcement status section.""" + with ( + patch( + "scc_cli.commands.config.config.load_cached_org_config", + return_value=mock_org_config, + ), + patch("scc_cli.commands.config.config.get_selected_profile", return_value="dev"), + patch( + "scc_cli.commands.config.compute_effective_config", + return_value=effective_config_basic, + ), + ): + result = runner.invoke(cli.app, ["config", "explain"]) + + assert result.exit_code == 0 + assert "Enforcement Status" in result.output + def test_explain_shows_source_attribution(self, effective_config_basic, mock_org_config): """Should show where each setting came from.""" with ( @@ -593,6 +612,114 @@ def test_explain_help(self): assert "config" in result.output.lower() +class TestConfigExplainWarnings: + """Tests for advisory warnings in config explain.""" + + def test_explain_warns_on_auto_resume(self, effective_config_basic): + """Should warn when session.auto_resume is set in config.""" + org_config = { + "schema_version": "1.0.0", + "organization": {"name": "Test Org", "id": "test-org"}, + "defaults": {"session": {"auto_resume": True}}, + "profiles": {"dev": {"description": "Dev team"}}, + } + + with ( + patch( + "scc_cli.commands.config.config.load_cached_org_config", + return_value=org_config, + ), + patch("scc_cli.commands.config.config.get_selected_profile", return_value="dev"), + patch( + "scc_cli.commands.config.compute_effective_config", + return_value=effective_config_basic, + ), + ): + result = runner.invoke(cli.app, ["config", "explain"]) + + assert result.exit_code == 0 + assert "Warnings" in result.output + assert "auto_resume" in result.output + + def test_explain_warns_on_team_network_policy(self, effective_config_basic): + """Should warn when team network_policy is less restrictive than org default.""" + effective_config_basic.network_policy = "isolated" + org_config = { + "schema_version": "1.0.0", + "organization": {"name": "Test Org", "id": "test-org"}, + "defaults": {"network_policy": "isolated"}, + "profiles": {"dev": {"description": "Dev team", "network_policy": "unrestricted"}}, + } + + with ( + patch( + "scc_cli.commands.config.config.load_cached_org_config", + return_value=org_config, + ), + patch("scc_cli.commands.config.config.get_selected_profile", return_value="dev"), + patch( + "scc_cli.commands.config.compute_effective_config", + return_value=effective_config_basic, + ), + ): + result = runner.invoke(cli.app, ["config", "explain"]) + + assert result.exit_code == 0 + assert "network_policy" in result.output + assert "ignored" in result.output.lower() + + def test_explain_warns_on_missing_proxy_env(self, effective_config_basic): + """Should warn when corp-proxy-only has no proxy env configured.""" + effective_config_basic.network_policy = "corp-proxy-only" + org_config = { + "schema_version": "1.0.0", + "organization": {"name": "Test Org", "id": "test-org"}, + "defaults": {"network_policy": "corp-proxy-only"}, + "profiles": {"dev": {"description": "Dev team"}}, + } + + with ( + patch( + "scc_cli.commands.config.config.load_cached_org_config", + return_value=org_config, + ), + patch("scc_cli.commands.config.config.get_selected_profile", return_value="dev"), + patch( + "scc_cli.commands.config.compute_effective_config", + return_value=effective_config_basic, + ), + patch.dict("os.environ", {}, clear=True), + ): + result = runner.invoke(cli.app, ["config", "explain"]) + + assert result.exit_code == 0 + assert "proxy" in result.output.lower() + + +class TestConfigExplainJsonOutput: + """Tests for JSON output for config explain.""" + + def test_explain_json_includes_enforcement(self, effective_config_basic, mock_org_config): + """Should emit a JSON envelope with enforcement status.""" + with ( + patch( + "scc_cli.commands.config.config.load_cached_org_config", + return_value=mock_org_config, + ), + patch("scc_cli.commands.config.config.get_selected_profile", return_value="dev"), + patch( + "scc_cli.commands.config.compute_effective_config", + return_value=effective_config_basic, + ), + ): + result = runner.invoke(cli.app, ["config", "explain", "--json"]) + + assert result.exit_code == 0 + payload = json.loads(result.output) + assert payload["kind"] == "ConfigExplain" + assert "enforcement" in payload["data"] + + # ═══════════════════════════════════════════════════════════════════════════════ # Golden tests for explain output format # ═══════════════════════════════════════════════════════════════════════════════ diff --git a/tests/test_config_inheritance.py b/tests/test_config_inheritance.py index 9555dc1..d6b6c03 100644 --- a/tests/test_config_inheritance.py +++ b/tests/test_config_inheritance.py @@ -436,6 +436,59 @@ def test_minimal_config_uses_defaults(self, minimal_org_config): assert result.network_policy is None or result.network_policy == "" +class TestComputeEffectiveConfigNetworkPolicy: + """Tests for network_policy merging and enforcement.""" + + def test_team_network_policy_more_restrictive(self, valid_org_config): + """Team can tighten org network policy.""" + from scc_cli.application.compute_effective_config import compute_effective_config + + valid_org_config["defaults"]["network_policy"] = "unrestricted" + valid_org_config["profiles"]["urban-planning"]["network_policy"] = "isolated" + + result = compute_effective_config( + org_config=valid_org_config, + team_name="urban-planning", + ) + + assert result.network_policy == "isolated" + + def test_team_network_policy_less_restrictive(self, valid_org_config): + """Team cannot loosen org network policy.""" + from scc_cli.application.compute_effective_config import compute_effective_config + + valid_org_config["defaults"]["network_policy"] = "isolated" + valid_org_config["profiles"]["urban-planning"]["network_policy"] = "unrestricted" + + result = compute_effective_config( + org_config=valid_org_config, + team_name="urban-planning", + ) + + assert result.network_policy == "isolated" + + def test_isolated_blocks_network_mcp(self, valid_org_config): + """Isolated policy blocks HTTP/SSE MCP servers.""" + from scc_cli.application.compute_effective_config import compute_effective_config + + valid_org_config["defaults"]["network_policy"] = "isolated" + valid_org_config["profiles"]["urban-planning"]["additional_mcp_servers"] = [ + { + "name": "http-mcp", + "type": "http", + "url": "https://api.sundsvall.se/mcp", + } + ] + + result = compute_effective_config( + org_config=valid_org_config, + team_name="urban-planning", + ) + + assert result.mcp_servers == [] + assert any(item.item == "http-mcp" for item in result.blocked_items) + + class TestComputeEffectiveConfigDelegation: """Tests for delegation hierarchy enforcement. @@ -627,6 +680,30 @@ def test_blocked_mcp_server_rejected(self, valid_org_config): mcp_names = [s.name for s in result.mcp_servers] assert "gis-internal" not in mcp_names + def test_blocked_mcp_server_rejected_by_command(self, valid_org_config): + """Blocked MCP patterns should match stdio command paths.""" + from scc_cli.application.compute_effective_config import compute_effective_config + + valid_org_config["security"]["allow_stdio_mcp"] = True + valid_org_config["security"]["allowed_stdio_prefixes"] = ["/usr/local/bin"] + valid_org_config["security"]["blocked_mcp_servers"] = ["/usr/local/bin/blocked-tool"] + valid_org_config["profiles"]["urban-planning"]["additional_mcp_servers"] = [ + { + "name": "blocked-stdio", + "type": "stdio", + "command": "/usr/local/bin/blocked-tool", + } + ] + + result = compute_effective_config( + org_config=valid_org_config, + team_name="urban-planning", + project_config=None, + ) + + mcp_names = [s.name for s in result.mcp_servers] + assert "blocked-stdio" not in mcp_names + def test_security_blocks_cannot_be_overridden(self, valid_org_config): """Security blocks apply regardless of delegation settings.""" from scc_cli.application.compute_effective_config import compute_effective_config @@ -729,6 +806,23 @@ def test_exact_match_pattern(self, valid_org_config): assert "exact-plugin-extended" in result.plugins # Not blocked assert "other-exact-plugin" in result.plugins # Not blocked + def test_bare_pattern_blocks_any_marketplace(self, valid_org_config): + """Bare plugin patterns should match regardless of marketplace.""" + from scc_cli.application.compute_effective_config import compute_effective_config + + valid_org_config["security"]["blocked_plugins"] = ["*-experimental"] + valid_org_config["defaults"]["enabled_plugins"] = [ + "tool-experimental@internal", + "safe-tool@internal", + ] + + result = compute_effective_config( + org_config=valid_org_config, team_name=None, project_config=None + ) + + assert "tool-experimental@internal" not in result.plugins + assert "safe-tool@internal" in result.plugins + class TestComputeEffectiveConfigDecisionTracking: """Tests for decision tracking (for scc config explain).""" diff --git a/tests/test_docker.py b/tests/test_docker.py index e5d745a..a529363 100644 --- a/tests/test_docker.py +++ b/tests/test_docker.py @@ -20,9 +20,13 @@ class TestResetGlobalSettings: def test_reset_global_settings_success(self): """reset_global_settings should return True on success.""" - with patch( - "scc_cli.docker.launch.inject_file_to_sandbox_volume", return_value=True - ) as mock_inject: + with ( + patch("scc_cli.docker.launch.reset_plugin_caches", return_value=True), + patch( + "scc_cli.docker.launch.inject_file_to_sandbox_volume", + return_value=True, + ) as mock_inject, + ): result = docker.reset_global_settings() assert result is True @@ -30,7 +34,13 @@ def test_reset_global_settings_success(self): def test_reset_global_settings_failure(self): """reset_global_settings should return False on failure.""" - with patch("scc_cli.docker.launch.inject_file_to_sandbox_volume", return_value=False): + with ( + patch("scc_cli.docker.launch.reset_plugin_caches", return_value=True), + patch( + "scc_cli.docker.launch.inject_file_to_sandbox_volume", + return_value=False, + ), + ): result = docker.reset_global_settings() assert result is False diff --git a/tests/test_docker_core.py b/tests/test_docker_core.py index 1976001..82e96b6 100644 --- a/tests/test_docker_core.py +++ b/tests/test_docker_core.py @@ -254,6 +254,14 @@ def test_includes_workspace_flag(self, tmp_path): claude_idx = cmd.index("claude") assert w_idx < claude_idx + def test_env_vars_injected(self): + """Should include -e flags when env_vars are provided.""" + cmd = docker.build_command(env_vars={"HTTP_PROXY": "http://proxy", "NO_PROXY": "localhost"}) + + assert "-e" in cmd + assert "HTTP_PROXY=http://proxy" in cmd + assert "NO_PROXY=localhost" in cmd + def test_continue_session_flag(self): """Should include -c flag after claude when continue_session is True.""" cmd = docker.build_command(continue_session=True) diff --git a/tests/test_help_grouping.py b/tests/test_help_grouping.py index 9ee48fd..075f137 100644 --- a/tests/test_help_grouping.py +++ b/tests/test_help_grouping.py @@ -8,6 +8,7 @@ - Consistent grouping across related commands """ +import os import subprocess # ═══════════════════════════════════════════════════════════════════════════════ @@ -158,9 +159,13 @@ class TestHelpOutput: def test_help_output_contains_group_headers(self) -> None: """scc --help should show group headers.""" + env = os.environ.copy() + env.setdefault("UV_OFFLINE", "1") + env.setdefault("UV_NO_SYNC", "1") result = subprocess.run( ["uv", "run", "scc", "--help"], capture_output=True, + env=env, text=True, timeout=30, ) @@ -174,9 +179,13 @@ def test_help_output_contains_group_headers(self) -> None: def test_help_output_organized_not_flat(self) -> None: """scc --help should not show a flat list of commands.""" + env = os.environ.copy() + env.setdefault("UV_OFFLINE", "1") + env.setdefault("UV_NO_SYNC", "1") result = subprocess.run( ["uv", "run", "scc", "--help"], capture_output=True, + env=env, text=True, timeout=30, ) diff --git a/tests/test_marketplace_normalize.py b/tests/test_marketplace_normalize.py index 1a1613f..1746812 100644 --- a/tests/test_marketplace_normalize.py +++ b/tests/test_marketplace_normalize.py @@ -211,6 +211,13 @@ def test_case_insensitive(self) -> None: assert matches_pattern("tool@internal", "TOOL@INTERNAL") is True assert matches_pattern("MALICIOUS-tool@shared", "malicious-*") is True + def test_bare_pattern_matches_plugin_name(self) -> None: + """Bare patterns should match plugin names across marketplaces.""" + from scc_cli.marketplace.normalize import matches_pattern + + assert matches_pattern("tool-experimental@internal", "*-experimental") is True + assert matches_pattern("tool@internal", "tool") is True + def test_special_characters(self) -> None: """Handles special characters in names.""" from scc_cli.marketplace.normalize import matches_pattern diff --git a/tests/test_plugin_isolation.py b/tests/test_plugin_isolation.py index f349f75..d3b9001 100644 --- a/tests/test_plugin_isolation.py +++ b/tests/test_plugin_isolation.py @@ -14,9 +14,12 @@ class TestResetGlobalSettings: def test_reset_global_settings_writes_empty_json(self): """reset_global_settings should write empty JSON to settings.json.""" - with patch( - "scc_cli.docker.launch.inject_file_to_sandbox_volume", return_value=True - ) as mock_inject: + with ( + patch( + "scc_cli.docker.launch.inject_file_to_sandbox_volume", return_value=True + ) as mock_inject, + patch("scc_cli.docker.launch.reset_plugin_caches", return_value=True), + ): result = docker.reset_global_settings() assert result is True @@ -73,6 +76,7 @@ def test_settings_reset_clears_old_plugins(self): patch( "scc_cli.docker.launch.inject_file_to_sandbox_volume", return_value=True ) as mock_inject, + patch("scc_cli.docker.launch.reset_plugin_caches", return_value=True), ): result = docker.reset_global_settings() diff --git a/uv.lock b/uv.lock index 47f5e44..4383083 100644 --- a/uv.lock +++ b/uv.lock @@ -951,7 +951,7 @@ wheels = [ [[package]] name = "scc-cli" -version = "1.6.4" +version = "1.7.0" source = { editable = "." } dependencies = [ { name = "jsonschema" },