config_validation.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/config_validation.py
 
 
"""
Runtime configuration validation.
 
Purpose
-------
Validate user-controlled environment variables before OC Chat builds
provider clients, session state, or endpoint registries. The checks here
are intentionally local and deterministic: no network probes, no writes,
and no dependency on an interactive terminal.
 
Responsibilities
----------------
- :func:`validate_runtime_config` - inspect env-derived startup config
  and return structured errors / warnings.
- :func:`format_config_issue` - convert one issue into a user-facing
  one-line diagnostic.
- :class:`ConfigValidationResult` - group issues and expose convenience
  accessors for callers that need to stop on errors.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - startup validation summary.
    L3 - validation details.
    L4 - module load.
 
Contracts
---------
- Validation never mutates the filesystem or environment.
- Warnings are advisory; errors should stop startup before provider
  clients or sessions are constructed.
- Provider-specific API keys are required only when the provider
  declares a non-empty ``api_key_env_var``.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.provider_oauth (lazy; stored-credential check when the API key variable is unset)
- oc_chat.core.provider_plugin (lazy; ProviderPlugin type of the provider being validated)
"""
from __future__ import annotations
 
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
import ipaddress
import os
from pathlib import Path
import re
from typing import TYPE_CHECKING, Literal
from urllib.parse import urlparse
 
from oc_diagnostics import _dbg
 
if TYPE_CHECKING:
    from oc_chat.core.provider_plugin import ProviderPlugin
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
Severity = Literal["error", "warning"]
 
_ENDPOINT_ENTRY_RE = re.compile(r"^\s*([A-Za-z0-9_.\-]+)\s*=\s*(\S.*?)\s*$")
_LOCAL_HOSTS = frozenset({"localhost", "127.0.0.1", "::1", "0.0.0.0"})
 
 
@dataclass(frozen=True)
class ConfigIssue:
    """One startup configuration diagnostic."""
 
    severity: Severity
    key: str
    message: str
    hint: str = ""
 
 
@dataclass(frozen=True)
class ConfigValidationResult:
    """Collection of startup configuration diagnostics."""
 
    issues: tuple[ConfigIssue, ...] = ()
 
    @property
    def errors(self) -> tuple[ConfigIssue, ...]:
        """Return fatal validation issues."""
        return tuple(issue for issue in self.issues if issue.severity == "error")
 
    @property
    def warnings(self) -> tuple[ConfigIssue, ...]:
        """Return non-fatal validation issues."""
        return tuple(issue for issue in self.issues if issue.severity == "warning")
 
    @property
    def ok(self) -> bool:
        """True when no fatal validation errors were found."""
        return not self.errors
 
 
def format_config_issue(issue: ConfigIssue) -> str:
    """Return a user-facing one-line diagnostic for *issue*."""
    prefix = "ERROR" if issue.severity == "error" else "WARNING"
    text = f"{prefix}: {issue.key}: {issue.message}"
    if issue.hint:
        text += f" Hint: {issue.hint}"
    return text
 
 
def format_config_issues(issues: Sequence[ConfigIssue]) -> list[str]:
    """Format multiple configuration diagnostics."""
    return [format_config_issue(issue) for issue in issues]
 
 
def _add(
    issues: list[ConfigIssue],
    severity: Severity,
    key: str,
    message: str,
    hint: str = "",
) -> None:
    issues.append(ConfigIssue(severity=severity, key=key, message=message, hint=hint))
 
 
def _validate_http_url(
    issues: list[ConfigIssue],
    key: str,
    value: str,
) -> bool:
    parsed = urlparse(value)
    if parsed.scheme not in ("http", "https") or not parsed.netloc or not parsed.hostname:
        _add(
            issues,
            "error",
            key,
            f"{value!r} is not a valid HTTP(S) URL.",
            "Use a URL like http://localhost:11434 or https://api.example.com.",
        )
        return False
    return True
 
 
