policy_loader.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mcp/policy_loader.py
 
"""
Purpose
-------
Centralised policy loader for the ``oct-mcp`` server (Phase 5C, M-C7).
 
Loads policy override JSON from a local filesystem path or an HTTPS URL
at server startup. The loaded overrides are applied on top of the base
:class:`~oct.mcp.config.McpConfig` before the policy engine is
instantiated, enabling enterprise deployments to distribute a shared
security policy from a central location.
 
Configuration
-------------
Set ``policy_source`` in ``~/.oct-mcp/config.json``::
 
    {
        "policy_source": "https://internal.company.com/oct-policy.json"
    }
 
Or provide a local path::
 
    {
        "policy_source": "/etc/oct/policy.json"
    }
 
Policy JSON schema (all fields optional)::
 
    {
        "profile": "strict",
        "rate_limit_per_minute": 20,
        "timeout_default_seconds": 30,
        "max_output_bytes": 524288,
        "dry_run_default_for_writes": true
    }
 
Only the keys listed in :data:`_ALLOWED_OVERRIDE_KEYS` are honoured.
Unknown keys are silently dropped. Type mismatches are silently ignored.
 
Security
--------
- HTTPS URLs only. HTTP URLs are rejected to prevent cleartext policy
  transmission.
- File size capped at 64 KB (same as ``config.json``).
- Schema is an allowlist — no arbitrary key injection possible.
- Any failure (network error, parse error, schema violation) falls back
  to built-in defaults silently without crashing the server.
 
Diagnostics
-----------
Domain: MCP
Levels:
    L1 — errors: load failures, schema violations
    L2 — lifecycle: policy loaded, overrides applied
    L4 — deep trace: resolved field values
 
Contracts
---------
- :func:`load_policy_source` never raises — returns ``{}`` on any error.
- :func:`apply_policy_overrides` never raises — returns *config* unchanged
  on any error.
- No ``http://`` URLs are accepted — HTTPS only.
"""
 
from __future__ import annotations
 
import dataclasses
import json
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 bytes read from a policy source (matches config.json guard).
_POLICY_MAX_BYTES: int = 64 * 1024  # 64 KB
 
#: Only these McpConfig fields may be overridden via a policy source.
#: This is an allowlist — unknown keys are silently dropped.
_ALLOWED_OVERRIDE_KEYS: frozenset[str] = frozenset({
    "profile",
    "rate_limit_per_minute",
    "timeout_default_seconds",
    "max_output_bytes",
    "dry_run_default_for_writes",
})
 
 
# ---------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------
 
 
def load_policy_source(source: str, timeout_s: int = 5) -> dict:
    """Load policy override JSON from *source*.
 
    Parameters
    ----------
    source
        A local filesystem path or an ``https://`` URL. ``http://`` URLs
        are rejected. Local paths may be absolute or relative.
    timeout_s
        Network timeout in seconds for HTTPS requests (default 5).
 
    Returns
    -------
    A ``dict`` containing only the keys in :data:`_ALLOWED_OVERRIDE_KEYS`
    with validated values. Returns ``{}`` on any failure.
    """
    func = "load_policy_source"
 
    if not source:
        return {}
 
    try:
        if source.startswith("http://"):
            _dbg(
                "MCP", func,
                "SYSTEM_ERROR: HTTP policy source rejected; use HTTPS",
                1,
            )
            return {}
 
        if source.startswith("https://"):
            raw = _fetch_url(source, timeout_s)
        else:
            raw = _read_file(source)
 
        if raw is None:
            return {}
 
        data = json.loads(raw)
        if not isinstance(data, dict):
            _dbg("MCP", func, "SYSTEM_ERROR: policy source is not a JSON object", 1)
            return {}
 
        overrides = _validate_overrides(data)
        _dbg("MCP", func, f"loaded {len(overrides)} override(s) from {source}", 2)
        return overrides
 
    except Exception as exc:
        _dbg("MCP", func, f"SYSTEM_ERROR: failed to load policy source: {exc}", 1)
        return {}
 
 
def apply_policy_overrides(config: object, overrides: dict) -> object:
    """Return a new :class:`~oct.mcp.config.McpConfig` with *overrides* applied.
 
    Parameters
    ----------
    config
        The base :class:`~oct.mcp.config.McpConfig` instance.
    overrides
        Dict of validated overrides from :func:`load_policy_source`.
 
    Returns
    -------
    A new ``McpConfig`` with the override values merged in. Returns
    *config* unchanged if *overrides* is empty or an error occurs.
    """
    func = "apply_policy_overrides"
    if not overrides:
        return config
    try:
        result = dataclasses.replace(config, **overrides)
        _dbg("MCP", func, f"applied overrides: {list(overrides.keys())}", 2)
        return result
    except Exception as exc:  # pragma: no cover
        _dbg("MCP", func, f"SYSTEM_ERROR: failed to apply overrides: {exc}", 1)
        return config
 
 
# ---------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------
 
 
def _fetch_url(url: str, timeout_s: int) -> bytes | None:
    """Fetch *url* via HTTPS and return raw bytes, or ``None`` on error."""
    func = "_fetch_url"
    try:
        import urllib.request
        req = urllib.request.Request(
            url,
            headers={"User-Agent": "oct-mcp/policy-loader"},
        )
        with urllib.request.urlopen(req, timeout=timeout_s) as resp:
            raw = resp.read(_POLICY_MAX_BYTES + 1)
        if len(raw) > _POLICY_MAX_BYTES:
            _dbg("MCP", func, "SYSTEM_ERROR: policy source too large (>64 KB)", 1)
            return None
        return raw
    except Exception as exc:
        _dbg("MCP", func, f"SYSTEM_ERROR: failed to fetch {url}: {exc}", 1)
        return None
 
 
def _read_file(path: str) -> bytes | None:
    """Read *path* from the filesystem and return raw bytes, or ``None``."""
    func = "_read_file"
    try:
        p = Path(path)
        size = p.stat().st_size
        if size > _POLICY_MAX_BYTES:
            _dbg("MCP", func, f"SYSTEM_ERROR: policy file too large ({size} bytes)", 1)
            return None
        return p.read_bytes()
    except Exception as exc:
        _dbg("MCP", func, f"SYSTEM_ERROR: failed to read {path}: {exc}", 1)
        return None
 
 
def _validate_overrides(data: dict) -> dict:
    """Return a sanitised override dict with only allowed, well-typed values."""
    from oct.mcp.config import VALID_MCP_PROFILES
 
    result: dict = {}
 
    if "profile" in data and data["profile"] in VALID_MCP_PROFILES:
        result["profile"] = data["profile"]
 
    for int_field in ("rate_limit_per_minute", "timeout_default_seconds", "max_output_bytes"):
        if int_field in data and isinstance(data[int_field], int) and data[int_field] > 0:
            result[int_field] = data[int_field]
 
    if "dry_run_default_for_writes" in data and isinstance(
        data["dry_run_default_for_writes"], bool
    ):
        result["dry_run_default_for_writes"] = data["dry_run_default_for_writes"]
 
    return result