#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mcp/config.py
"""
Purpose
-------
Configuration loader for the ``oct-mcp`` server. Reads
``~/.oct-mcp/config.json`` (user-level trusted config), merges with
environment variable overrides, and exposes a :class:`McpConfig`
dataclass consumed by every layer of the six-layer pipeline.
Responsibilities
----------------
- Define the canonical :class:`McpConfig` schema with safe defaults.
- Implement :func:`load_config` which locates, validates, and returns a
populated ``McpConfig``.
- Honour ``OCT_MCP_SAFE_MODE`` and ``OCT_TRUST_REPO_CONFIG`` env vars.
- Never read or trust project-level config (``debug_config.json``,
``.octrc.json``) unless ``trust_repo_config`` is explicitly enabled.
- Apply a 64 KB file-size guard (matching OI-414) to the user config.
Diagnostics
-----------
Domain: MCP
Levels:
L1 — errors: config file read failures, schema errors
L2 — lifecycle: config loaded, safe-mode active
L4 — deep trace: resolved field values
Contracts
---------
- This module does not import ``click``, the MCP SDK, or any other
``oct.mcp.*`` module (no cycles).
- :func:`load_config` always returns a fully-populated ``McpConfig``
even if the file is missing — defaults are safe.
- Config values coming from JSON are validated and clamped so callers
never receive out-of-range integers.
"""
from __future__ import annotations
import json
import os
from dataclasses import dataclass, field
from pathlib import Path
try:
from oc_diagnostics import _dbg as _real_dbg
def _dbg(*args, **kwargs) -> None:
_real_dbg(*args, **kwargs)
except ImportError: # pragma: no cover
def _dbg(*args, **kwargs) -> None:
return None
# ---------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------
#: Maximum size for ``~/.oct-mcp/config.json``. Matches OI-414 (.octrc.json
#: cap) applied to the user config as defence-in-depth.
_CONFIG_MAX_BYTES: int = 64 * 1024 # 64 KB
#: Default location for the user-level MCP server config.
_DEFAULT_CONFIG_PATH: Path = Path.home() / ".oct-mcp" / "config.json"
#: Valid profile names. See blueprint §5.3.
VALID_MCP_PROFILES: tuple[str, ...] = ("default", "strict", "airgapped")
#: Minimum / maximum allowed values for integer config fields.
_RATE_LIMIT_MIN: int = 1
_RATE_LIMIT_MAX: int = 1000
_TIMEOUT_MIN: int = 5
_TIMEOUT_MAX: int = 300
_TIMEOUT_TEST_MAX: int = 600
_MAX_OUTPUT_MIN: int = 1024 # 1 KB
_MAX_OUTPUT_MAX: int = 10 * 1024 * 1024 # 10 MB
_MAX_JOBS_MIN: int = 1
_MAX_JOBS_MAX: int = 16
_MEMORY_LIMIT_MIN: int = 0 # 0 = disabled
_MEMORY_LIMIT_MAX: int = 8192 # MB
_METRICS_PORT_MIN: int = 1
_METRICS_PORT_MAX: int = 65535
#: Valid sandbox backend names. "auto" picks best available for the platform.
VALID_SANDBOX_BACKENDS: tuple[str, ...] = ("auto", "native", "bubblewrap", "sandbox_exec")
# ---------------------------------------------------------------------
# Data model
# ---------------------------------------------------------------------
[docs]
@dataclass
class McpConfig:
"""Runtime configuration for the ``oct-mcp`` server.
All fields have safe, conservative defaults that work without any
config file present. Fields loaded from JSON are clamped to the
valid ranges defined by the module-level constants.
``trust_repo_config`` defaults to ``False`` — repo-level config
(``debug_config.json``, ``.octrc.json``) is untrusted by default per
blueprint §8.1. Set to ``True`` via ``OCT_TRUST_REPO_CONFIG=1`` or
the config file field ``trust_repo_config: true``.
"""
profile: str = "default"
"""Active security profile: ``"default"``, ``"strict"``, or ``"airgapped"``."""
rate_limit_per_minute: int = 30
"""Maximum tool calls per session per minute (Gateway rate limiter)."""
timeout_default_seconds: int = 60
"""Per-tool subprocess timeout in seconds (except ``oct test``)."""
timeout_test_seconds: int = 300
"""Subprocess timeout for ``oct test`` (pytest may need longer)."""
max_output_bytes: int = 1_048_576 # 1 MB
"""Hard output-size cap per tool call. Output exceeding this is truncated."""
max_jobs: int = 4
"""Maximum ``--jobs`` value forwarded to tools that support parallelism."""
auto_approve_read_tools: bool = True
"""If True, read-only tools execute without confirmation."""
dry_run_default_for_writes: bool = True
"""If True, write tools default to dry-run mode (Phase 5B guard)."""
redaction_always_on: bool = False
"""Force redaction even when not in production mode."""
audit_log_path: str = str(Path.home() / ".oct-mcp" / "audit.log")
"""Path to the JSONL audit log. Rotation files go in the same directory."""
audit_log_max_files: int = 20
"""Maximum number of rotated audit log files to keep."""
audit_log_max_size_mb: float = 5.0
"""Rotation threshold per audit file, in megabytes."""
trust_repo_config: bool = False
"""Whether to load and trust repo-level config (``.octrc.json``)."""
safe_mode: bool = False
"""Hard-override: all tools return error; set via ``OCT_MCP_SAFE_MODE=1``."""
sandbox_backend: str = "auto"
"""OS-level sandbox backend: ``"auto"``, ``"native"``, ``"bubblewrap"``,
or ``"sandbox_exec"``. ``"auto"`` picks the best available backend for
the current platform, falling back to ``"native"`` if no enhanced backend
is found. Set via ``OCT_MCP_SANDBOX`` env var."""
memory_limit_mb: int = 512
"""Per-subprocess memory limit in megabytes (POSIX only; 0 = disabled).
On Windows this field is accepted but has no effect — use Docker for
memory enforcement on Windows. Range: 0–8192."""
metrics_port: int | None = None
"""If set, starts a Prometheus ``/metrics`` HTTP server on this port
in a background thread. Requires ``oct[metrics]`` (prometheus-client).
``None`` (default) disables metrics collection."""
policy_source: str | None = None
"""If set, load additional policy overrides from this source at startup.
Accepts a local filesystem path or an ``https://`` URL. Overrides are
merged on top of the base config. Any load failure falls back to
built-in defaults silently."""
# ---------------------------------------------------------------------
# Loader
# ---------------------------------------------------------------------
[docs]
def load_config(path: Path | None = None) -> McpConfig:
"""Load and return a :class:`McpConfig`.
Resolution order:
1. Read ``~/.oct-mcp/config.json`` (or *path* if supplied).
2. Validate and clamp JSON field values.
3. Apply environment variable overrides:
- ``OCT_MCP_SAFE_MODE=1`` → ``safe_mode = True``
- ``OCT_TRUST_REPO_CONFIG=1`` → ``trust_repo_config = True``
- ``OCT_MCP_PROFILE=<name>`` → ``profile = <name>``
4. Return a fully-populated ``McpConfig``.
Never raises. Any read or parse failure falls back to the default
config and logs an error at level 1.
"""
func = "load_config"
cfg_path = path or _DEFAULT_CONFIG_PATH
config = McpConfig()
if cfg_path.exists():
try:
size = cfg_path.stat().st_size
if size > _CONFIG_MAX_BYTES:
_dbg(
"MCP", func,
f"SYSTEM_ERROR: config file too large ({size} bytes > "
f"{_CONFIG_MAX_BYTES}); using defaults",
1,
)
else:
raw = json.loads(cfg_path.read_text(encoding="utf-8"))
if isinstance(raw, dict):
config = _apply_json(config, raw)
_dbg("MCP", func, f"loaded config from {cfg_path}", 2)
else:
_dbg(
"MCP", func,
"SYSTEM_ERROR: config file is not a JSON object; using defaults",
1,
)
except (OSError, json.JSONDecodeError, ValueError) as exc:
_dbg("MCP", func, f"SYSTEM_ERROR: cannot read config: {exc}", 1)
else:
_dbg("MCP", func, f"no config file at {cfg_path}; using defaults", 4)
# Environment variable overrides (always trusted — set by user/CI).
if os.environ.get("OCT_MCP_SAFE_MODE", "").strip() == "1":
config.safe_mode = True
_dbg("MCP", func, "safe_mode=True (OCT_MCP_SAFE_MODE=1)", 2)
if os.environ.get("OCT_TRUST_REPO_CONFIG", "").strip() == "1":
config.trust_repo_config = True
_dbg("MCP", func, "trust_repo_config=True (OCT_TRUST_REPO_CONFIG=1)", 2)
profile_env = os.environ.get("OCT_MCP_PROFILE", "").strip()
if profile_env in VALID_MCP_PROFILES:
config.profile = profile_env
_dbg("MCP", func, f"profile={config.profile} (OCT_MCP_PROFILE)", 2)
sandbox_env = os.environ.get("OCT_MCP_SANDBOX", "").strip()
if sandbox_env in VALID_SANDBOX_BACKENDS:
config.sandbox_backend = sandbox_env
_dbg("MCP", func, f"sandbox_backend={config.sandbox_backend} (OCT_MCP_SANDBOX)", 2)
return config
def _apply_json(base: McpConfig, data: dict) -> McpConfig:
"""Return a new :class:`McpConfig` with fields from *data* merged in.
Only known fields are applied; unknown keys are silently ignored
(the config file is user-controlled but we don't crash on new keys).
Field values are clamped to the valid ranges defined by the module
constants. Booleans must be actual JSON booleans, not strings.
"""
def _clamp(val: int | float, lo: int | float, hi: int | float):
return max(lo, min(hi, val))
fields: dict = {}
if "profile" in data and data["profile"] in VALID_MCP_PROFILES:
fields["profile"] = data["profile"]
if "rate_limit_per_minute" in data:
v = data["rate_limit_per_minute"]
if isinstance(v, int):
fields["rate_limit_per_minute"] = _clamp(v, _RATE_LIMIT_MIN, _RATE_LIMIT_MAX)
if "timeout_default_seconds" in data:
v = data["timeout_default_seconds"]
if isinstance(v, int):
fields["timeout_default_seconds"] = _clamp(v, _TIMEOUT_MIN, _TIMEOUT_MAX)
if "timeout_test_seconds" in data:
v = data["timeout_test_seconds"]
if isinstance(v, int):
fields["timeout_test_seconds"] = _clamp(v, _TIMEOUT_MIN, _TIMEOUT_TEST_MAX)
if "max_output_bytes" in data:
v = data["max_output_bytes"]
if isinstance(v, int):
fields["max_output_bytes"] = _clamp(v, _MAX_OUTPUT_MIN, _MAX_OUTPUT_MAX)
if "max_jobs" in data:
v = data["max_jobs"]
if isinstance(v, int):
fields["max_jobs"] = _clamp(v, _MAX_JOBS_MIN, _MAX_JOBS_MAX)
for bool_field in (
"auto_approve_read_tools",
"dry_run_default_for_writes",
"redaction_always_on",
"trust_repo_config",
):
if bool_field in data and isinstance(data[bool_field], bool):
fields[bool_field] = data[bool_field]
if "audit_log_path" in data and isinstance(data["audit_log_path"], str):
fields["audit_log_path"] = data["audit_log_path"]
if "audit_log_max_files" in data:
v = data["audit_log_max_files"]
if isinstance(v, int) and v >= 1:
fields["audit_log_max_files"] = min(v, 100)
if "audit_log_max_size_mb" in data:
v = data["audit_log_max_size_mb"]
if isinstance(v, (int, float)) and v > 0:
fields["audit_log_max_size_mb"] = min(float(v), 100.0)
if "sandbox_backend" in data and data["sandbox_backend"] in VALID_SANDBOX_BACKENDS:
fields["sandbox_backend"] = data["sandbox_backend"]
if "memory_limit_mb" in data:
v = data["memory_limit_mb"]
if isinstance(v, int):
fields["memory_limit_mb"] = _clamp(v, _MEMORY_LIMIT_MIN, _MEMORY_LIMIT_MAX)
if "metrics_port" in data:
v = data["metrics_port"]
if v is None:
fields["metrics_port"] = None
elif isinstance(v, int):
fields["metrics_port"] = _clamp(v, _METRICS_PORT_MIN, _METRICS_PORT_MAX)
if "policy_source" in data:
v = data["policy_source"]
if v is None or isinstance(v, str):
fields["policy_source"] = v or None
from dataclasses import replace as _replace
return _replace(base, **fields)