def _remote_allowed(env: Mapping[str, str]) -> bool:
    return env.get("OC_CHAT_ALLOW_REMOTE_ENDPOINTS", "") == "1"
 
 
def _classify_endpoint_url(base_url: str, env: Mapping[str, str]) -> str:
    parsed = urlparse(base_url)
    host = (parsed.hostname or "").strip().lower()
    if not host:
        return "blocked"
    if host in _LOCAL_HOSTS:
        return "local"
    try:
        ip = ipaddress.ip_address(host)
        if ip.is_loopback:
            return "local"
        if ip.is_private:
            return "lan"
    except ValueError:
        pass
    return "remote" if _remote_allowed(env) else "blocked"
 
 
def _validate_provider_selection(
    issues: list[ConfigIssue],
    provider_name: str,
    provider_names: Sequence[str] | None,
) -> None:
    if not provider_name:
        _add(
            issues,
            "error",
            "OC_CHAT_PROVIDER",
            "provider name is blank.",
            "Unset OC_CHAT_PROVIDER or set it to a registered provider.",
        )
        return
    if provider_names is not None and provider_name not in set(provider_names):
        available = ", ".join(sorted(provider_names)) or "(none)"
        _add(
            issues,
            "error",
            "OC_CHAT_PROVIDER",
            f"unknown provider {provider_name!r}.",
            f"Available providers: {available}.",
        )
 
 
def _validate_provider_env(
    issues: list[ConfigIssue],
    env: Mapping[str, str],
    provider: "ProviderPlugin",
) -> None:
    from oc_chat.core.provider_oauth import has_usable_credentials
    if (provider.api_key_env_var and not env.get(provider.api_key_env_var, "").strip()
            and not has_usable_credentials(provider)):
        _add(
            issues,
            "error",
            provider.api_key_env_var,
            "required API key is not set.",
            f"Set {provider.api_key_env_var} in the environment or .env file.",
        )
 
    if not provider.base_url_env_var:
        return
 
    if provider.base_url_env_var in env and not env[provider.base_url_env_var].strip():
        _add(
            issues,
            "error",
            provider.base_url_env_var,
            "base URL is set to a blank value.",
            "Unset it to use the provider default, or set a valid HTTP(S) URL.",
        )
        return
 
    base_url = env.get(provider.base_url_env_var, provider.default_base_url).strip()
    if base_url:
        _validate_http_url(issues, provider.base_url_env_var, base_url)
 
 
def _validate_endpoint_env(issues: list[ConfigIssue], env: Mapping[str, str]) -> None:
    allow_remote = env.get("OC_CHAT_ALLOW_REMOTE_ENDPOINTS")
    if allow_remote not in (None, "", "1"):
        _add(
            issues,
            "warning",
            "OC_CHAT_ALLOW_REMOTE_ENDPOINTS",
            f"value {allow_remote!r} does not enable remote endpoints.",
            # OCT-LINT: disable=no_hardcoded_secrets reason="Environment variable names and user-facing hints, not secrets"
            "Set OC_CHAT_ALLOW_REMOTE_ENDPOINTS=1 to opt in, or unset it.",
        )
 
    raw = env.get("OLLAMA_BASE_URLS", "").strip()
    if not raw:
        return
 
    seen: set[str] = set()
    for entry in raw.split(","):
        entry = entry.strip()
        if not entry:
            continue
        match = _ENDPOINT_ENTRY_RE.match(entry)
        if not match:
            _add(
                issues,
                "error",
                "OLLAMA_BASE_URLS",
                f"malformed endpoint entry {entry!r}.",
                "Use name=url entries separated by commas.",
            )
            continue
        name, url = match.group(1), match.group(2)
        if name in seen:
            _add(
                issues,
                "warning",
                "OLLAMA_BASE_URLS",
                f"endpoint {name!r} is defined more than once; later value wins.",
            )
        seen.add(name)
        if not _validate_http_url(issues, "OLLAMA_BASE_URLS", url):
            continue
        if _classify_endpoint_url(url, env) == "blocked":
            _add(
                issues,
                "warning",
                "OLLAMA_BASE_URLS",
                f"endpoint {name!r} points to a public/remote host and will be blocked.",
                # OCT-LINT: disable=no_hardcoded_secrets reason="Environment variable names and user-facing hints, not secrets"
                "Set OC_CHAT_ALLOW_REMOTE_ENDPOINTS=1 only if you intend to allow it.",
            )
 
 
