460 lines
17 KiB
Python
460 lines
17 KiB
Python
"""Configuration management for AegisGitea MCP.
|
|
|
|
All settings come from environment variables (case-insensitive) or an env file.
|
|
The env file is never trusted implicitly: see :func:`select_env_file`. Secrets
|
|
are ``SecretStr`` so they never appear in ``repr()`` or model dumps. Requirements that depend on how the server is started
|
|
(for example ``GITEA_TOKEN`` for stdio) are checked at serve time, not on import.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import os
|
|
import stat
|
|
import sys
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
from typing import Annotated, Any, Literal
|
|
from urllib.parse import urlsplit
|
|
|
|
from pydantic import Field, SecretStr, field_validator
|
|
from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
|
|
|
|
ALLOWED_LOG_LEVELS = frozenset({"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"})
|
|
|
|
#: Tool domains accepted by ``TOOLSETS`` (plus the ``all`` shortcut).
|
|
TOOL_DOMAINS = frozenset(
|
|
{
|
|
"admin",
|
|
"repo",
|
|
"issue",
|
|
"pull",
|
|
"release",
|
|
"wiki",
|
|
"org",
|
|
"team",
|
|
"user",
|
|
"actions",
|
|
"hook",
|
|
"package",
|
|
"notification",
|
|
"misc",
|
|
}
|
|
)
|
|
|
|
#: Granular Gitea OAuth scopes. Gitea grants full access for unknown scope
|
|
#: strings, so anything outside this set is rejected up front.
|
|
GITEA_SCOPE_CATEGORIES = (
|
|
"activitypub",
|
|
"admin",
|
|
"issue",
|
|
"misc",
|
|
"notification",
|
|
"organization",
|
|
"package",
|
|
"repository",
|
|
"user",
|
|
)
|
|
GITEA_SCOPES = frozenset(
|
|
f"{access}:{category}" for access in ("read", "write") for category in GITEA_SCOPE_CATEGORIES
|
|
)
|
|
|
|
DEFAULT_OAUTH_SCOPES = ",".join(f"write:{category}" for category in GITEA_SCOPE_CATEGORIES)
|
|
|
|
_MIN_SECRET_CHARS = 32
|
|
_LOOPBACK_NAMES = frozenset({"localhost"})
|
|
|
|
|
|
def default_data_dir() -> Path:
|
|
"""Return the default state directory (``~/.local/state/aegis-gitea-mcp``)."""
|
|
return Path("~/.local/state/aegis-gitea-mcp").expanduser()
|
|
|
|
|
|
def is_loopback_host(host: str) -> bool:
|
|
"""Return True when ``host`` is localhost or a loopback IP (127.0.0.0/8, ::1)."""
|
|
candidate = host.strip().strip("[]").lower()
|
|
if candidate in _LOOPBACK_NAMES:
|
|
return True
|
|
try:
|
|
return ipaddress.ip_address(candidate).is_loopback
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _split_list(value: Any) -> Any:
|
|
"""Split a comma or whitespace separated string into a list of stripped items."""
|
|
if isinstance(value, str):
|
|
return [item for item in value.replace(",", " ").split() if item]
|
|
return value
|
|
|
|
|
|
def _normalize_http_url(value: str, *, field: str) -> str:
|
|
"""Validate an http(s) URL and return it without query, fragment or trailing slash."""
|
|
candidate = value.strip()
|
|
parts = urlsplit(candidate)
|
|
if parts.scheme not in {"http", "https"}:
|
|
raise ValueError(f"{field} must start with http:// or https://")
|
|
if not parts.hostname:
|
|
raise ValueError(f"{field} must include a host")
|
|
if parts.username is not None or parts.password is not None:
|
|
raise ValueError(f"{field} must not contain credentials (user:password@)")
|
|
if parts.query or parts.fragment or "?" in candidate or "#" in candidate:
|
|
raise ValueError(f"{field} must not contain a query string or fragment")
|
|
try:
|
|
parts.port # noqa: B018 (accessing .port validates it)
|
|
except ValueError as exc:
|
|
raise ValueError(f"{field} has an invalid port") from exc
|
|
return candidate.rstrip("/")
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
"""Application settings loaded from environment variables."""
|
|
|
|
model_config = SettingsConfigDict(
|
|
# No implicit env_file: get_settings() decides which file to load (SEC-04).
|
|
env_file_encoding="utf-8",
|
|
case_sensitive=False,
|
|
extra="ignore",
|
|
)
|
|
|
|
# Gitea
|
|
gitea_url: str = Field(..., description="Base URL of the Gitea instance")
|
|
gitea_token: SecretStr | None = Field(
|
|
default=None,
|
|
description="Gitea token for stdio and admin-token mode (checked at serve time)",
|
|
)
|
|
|
|
# MCP transport
|
|
mcp_transport: Literal["stdio", "http"] = Field(default="stdio")
|
|
mcp_host: str = Field(default="127.0.0.1", min_length=1)
|
|
mcp_port: int = Field(default=8080, ge=1, le=65535)
|
|
mcp_path: str = Field(default="/mcp")
|
|
mcp_bearer_token: SecretStr | None = Field(default=None)
|
|
public_base_url: str | None = Field(default=None)
|
|
allowed_hosts: Annotated[list[str], NoDecode] = Field(default_factory=list)
|
|
forwarded_allow_ips: str = Field(default="127.0.0.1")
|
|
|
|
# Per-user OAuth (HTTP only)
|
|
oauth_enabled: bool = Field(default=False)
|
|
gitea_oauth_client_id: str | None = Field(default=None)
|
|
gitea_oauth_client_secret: SecretStr | None = Field(default=None)
|
|
token_encryption_key: SecretStr | None = Field(default=None)
|
|
oauth_scopes: Annotated[list[str], NoDecode] = Field(
|
|
default_factory=lambda: DEFAULT_OAUTH_SCOPES.split(",")
|
|
)
|
|
|
|
# State
|
|
data_dir: Path = Field(default_factory=default_data_dir)
|
|
audit_log_path: Path | None = Field(default=None)
|
|
|
|
# Tool exposure
|
|
read_only: bool = Field(default=False)
|
|
toolsets: Annotated[list[str], NoDecode] = Field(default_factory=lambda: ["all"])
|
|
disabled_tools: Annotated[list[str], NoDecode] = Field(default_factory=list)
|
|
|
|
# Limits and safety
|
|
request_timeout_seconds: int = Field(default=30, ge=1, le=600)
|
|
max_tool_response_items: int = Field(default=200, ge=1)
|
|
max_tool_response_chars: int = Field(default=60_000, ge=1)
|
|
max_binary_bytes: int = Field(default=5_242_880, ge=1)
|
|
secret_detection_mode: Literal["off", "mask", "block"] = Field(default="mask")
|
|
log_level: str = Field(default="INFO")
|
|
expose_error_details: bool = Field(default=False)
|
|
rate_limit_per_minute: int = Field(default=120, ge=1)
|
|
|
|
# Validators
|
|
|
|
@field_validator(
|
|
"gitea_token",
|
|
"mcp_bearer_token",
|
|
"public_base_url",
|
|
"gitea_oauth_client_id",
|
|
"gitea_oauth_client_secret",
|
|
"token_encryption_key",
|
|
"audit_log_path",
|
|
mode="before",
|
|
)
|
|
@classmethod
|
|
def _blank_is_none(cls, value: Any) -> Any:
|
|
"""Treat an empty or whitespace-only value (``KEY=`` in .env) as unset."""
|
|
if isinstance(value, str) and not value.strip():
|
|
return None
|
|
return value
|
|
|
|
@field_validator("gitea_url")
|
|
@classmethod
|
|
def _validate_gitea_url(cls, value: str) -> str:
|
|
"""Normalize the Gitea base URL; a trailing ``/api/v1`` is accepted and removed."""
|
|
normalized = _normalize_http_url(value, field="GITEA_URL")
|
|
if normalized.endswith("/api/v1"):
|
|
normalized = normalized[: -len("/api/v1")].rstrip("/")
|
|
if not urlsplit(normalized).hostname: # pragma: no cover - defensive
|
|
raise ValueError("GITEA_URL must include a host")
|
|
return normalized
|
|
|
|
@field_validator("public_base_url")
|
|
@classmethod
|
|
def _validate_public_base_url(cls, value: str | None) -> str | None:
|
|
"""Require https for the public URL unless it points at a loopback host."""
|
|
if value is None:
|
|
return None
|
|
normalized = _normalize_http_url(value, field="PUBLIC_BASE_URL")
|
|
parts = urlsplit(normalized)
|
|
if parts.scheme != "https" and not is_loopback_host(parts.hostname or ""):
|
|
raise ValueError("PUBLIC_BASE_URL must use https unless the host is loopback")
|
|
return normalized
|
|
|
|
@field_validator("forwarded_allow_ips", mode="before")
|
|
@classmethod
|
|
def _validate_forwarded_allow_ips(cls, value: Any) -> Any:
|
|
"""Accept ``*`` or a comma list of IP addresses and networks (blank means the default)."""
|
|
if not isinstance(value, str):
|
|
return value
|
|
items = [item.strip() for item in value.split(",") if item.strip()]
|
|
if not items:
|
|
return "127.0.0.1"
|
|
if "*" in items and len(items) > 1:
|
|
raise ValueError("FORWARDED_ALLOW_IPS: '*' cannot be combined with other entries")
|
|
for item in items:
|
|
if item == "*":
|
|
continue
|
|
try:
|
|
ipaddress.ip_network(item, strict=False)
|
|
except ValueError:
|
|
raise ValueError(
|
|
"FORWARDED_ALLOW_IPS must be '*' or a comma list of IP addresses or networks"
|
|
) from None
|
|
return ",".join(items)
|
|
|
|
@field_validator("mcp_path")
|
|
@classmethod
|
|
def _validate_mcp_path(cls, value: str) -> str:
|
|
"""Require an absolute path without query or trailing slash."""
|
|
candidate = value.strip()
|
|
if not candidate.startswith("/") or candidate.startswith("//"):
|
|
raise ValueError("MCP_PATH must start with a single '/'")
|
|
if "?" in candidate or "#" in candidate or " " in candidate:
|
|
raise ValueError("MCP_PATH must be a plain path")
|
|
candidate = candidate.rstrip("/")
|
|
if not candidate:
|
|
raise ValueError("MCP_PATH must not be '/'")
|
|
return candidate
|
|
|
|
@field_validator("mcp_bearer_token", "token_encryption_key")
|
|
@classmethod
|
|
def _validate_min_secret_length(cls, value: SecretStr | None, info: Any) -> SecretStr | None:
|
|
"""Reject short shared secrets."""
|
|
if value is not None and len(value.get_secret_value()) < _MIN_SECRET_CHARS:
|
|
raise ValueError(
|
|
f"{info.field_name.upper()} must be at least {_MIN_SECRET_CHARS} chars"
|
|
)
|
|
return value
|
|
|
|
@field_validator("allowed_hosts", "oauth_scopes", "toolsets", "disabled_tools", mode="before")
|
|
@classmethod
|
|
def _parse_list(cls, value: Any) -> Any:
|
|
"""Accept comma or whitespace separated strings for list settings."""
|
|
return _split_list(value)
|
|
|
|
@field_validator("oauth_scopes")
|
|
@classmethod
|
|
def _validate_oauth_scopes(cls, value: list[str]) -> list[str]:
|
|
"""Accept only granular Gitea scopes (unknown scopes make Gitea grant full access)."""
|
|
unknown = sorted({scope for scope in value if scope not in GITEA_SCOPES})
|
|
if unknown:
|
|
raise ValueError(
|
|
f"OAUTH_SCOPES contains unknown scope(s): {', '.join(unknown)}. "
|
|
"Gitea grants full access for unknown scopes, so only granular "
|
|
"read:/write: scopes are accepted"
|
|
)
|
|
if not value:
|
|
raise ValueError("OAUTH_SCOPES must not be empty")
|
|
return list(dict.fromkeys(value))
|
|
|
|
@field_validator("toolsets")
|
|
@classmethod
|
|
def _validate_toolsets(cls, value: list[str]) -> list[str]:
|
|
"""Accept known tool domains and ``all``."""
|
|
lowered = [item.lower() for item in value]
|
|
unknown = sorted({item for item in lowered if item != "all" and item not in TOOL_DOMAINS})
|
|
if unknown:
|
|
allowed = ", ".join(sorted(TOOL_DOMAINS | {"all"}))
|
|
raise ValueError(
|
|
f"TOOLSETS contains unknown domain(s): {', '.join(unknown)}; use {allowed}"
|
|
)
|
|
if not lowered:
|
|
raise ValueError("TOOLSETS must not be empty (use 'all')")
|
|
return list(dict.fromkeys(lowered))
|
|
|
|
@field_validator("disabled_tools")
|
|
@classmethod
|
|
def _normalize_disabled_tools(cls, value: list[str]) -> list[str]:
|
|
"""Lowercase tool names and drop duplicates."""
|
|
return list(dict.fromkeys(item.lower() for item in value))
|
|
|
|
@field_validator("allowed_hosts")
|
|
@classmethod
|
|
def _normalize_allowed_hosts(cls, value: list[str]) -> list[str]:
|
|
"""Drop duplicates, keep order."""
|
|
return list(dict.fromkeys(value))
|
|
|
|
@field_validator("log_level")
|
|
@classmethod
|
|
def _validate_log_level(cls, value: str) -> str:
|
|
"""Normalize and validate the logging level."""
|
|
upper = value.strip().upper()
|
|
if upper not in ALLOWED_LOG_LEVELS:
|
|
raise ValueError(f"LOG_LEVEL must be one of {sorted(ALLOWED_LOG_LEVELS)}")
|
|
return upper
|
|
|
|
@field_validator("data_dir", "audit_log_path")
|
|
@classmethod
|
|
def _expand_user(cls, value: Path | None) -> Path | None:
|
|
"""Expand a leading ``~`` in paths."""
|
|
return value.expanduser() if value is not None else None
|
|
|
|
# Derived values and helpers
|
|
|
|
@property
|
|
def gitea_api_url(self) -> str:
|
|
"""Base URL of the Gitea REST API (``<GITEA_URL>/api/v1``)."""
|
|
return f"{self.gitea_url}/api/v1"
|
|
|
|
@property
|
|
def resolved_audit_log_path(self) -> Path:
|
|
"""Audit log path, defaulting to ``DATA_DIR/audit.log``."""
|
|
return self.audit_log_path or (self.data_dir / "audit.log")
|
|
|
|
def is_loopback_bind(self) -> bool:
|
|
"""Return True when ``MCP_HOST`` is a loopback address (127.0.0.0/8, ::1, localhost)."""
|
|
return is_loopback_host(self.mcp_host)
|
|
|
|
def oauth_config_errors(self) -> list[str]:
|
|
"""List missing settings required by OAuth mode (empty when OAuth is off or complete)."""
|
|
if not self.oauth_enabled:
|
|
return []
|
|
errors: list[str] = []
|
|
if not self.gitea_oauth_client_id:
|
|
errors.append("GITEA_OAUTH_CLIENT_ID is required when OAUTH_ENABLED=true")
|
|
if self.gitea_oauth_client_secret is None:
|
|
errors.append("GITEA_OAUTH_CLIENT_SECRET is required when OAUTH_ENABLED=true")
|
|
if self.token_encryption_key is None:
|
|
errors.append("TOKEN_ENCRYPTION_KEY is required when OAUTH_ENABLED=true")
|
|
if not self.public_base_url:
|
|
errors.append("PUBLIC_BASE_URL is required when OAUTH_ENABLED=true")
|
|
return errors
|
|
|
|
def http_auth_configured(self) -> bool:
|
|
"""Return True when the HTTP endpoint has a bearer token or a complete OAuth setup."""
|
|
if self.mcp_bearer_token is not None:
|
|
return True
|
|
return self.oauth_enabled and not self.oauth_config_errors()
|
|
|
|
|
|
#: Environment variable naming an env file (same as ``--env-file``).
|
|
ENV_FILE_VARIABLE = "AEGIS_ENV_FILE"
|
|
DEFAULT_ENV_FILE = ".env"
|
|
|
|
|
|
class EnvFileError(ValueError):
|
|
"""An explicitly requested env file cannot be used (the message is safe to print)."""
|
|
|
|
|
|
def env_file_problem(path: Path) -> str | None:
|
|
"""Return why the implicit env file ``path`` is not trusted, or None when it may be loaded.
|
|
|
|
Trusted means: a regular file (not a symlink) owned by the current user that
|
|
group and others cannot access (mode 0600, which ``setup`` writes). A missing
|
|
file is not a problem here.
|
|
"""
|
|
try:
|
|
info = os.lstat(path)
|
|
except FileNotFoundError:
|
|
return None
|
|
except OSError as exc:
|
|
return f"it cannot be inspected ({exc.strerror or type(exc).__name__})"
|
|
if stat.S_ISLNK(info.st_mode):
|
|
return "it is a symlink"
|
|
if not stat.S_ISREG(info.st_mode):
|
|
return "it is not a regular file"
|
|
if info.st_uid != getattr(os, "geteuid", lambda: info.st_uid)():
|
|
return "it is owned by another user"
|
|
mode = stat.S_IMODE(info.st_mode)
|
|
if mode & 0o077:
|
|
return f"its permissions are {mode:04o} (group or others have access)"
|
|
return None
|
|
|
|
|
|
def select_env_file(
|
|
explicit: str | Path | None = None,
|
|
*,
|
|
environ: Mapping[str, str] | None = None,
|
|
directory: Path | None = None,
|
|
) -> tuple[Path | None, str | None]:
|
|
"""Choose the env file to load; returns ``(path, warning)``.
|
|
|
|
Order: ``explicit`` (``--env-file``), then ``AEGIS_ENV_FILE``, then ``./.env``.
|
|
A file named explicitly is the operator's choice and is used as is. The
|
|
implicit ``./.env`` is loaded only when :func:`env_file_problem` finds nothing
|
|
wrong; otherwise it is ignored and ``warning`` says why and how to fix it. A
|
|
working directory is not trustworthy by itself: clients start stdio servers
|
|
inside the project (possibly a cloned repository) they are working on, and a
|
|
planted ``.env`` could switch off redaction or audit, or redirect the token.
|
|
Real environment variables always take precedence over the file.
|
|
|
|
Raises:
|
|
EnvFileError: An explicitly named file is missing or not a regular file.
|
|
"""
|
|
env = os.environ if environ is None else environ
|
|
named = explicit or env.get(ENV_FILE_VARIABLE, "").strip() or None
|
|
if named is not None:
|
|
path = Path(named).expanduser()
|
|
try:
|
|
regular = stat.S_ISREG(os.stat(path).st_mode)
|
|
except OSError:
|
|
regular = False
|
|
if not regular:
|
|
raise EnvFileError(f"env file {path} does not exist or is not a regular file")
|
|
return path, None
|
|
path = (directory or Path.cwd()) / DEFAULT_ENV_FILE
|
|
problem = env_file_problem(path)
|
|
if problem is not None:
|
|
return None, (
|
|
f"ignoring ./{DEFAULT_ENV_FILE}: {problem}. Run 'chmod 600 {DEFAULT_ENV_FILE}' "
|
|
f"(and make sure you own it), or pass --env-file PATH (or set {ENV_FILE_VARIABLE}) "
|
|
"to load a file explicitly"
|
|
)
|
|
return (path if path.exists() else None), None
|
|
|
|
|
|
_settings: Settings | None = None
|
|
_env_file_choice: str | Path | None = None
|
|
|
|
|
|
def set_env_file(path: str | Path | None) -> None:
|
|
"""Choose the env file for the next :func:`get_settings` call (``--env-file``)."""
|
|
global _env_file_choice
|
|
_env_file_choice = path
|
|
reset_settings()
|
|
|
|
|
|
def get_settings() -> Settings:
|
|
"""Get or create the global settings instance.
|
|
|
|
Raises:
|
|
EnvFileError: An explicitly named env file cannot be used.
|
|
"""
|
|
global _settings
|
|
if _settings is None:
|
|
path, warning = select_env_file(_env_file_choice)
|
|
if warning:
|
|
print(f"aegis-gitea-mcp: warning: {warning}", file=sys.stderr)
|
|
_settings = Settings(_env_file=path) # type: ignore[call-arg]
|
|
return _settings
|
|
|
|
|
|
def reset_settings() -> None:
|
|
"""Reset the global settings instance (primarily for tests)."""
|
|
global _settings
|
|
_settings = None
|