def _validate_sessions_dir(
    issues: list[ConfigIssue],
    env: Mapping[str, str],
    sessions_dir: str | None,
) -> None:
    if "OC_CHAT_SESSIONS_DIR" in env and not env["OC_CHAT_SESSIONS_DIR"].strip():
        _add(
            issues,
            "error",
            "OC_CHAT_SESSIONS_DIR",
            "sessions directory is set to a blank value.",
            "Unset it or set a writable directory path.",
        )
        return
 
    path_text = sessions_dir or env.get("OC_CHAT_SESSIONS_DIR", "")
    if not path_text:
        return
    path = Path(path_text)
    if path.exists() and not path.is_dir():
        _add(
            issues,
            "error",
            "OC_CHAT_SESSIONS_DIR",
            f"{str(path)!r} exists but is not a directory.",
            "Set OC_CHAT_SESSIONS_DIR to a directory path.",
        )
 
 
def _validate_context_file(
    issues: list[ConfigIssue],
    env: Mapping[str, str],
    cwd: Path,
) -> None:
    if "OC_CHAT_CONTEXT_FILE" not in env:
        return
    raw = env.get("OC_CHAT_CONTEXT_FILE", "")
    if not raw.strip():
        return
    path = Path(raw)
    candidates = [path] if path.is_absolute() else [cwd / path]
    if not any(candidate.exists() and candidate.is_file() for candidate in candidates):
        _add(
            issues,
            "warning",
            "OC_CHAT_CONTEXT_FILE",
            f"context file {raw!r} was not found.",
            "Unset OC_CHAT_CONTEXT_FILE or point it at a readable text file.",
        )
 
 
def _validate_color_env(issues: list[ConfigIssue], env: Mapping[str, str]) -> None:
    if "NO_COLOR" in env and "FORCE_COLOR" in env:
        _add(
            issues,
            "warning",
            "NO_COLOR/FORCE_COLOR",
            "both are set; FORCE_COLOR takes precedence.",
            "Unset one of them to make terminal color behavior explicit.",
        )
 
 
def validate_runtime_config(
    *,
    env: Mapping[str, str] | None = None,
    provider_name: str | None = None,
    provider_names: Sequence[str] | None = None,
    provider: "ProviderPlugin | None" = None,
    sessions_dir: str | None = None,
    cwd: Path | None = None,
) -> ConfigValidationResult:
    """Validate runtime config and return structured diagnostics."""
    func = "validate_runtime_config"
    env_map = os.environ if env is None else env
    selected_provider = (
        provider_name
        if provider_name is not None
        else env_map.get("OC_CHAT_PROVIDER", "deepseek")
    ).strip()
    root = Path.cwd() if cwd is None else cwd
 
    issues: list[ConfigIssue] = []
    _validate_provider_selection(issues, selected_provider, provider_names)
    if provider is not None:
        _validate_provider_env(issues, env_map, provider)
    _validate_endpoint_env(issues, env_map)
    _validate_sessions_dir(issues, env_map, sessions_dir)
    _validate_context_file(issues, env_map, root)
    _validate_color_env(issues, env_map)
 
    result = ConfigValidationResult(tuple(issues))
    _dbg(
        "OC_CHAT.MAIN",
        func,
        f"errors={len(result.errors)} warnings={len(result.warnings)}",
        3,
    )
    return result
 
 
__all__ = [
    "ConfigIssue",
    "ConfigValidationResult",
    "format_config_issue",
    "format_config_issues",
    "validate_runtime_config",
]