From b82e2866cbb26ad6fa1865e0ce3a1d57a440b5cf Mon Sep 17 00:00:00 2001 From: latte Date: Wed, 14 Oct 2026 15:40:37 +0200 Subject: [PATCH] feat: new settings, CLI skeleton and interface skeletons Adds the pydantic-settings Settings with validated GITEA_URL, OAuth scopes and toolsets, the argparse CLI skeleton, catalog.model, identity and GiteaAPIError. Test fixtures now isolate env, cwd and state paths in tmp. --- scripts/validate_audit_log.py | 15 +- src/aegis_gitea_mcp/__init__.py | 2 +- src/aegis_gitea_mcp/audit.py | 12 +- src/aegis_gitea_mcp/catalog/__init__.py | 5 + src/aegis_gitea_mcp/catalog/model.py | 142 ++++++ src/aegis_gitea_mcp/cli.py | 55 ++ src/aegis_gitea_mcp/config.py | 650 +++++++++++------------- src/aegis_gitea_mcp/errors.py | 86 +++- src/aegis_gitea_mcp/identity.py | 34 ++ src/aegis_gitea_mcp/rate_limit.py | 2 +- src/aegis_gitea_mcp/security.py | 5 +- tests/conftest.py | 87 +--- tests/test_audit.py | 38 +- tests/test_catalog_model.py | 117 +++++ tests/test_cli.py | 37 ++ tests/test_config.py | 463 +++++++++++++---- tests/test_errors.py | 73 +++ tests/test_identity.py | 31 ++ tests/test_logging_utils.py | 52 +- tests/test_no_host_paths.py | 19 + tests/test_rate_limit.py | 55 ++ tests/test_request_context.py | 35 ++ tests/test_response_limits.py | 46 ++ tests/test_validate_audit_script.py | 56 ++ 24 files changed, 1512 insertions(+), 605 deletions(-) create mode 100644 src/aegis_gitea_mcp/catalog/__init__.py create mode 100644 src/aegis_gitea_mcp/catalog/model.py create mode 100644 src/aegis_gitea_mcp/cli.py create mode 100644 src/aegis_gitea_mcp/identity.py create mode 100644 tests/test_catalog_model.py create mode 100644 tests/test_cli.py create mode 100644 tests/test_errors.py create mode 100644 tests/test_identity.py create mode 100644 tests/test_no_host_paths.py create mode 100644 tests/test_rate_limit.py create mode 100644 tests/test_request_context.py create mode 100644 tests/test_response_limits.py create mode 100644 tests/test_validate_audit_script.py diff --git a/scripts/validate_audit_log.py b/scripts/validate_audit_log.py index 85215e8..bfa1c2d 100755 --- a/scripts/validate_audit_log.py +++ b/scripts/validate_audit_log.py @@ -4,10 +4,21 @@ from __future__ import annotations import argparse -import sys +import os from pathlib import Path from aegis_gitea_mcp.audit import validate_audit_log_integrity +from aegis_gitea_mcp.config import default_data_dir + + +def default_audit_path() -> Path: + """Resolve the audit log path the server would use (env first, then DATA_DIR).""" + explicit = os.environ.get("AUDIT_LOG_PATH", "").strip() + if explicit: + return Path(explicit).expanduser() + data_dir = os.environ.get("DATA_DIR", "").strip() + base = Path(data_dir).expanduser() if data_dir else default_data_dir() + return base / "audit.log" def parse_args() -> argparse.Namespace: @@ -16,7 +27,7 @@ def parse_args() -> argparse.Namespace: parser.add_argument( "--path", type=Path, - default=Path("/var/log/aegis-mcp/audit.log"), + default=default_audit_path(), help="Path to audit log file", ) return parser.parse_args() diff --git a/src/aegis_gitea_mcp/__init__.py b/src/aegis_gitea_mcp/__init__.py index 46269a1..9a105c3 100644 --- a/src/aegis_gitea_mcp/__init__.py +++ b/src/aegis_gitea_mcp/__init__.py @@ -1,3 +1,3 @@ """AegisGitea MCP - Security-first MCP gateway for self-hosted Gitea.""" -__version__ = "0.2.0" +__version__ = "1.0.0.dev0" diff --git a/src/aegis_gitea_mcp/audit.py b/src/aegis_gitea_mcp/audit.py index 4f0e558..761910d 100644 --- a/src/aegis_gitea_mcp/audit.py +++ b/src/aegis_gitea_mcp/audit.py @@ -2,6 +2,7 @@ from __future__ import annotations +import contextlib import hashlib import json import threading @@ -31,10 +32,11 @@ class AuditLogger: log_path: Path to audit log file (defaults to config value). """ self.settings = get_settings() - self.log_path = log_path or self.settings.audit_log_path + self.log_path = log_path or self.settings.resolved_audit_log_path self.log_path.parent.mkdir(parents=True, exist_ok=True) self._lock = threading.Lock() - self._log_file = open(self.log_path, "a+", encoding="utf-8") + # Held open for the process lifetime; closed in close(). + self._log_file = open(self.log_path, "a+", encoding="utf-8") # noqa: SIM115 self._last_hash = self._read_last_hash() def _read_last_hash(self) -> str: @@ -82,7 +84,7 @@ class AuditLogger: payload["correlation_id"] = correlation_id # Security decision: sanitize all audit payloads before persistence. - mode = "mask" if self.settings.secret_detection_mode != "off" else "off" + mode = "mask" if self.settings.secret_detection_mode != "off" else "off" # noqa: S105 safe_payload = payload if mode == "off" else sanitize_data(payload, mode=mode) base_entry: dict[str, Any] = { @@ -108,10 +110,8 @@ class AuditLogger: def close(self) -> None: """Close open audit log resources.""" - try: + with contextlib.suppress(OSError): self._log_file.close() - except Exception: - pass def log_tool_invocation( self, diff --git a/src/aegis_gitea_mcp/catalog/__init__.py b/src/aegis_gitea_mcp/catalog/__init__.py new file mode 100644 index 0000000..e4ca1b4 --- /dev/null +++ b/src/aegis_gitea_mcp/catalog/__init__.py @@ -0,0 +1,5 @@ +"""Generated Gitea tool catalog (data model now, generator and loader in chunk B2).""" + +from aegis_gitea_mcp.catalog.model import Operation, Parameter, ParamLocation + +__all__ = ["Operation", "ParamLocation", "Parameter"] diff --git a/src/aegis_gitea_mcp/catalog/model.py b/src/aegis_gitea_mcp/catalog/model.py new file mode 100644 index 0000000..c507589 --- /dev/null +++ b/src/aegis_gitea_mcp/catalog/model.py @@ -0,0 +1,142 @@ +"""Data model for the generated tool catalog. + +One :class:`Operation` per Gitea swagger operation. The catalog generator +(chunk B2) writes ``catalog/tools.json`` from these types, the executor +(chunk B3) builds HTTP requests from them, and the server (chunk B4) turns them +into MCP tools. The model is plain data so it round-trips through JSON. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Literal, get_args + +ParamLocation = Literal["path", "query", "body", "formData"] +_LOCATIONS = frozenset(get_args(ParamLocation)) + + +@dataclass(frozen=True) +class Parameter: + """One argument of a Gitea operation. + + Attributes: + name: Parameter name as Gitea knows it. + arg_name: Tool argument name (identical except the JSON body, which is ``body``). + location: Where the value goes in the HTTP request. + required: Whether the tool argument is required. + schema: JSON-schema fragment describing the argument. + is_file: True for a ``formData`` file; the tool takes base64 content. + """ + + name: str + arg_name: str + location: ParamLocation + required: bool + schema: dict[str, Any] = field(default_factory=dict) + is_file: bool = False + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> Parameter: + """Build a parameter from its JSON form, validating the location.""" + location = data["location"] + if location not in _LOCATIONS: + raise ValueError(f"invalid parameter location: {location!r}") + return cls( + name=str(data["name"]), + arg_name=str(data["arg_name"]), + location=location, + required=bool(data["required"]), + schema=dict(data["schema"]), + is_file=bool(data.get("is_file", False)), + ) + + def to_dict(self) -> dict[str, Any]: + """Return the JSON-serializable form.""" + return { + "name": self.name, + "arg_name": self.arg_name, + "location": self.location, + "required": self.required, + "schema": self.schema, + "is_file": self.is_file, + } + + +@dataclass(frozen=True) +class Operation: + """One Gitea API operation exposed as an MCP tool. + + Attributes: + tool_name: ``gitea__``, lowercase snake_case, at most 64 chars. + domain: Tool domain (admin, repo, issue, pull, ...). + operation_id: Swagger operationId. + method: Upper-case HTTP method. + path: Path relative to ``/api/v1``, e.g. ``/repos/{owner}/{repo}/issues``. + summary: Short swagger summary. + description: Longer swagger description (may be empty). + parameters: All parameters in swagger order. + input_schema: Complete JSON input schema of the tool. + consumes: Request content types. + produces: Response content types. + read_only: Maps to the MCP ``readOnlyHint`` annotation. + destructive: Maps to ``destructiveHint``. + idempotent: Maps to ``idempotentHint``. + paginated: True when the operation supports ``page``/``limit``. + """ + + tool_name: str + domain: str + operation_id: str + method: str + path: str + summary: str + description: str + parameters: tuple[Parameter, ...] + input_schema: dict[str, Any] + consumes: tuple[str, ...] + produces: tuple[str, ...] + read_only: bool + destructive: bool + idempotent: bool + paginated: bool + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> Operation: + """Build an operation from its JSON form (as written by ``to_dict``).""" + return cls( + tool_name=str(data["tool_name"]), + domain=str(data["domain"]), + operation_id=str(data["operation_id"]), + method=str(data["method"]).upper(), + path=str(data["path"]), + summary=str(data.get("summary", "")), + description=str(data.get("description", "")), + parameters=tuple(Parameter.from_dict(item) for item in data["parameters"]), + input_schema=dict(data["input_schema"]), + consumes=tuple(str(item) for item in data.get("consumes", ())), + produces=tuple(str(item) for item in data.get("produces", ())), + read_only=bool(data["read_only"]), + destructive=bool(data["destructive"]), + idempotent=bool(data["idempotent"]), + paginated=bool(data["paginated"]), + ) + + def to_dict(self) -> dict[str, Any]: + """Return the JSON-serializable form (tuples become lists).""" + return { + "tool_name": self.tool_name, + "domain": self.domain, + "operation_id": self.operation_id, + "method": self.method, + "path": self.path, + "summary": self.summary, + "description": self.description, + "parameters": [parameter.to_dict() for parameter in self.parameters], + "input_schema": self.input_schema, + "consumes": list(self.consumes), + "produces": list(self.produces), + "read_only": self.read_only, + "destructive": self.destructive, + "idempotent": self.idempotent, + "paginated": self.paginated, + } diff --git a/src/aegis_gitea_mcp/cli.py b/src/aegis_gitea_mcp/cli.py new file mode 100644 index 0000000..0626802 --- /dev/null +++ b/src/aegis_gitea_mcp/cli.py @@ -0,0 +1,55 @@ +"""Command line entry point (``aegis-gitea-mcp``). + +Subcommands: ``serve`` (default), ``setup`` and ``version``. ``serve`` and +``setup`` are filled in by later chunks. +""" + +from __future__ import annotations + +import argparse +import sys +from collections.abc import Sequence + +from aegis_gitea_mcp import __version__ + + +def build_parser() -> argparse.ArgumentParser: + """Build the argument parser.""" + parser = argparse.ArgumentParser( + prog="aegis-gitea-mcp", + description="MCP server exposing the Gitea API over stdio or streamable HTTP.", + ) + subparsers = parser.add_subparsers(dest="command") + + serve = subparsers.add_parser("serve", help="run the MCP server (default)") + serve.add_argument( + "--transport", + choices=("stdio", "http"), + default=None, + help="transport to serve (default: MCP_TRANSPORT, else stdio)", + ) + + subparsers.add_parser("setup", help="interactive setup against a Gitea instance") + subparsers.add_parser("version", help="print the version and exit") + return parser + + +def _not_implemented(command: str) -> int: + """Report an unbuilt command and return the usage-error exit code.""" + print(f"aegis-gitea-mcp {command}: not implemented yet", file=sys.stderr) + return 2 + + +def main(argv: Sequence[str] | None = None) -> int: + """Run the CLI and return the process exit code.""" + args = build_parser().parse_args(argv) + command = args.command or "serve" + + if command == "version": + print(__version__) + return 0 + return _not_implemented(command) + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/src/aegis_gitea_mcp/config.py b/src/aegis_gitea_mcp/config.py index eb12736..c2d94f2 100644 --- a/src/aegis_gitea_mcp/config.py +++ b/src/aegis_gitea_mcp/config.py @@ -1,15 +1,106 @@ -"""Configuration management for AegisGitea MCP server.""" +"""Configuration management for AegisGitea MCP. + +All settings come from environment variables (case-insensitive) or a ``.env`` +file in the working directory. 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 from pathlib import Path +from typing import Annotated, Any, Literal +from urllib.parse import urlsplit -from pydantic import Field, HttpUrl, field_validator, model_validator -from pydantic_settings import BaseSettings, SettingsConfigDict +from pydantic import Field, SecretStr, field_validator +from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict -_ALLOWED_LOG_LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"} -_ALLOWED_SECRET_MODES = {"off", "mask", "block"} -_ALLOWED_ENVIRONMENTS = {"development", "staging", "production", "test"} +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): @@ -20,392 +111,231 @@ class Settings(BaseSettings): env_file_encoding="utf-8", case_sensitive=False, extra="ignore", - env_parse_none_str="null", ) - # Runtime environment - environment: str = Field( - default="production", - description="Runtime environment name", - ) - - # Gitea configuration - gitea_url: HttpUrl = Field(..., description="Base URL of the Gitea instance") - gitea_token: str = Field( - default="", - description=("Deprecated shared bot token. Not used for MCP tool execution in OAuth mode."), - ) - - # MCP server configuration - mcp_host: str = Field( - default="127.0.0.1", - description="Host interface to bind MCP server to", - ) - mcp_port: int = Field(default=8080, description="Port to bind MCP server to", ge=1, le=65535) - allow_insecure_bind: bool = Field( - default=False, - description="Allow binding to 0.0.0.0 (disabled by default for local hardening)", - ) - public_base_url: HttpUrl | None = Field( + # Gitea + gitea_url: str = Field(..., description="Base URL of the Gitea instance") + gitea_token: SecretStr | None = Field( default=None, - description=( - "Public externally-reachable base URL for this MCP server. " - "When set, OAuth metadata endpoints use this URL for absolute links." - ), + description="Gitea token for stdio and admin-token mode (checked at serve time)", ) - # Logging and observability - log_level: str = Field(default="INFO", description="Application logging level") - audit_log_path: Path = Field( - default=Path("/var/log/aegis-mcp/audit.log"), - description="Path to tamper-evident audit log file", - ) - metrics_enabled: bool = Field(default=True, description="Enable Prometheus metrics endpoint") - expose_error_details: bool = Field( - default=False, - description="Return internal error details in API responses (disabled by default)", - ) - startup_validate_gitea: bool = Field( - default=True, - description="Validate Gitea connectivity during startup", + # 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) + + # 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(",") ) - # Security limits - max_file_size_bytes: int = Field( - default=1_048_576, - description="Maximum file size that can be read (bytes)", - ge=1, - ) - request_timeout_seconds: int = Field( - default=30, - description="Timeout for Gitea API requests (seconds)", - ge=1, - ) - rate_limit_per_minute: int = Field( - default=60, - description="Maximum requests per minute for a single IP", - ge=1, - ) - token_rate_limit_per_minute: int = Field( - default=120, - description="Maximum requests per minute per authenticated token", - ge=1, - ) - max_tool_response_items: int = Field( - default=200, - description="Maximum list items returned by a tool response", - ge=1, - ) - max_tool_response_chars: int = Field( - default=20_000, - description="Maximum characters returned in text fields", - ge=1, - ) - secret_detection_mode: str = Field( - default="mask", - description="Secret detection mode: off, mask, or block", - ) + # State + data_dir: Path = Field(default_factory=default_data_dir) + audit_log_path: Path | None = Field(default=None) - # OAuth2 configuration (for per-client Gitea authentication) - oauth_mode: bool = Field( - default=False, - description=( - "Enable per-user OAuth2 authentication mode. " - "When true, each client user authenticates with their own Gitea account. " - "GITEA_TOKEN and MCP_API_KEYS are not required in this mode." - ), - ) - gitea_oauth_client_id: str = Field( - default="", - description="Gitea OAuth2 application client ID (required when oauth_mode=true)", - ) - gitea_oauth_client_secret: str = Field( - default="", - description="Gitea OAuth2 application client secret (required when oauth_mode=true)", - ) - oauth_expected_audience: str = Field( - default="", - description=( - "Additional expected OIDC audience for access tokens. The canonical MCP " - "resource URL and the Gitea OAuth client id are always accepted; set this " - "to require an extra audience value." - ), - ) - oauth_cache_ttl_seconds: int = Field( - default=300, - description="OIDC discovery/JWKS cache TTL in seconds", - ge=30, - ) - oauth_resource_documentation: str = Field( - default="https://hiddenden.cafe/docs/mcp-gitea", - description="Public documentation URL for OAuth-protected MCP resource behavior", - ) - oauth_state_secret: str = Field( - default="", - description=( - "Server secret used to HMAC-sign the OAuth proxy state parameter. " - "Required when OAUTH_MODE=true so callback state is tamper-evident." - ), - ) - oauth_redirect_allowlist_raw: str = Field( - default="", - description=( - "Comma-separated additional allowed client redirect URIs for the OAuth " - "callback proxy. Claude's callback URLs and loopback URIs are always allowed." - ), - alias="OAUTH_REDIRECT_ALLOWLIST", - ) - dcr_enabled: bool = Field( - default=True, - description=( - "Enable RFC 7591 Dynamic Client Registration at /register. Claude's " - "connectors register dynamically; disable to require manual client_id/secret." - ), - ) - dcr_storage_path: Path = Field( - default=Path("/var/lib/aegis-mcp/dcr_clients.json"), - description="Path to the JSON file that persists dynamically registered clients", - ) - repo_authz_cache_ttl_seconds: int = Field( - default=60, - description="TTL (seconds) for cached per-user repository permission decisions", - ge=1, - ) + # 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) - # Authentication configuration - auth_enabled: bool = Field( - default=True, - description="Enable API key authentication (disable only in controlled testing)", - ) - mcp_api_keys_raw: str = Field( - default="", - description="Comma-separated API keys for MCP access", - alias="MCP_API_KEYS", - ) - max_auth_failures: int = Field( - default=5, - description="Maximum authentication failures before auth rate limiting", - ge=1, - ) - auth_failure_window: int = Field( - default=300, - description="Time window for counting auth failures (seconds)", - ge=1, - ) + # 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) - # Policy and write-mode configuration - policy_file_path: Path = Field( - default=Path("policy.yaml"), - description="Path to YAML authorization policy file", - ) - write_mode: bool = Field(default=False, description="Enable write-capable tools") - write_repository_whitelist_raw: str = Field( - default="", - description="Comma-separated repository whitelist for write mode (owner/repo)", - alias="WRITE_REPOSITORY_WHITELIST", - ) - write_allow_all_token_repos: bool = Field( - default=False, - description=( - "Allow write-mode operations on any repository the token can access. " - "Disabled by default." - ), - ) - # Raw API dispatch (gitea_request escape hatch) - raw_api_enabled: bool = Field( - default=True, - description="Enable the generic gitea_request raw API dispatch tool", - ) - raw_api_allow_sensitive: bool = Field( - default=False, - description=( - "Allow gitea_request to reach admin/credential endpoints " - "(/admin, *tokens*, *secrets*, *hooks*, *keys*, applications/oauth2, " - "runner registration tokens). Disabled by default." - ), - ) - automation_enabled: bool = Field( - default=False, - description="Enable automation endpoints and workflows", - ) - automation_scheduler_enabled: bool = Field( - default=False, - description="Enable built-in scheduled job loop", - ) - automation_stale_days: int = Field( - default=30, - description="Number of days before an issue is considered stale", - ge=1, - ) + # Validators - @field_validator("environment") + @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 validate_environment(cls, value: str) -> str: - """Validate deployment environment name.""" - normalized = value.strip().lower() - if normalized not in _ALLOWED_ENVIRONMENTS: - raise ValueError(f"environment must be one of {_ALLOWED_ENVIRONMENTS}") - return normalized - - @field_validator("log_level") - @classmethod - def validate_log_level(cls, value: str) -> str: - """Validate log level is one of the allowed values.""" - normalized = value.upper() - if normalized not in _ALLOWED_LOG_LEVELS: - raise ValueError(f"log_level must be one of {_ALLOWED_LOG_LEVELS}") - return normalized - - @field_validator("public_base_url", mode="before") - @classmethod - def normalize_public_base_url(cls, value: object) -> object: - """Treat empty PUBLIC_BASE_URL as unset.""" + 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_token") + @field_validator("gitea_url") @classmethod - def validate_token_not_empty(cls, value: str) -> str: - """Validate Gitea token is trimmed (empty string allowed for oauth_mode).""" - cleaned = value.strip() - if value and not cleaned: - raise ValueError("gitea_token cannot be whitespace-only") - return cleaned - - @field_validator("secret_detection_mode") - @classmethod - def validate_secret_detection_mode(cls, value: str) -> str: - """Validate secret detection behavior setting.""" - normalized = value.lower().strip() - if normalized not in _ALLOWED_SECRET_MODES: - raise ValueError(f"secret_detection_mode must be one of {_ALLOWED_SECRET_MODES}") + 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 - @model_validator(mode="after") - def validate_security_constraints(self) -> Settings: - """Validate cross-field security constraints.""" - parsed_keys: list[str] = [] - if self.mcp_api_keys_raw.strip(): - parsed_keys = [ - value.strip() for value in self.mcp_api_keys_raw.split(",") if value.strip() - ] - - object.__setattr__(self, "_mcp_api_keys", parsed_keys) - - write_repositories: list[str] = [] - if self.write_repository_whitelist_raw.strip(): - write_repositories = [ - value.strip() - for value in self.write_repository_whitelist_raw.split(",") - if value.strip() - ] - - for repository in write_repositories: - if "/" not in repository: - raise ValueError("WRITE_REPOSITORY_WHITELIST entries must be in owner/repo format") - - object.__setattr__(self, "_write_repository_whitelist", write_repositories) - - # Security decision: binding all interfaces requires explicit opt-in. - if self.mcp_host == "0.0.0.0" and not self.allow_insecure_bind: - raise ValueError( - "Binding to 0.0.0.0 is blocked by default. " - "Set ALLOW_INSECURE_BIND=true to explicitly permit this." - ) - - extra_redirect_uris: list[str] = [] - if self.oauth_redirect_allowlist_raw.strip(): - extra_redirect_uris = [ - value.strip() - for value in self.oauth_redirect_allowlist_raw.split(",") - if value.strip() - ] - object.__setattr__(self, "_oauth_redirect_allowlist", extra_redirect_uris) - - if self.oauth_mode: - # In OAuth mode, per-user Gitea tokens are used; no shared bot token or API keys needed. - if not self.gitea_oauth_client_id.strip(): - raise ValueError("GITEA_OAUTH_CLIENT_ID is required when OAUTH_MODE=true.") - if not self.gitea_oauth_client_secret.strip(): - raise ValueError("GITEA_OAUTH_CLIENT_SECRET is required when OAUTH_MODE=true.") - # The proxy state parameter carries the client's redirect_uri across the Gitea - # round-trip; it must be HMAC-signed, which requires a server-held secret. - if not self.oauth_state_secret.strip(): - raise ValueError( - "OAUTH_STATE_SECRET is required when OAUTH_MODE=true so the OAuth " - "proxy state parameter can be HMAC-signed and verified." - ) - # A short secret weakens the HMAC; require the same 32-char floor as API keys. - if len(self.oauth_state_secret.strip()) < 32: - raise ValueError( - "OAUTH_STATE_SECRET must be at least 32 characters long " - "(e.g. `openssl rand -hex 32`)." - ) - else: - # Standard API key mode: require bot token and at least one API key. - if not self.gitea_token.strip(): - raise ValueError("GITEA_TOKEN is required unless OAUTH_MODE=true.") - if self.auth_enabled and not parsed_keys: - raise ValueError( - "At least one API key must be configured when auth_enabled=True. " - "Set MCP_API_KEYS or disable auth explicitly for controlled testing." - ) - - # Enforce minimum key length to reduce brute-force success probability. - for key in parsed_keys: - if len(key) < 32: - raise ValueError("API keys must be at least 32 characters long") - - if self.write_mode and not self.write_allow_all_token_repos and not write_repositories: - raise ValueError( - "WRITE_MODE=true requires WRITE_REPOSITORY_WHITELIST to be configured " - "unless WRITE_ALLOW_ALL_TOKEN_REPOS=true" - ) - - return self - - @property - def mcp_api_keys(self) -> list[str]: - """Get parsed list of API keys.""" - return list(getattr(self, "_mcp_api_keys", [])) - - @property - def write_repository_whitelist(self) -> list[str]: - """Get parsed list of repositories allowed for write-mode operations.""" - return list(getattr(self, "_write_repository_whitelist", [])) - - @property - def oauth_redirect_allowlist(self) -> list[str]: - """Get parsed list of additional allowed client redirect URIs.""" - return list(getattr(self, "_oauth_redirect_allowlist", [])) - - @property - def gitea_base_url(self) -> str: - """Get Gitea base URL as normalized string.""" - return str(self.gitea_url).rstrip("/") - - @property - def public_base(self) -> str | None: - """Get normalized public base URL when explicitly configured.""" - if self.public_base_url is None: + @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 - return str(self.public_base_url).rstrip("/") + 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("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 (``/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() _settings: Settings | None = None def get_settings() -> Settings: - """Get or create global settings instance.""" + """Get or create the global settings instance.""" global _settings if _settings is None: - # Mypy limitation: BaseSettings loads from environment dynamically. _settings = Settings() # type: ignore[call-arg] return _settings def reset_settings() -> None: - """Reset global settings instance (primarily for testing).""" + """Reset the global settings instance (primarily for tests).""" global _settings _settings = None diff --git a/src/aegis_gitea_mcp/errors.py b/src/aegis_gitea_mcp/errors.py index 5219ccb..2ad4acd 100644 --- a/src/aegis_gitea_mcp/errors.py +++ b/src/aegis_gitea_mcp/errors.py @@ -1,25 +1,89 @@ -"""Transport-agnostic error types raised by the core. - -Core tool handlers and the authorization layer must not depend on the web stack -(FastAPI). They raise :class:`ToolError` carrying an advisory HTTP status code; -each transport adapter maps it to its own wire format (the HTTP adapter to -``fastapi.HTTPException``, the stdio adapter to an MCP error). This keeps the -core importable without FastAPI installed. -""" +"""Transport-agnostic error types raised by the core.""" from __future__ import annotations +import re + +from aegis_gitea_mcp.security import sanitize_data + +_MAX_MESSAGE_CHARS = 500 +_CREDENTIAL_PATTERN = re.compile( + r"(?i)\b(authorization|bearer|token|basic)\b[\s:=]+[A-Za-z0-9._~+/=-]{8,}" +) +_WHITESPACE = re.compile(r"\s+") + +_KIND_BY_STATUS = { + 400: "bad_request", + 401: "unauthorized", + 403: "forbidden", + 404: "not_found", + 405: "method_not_allowed", + 409: "conflict", + 410: "gone", + 412: "precondition_failed", + 413: "payload_too_large", + 422: "validation_failed", + 423: "locked", + 429: "rate_limited", +} + + +def kind_for_status(status: int) -> str: + """Map an HTTP status to a stable machine-readable error kind.""" + if status in _KIND_BY_STATUS: + return _KIND_BY_STATUS[status] + if 500 <= status <= 599: + return "server_error" + if 400 <= status <= 499: + return "client_error" + return "error" + + +def sanitize_message(message: str, max_chars: int = _MAX_MESSAGE_CHARS) -> str: + """Make an upstream error message safe to return to a model or log. + + Collapses whitespace, masks credentials and secret-like strings, and + truncates to ``max_chars``. + """ + collapsed = _WHITESPACE.sub(" ", message).strip() + redacted = _CREDENTIAL_PATTERN.sub(lambda m: f"{m.group(1)} [REDACTED]", collapsed) + masked = str(sanitize_data(redacted, mode="mask")) + if len(masked) > max_chars: + return masked[: max_chars - 3] + "..." + return masked + class ToolError(Exception): - """Error raised by a core tool handler or the authorization layer. + """Error raised by core code while executing a tool. Args: message: Human-readable, non-sensitive error detail. - status_code: Advisory HTTP status (e.g. 403 for denied). Adapters map - this to their transport; the stdio adapter only uses the message. + status_code: Advisory HTTP-style status. Transports map it to their own + wire format. """ def __init__(self, message: str, *, status_code: int = 400) -> None: super().__init__(message) self.status_code = status_code self.detail = message + + +class GiteaAPIError(ToolError): + """A Gitea API call failed. + + Attributes: + status: HTTP status returned by Gitea (0 when no response was received). + kind: Stable machine-readable category (``not_found``, ``forbidden``, ...). + message: Sanitized, truncated message safe to return to the client. + """ + + def __init__(self, status: int, message: str, *, kind: str | None = None) -> None: + safe_message = sanitize_message(message) + super().__init__(safe_message, status_code=status) + self.status = status + self.kind = kind or kind_for_status(status) + self.message = safe_message + + def to_dict(self) -> dict[str, object]: + """Return the ``error`` object of the ADR 0005 error envelope.""" + return {"status": self.status, "kind": self.kind, "message": self.message} diff --git a/src/aegis_gitea_mcp/identity.py b/src/aegis_gitea_mcp/identity.py new file mode 100644 index 0000000..b381ebb --- /dev/null +++ b/src/aegis_gitea_mcp/identity.py @@ -0,0 +1,34 @@ +"""Identity a tool call runs as. + +Admin-token mode uses the configured ``GITEA_TOKEN`` (stdio, or the static MCP +bearer on HTTP). OAuth mode uses the signed-in user's own Gitea token. The +identity travels with each call so the executor can pick the right token and the +result envelope and audit log can name who acted. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Literal + +IdentityMode = Literal["admin-token", "oauth"] + + +@dataclass(frozen=True) +class Identity: + """Who a tool call acts as. + + Attributes: + login: Gitea login when known (``None`` until resolved). + mode: ``admin-token`` or ``oauth``. + token: Gitea token for this identity. Excluded from ``repr`` and never + part of :meth:`metadata`. + """ + + login: str | None + mode: IdentityMode + token: str = field(repr=False) + + def metadata(self, sudo: str | None = None) -> dict[str, Any]: + """Return result and audit metadata: ``{"login", "mode", "sudo"}``, never the token.""" + return {"login": self.login, "mode": self.mode, "sudo": sudo} diff --git a/src/aegis_gitea_mcp/rate_limit.py b/src/aegis_gitea_mcp/rate_limit.py index e06c14e..7cd92c3 100644 --- a/src/aegis_gitea_mcp/rate_limit.py +++ b/src/aegis_gitea_mcp/rate_limit.py @@ -57,7 +57,7 @@ class RequestRateLimiter: settings = get_settings() self._audit = get_audit_logger() self._ip_limiter = SlidingWindowLimiter(settings.rate_limit_per_minute, 60) - self._token_limiter = SlidingWindowLimiter(settings.token_rate_limit_per_minute, 60) + self._token_limiter = SlidingWindowLimiter(settings.rate_limit_per_minute, 60) def check(self, client_ip: str, token: str | None) -> RateLimitDecision: """Evaluate request against IP and token limits. diff --git a/src/aegis_gitea_mcp/security.py b/src/aegis_gitea_mcp/security.py index fafb0f8..2c3161c 100644 --- a/src/aegis_gitea_mcp/security.py +++ b/src/aegis_gitea_mcp/security.py @@ -55,10 +55,7 @@ def detect_secrets(text: str) -> list[SecretMatch]: matches: list[SecretMatch] = [] for secret_type, pattern in _SECRET_PATTERNS: for found in pattern.findall(text): - if isinstance(found, tuple): - candidate = "".join(found) - else: - candidate = found + candidate = "".join(found) if isinstance(found, tuple) else found matches.append(SecretMatch(secret_type=secret_type, value=candidate)) return matches diff --git a/tests/conftest.py b/tests/conftest.py index 4eba9d3..bb02177 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,4 +1,11 @@ -"""Pytest configuration and fixtures.""" +"""Pytest configuration and fixtures. + +Every test runs with a scrubbed environment, an empty working directory (so a +developer's own ``.env`` is never read) and state paths inside ``tmp_path``. +Nothing may write to ``/var/log``, ``/var/lib`` or the real home directory. +""" + +from __future__ import annotations from collections.abc import Generator from pathlib import Path @@ -6,77 +13,37 @@ from pathlib import Path import pytest from aegis_gitea_mcp.audit import reset_audit_logger -from aegis_gitea_mcp.auth import reset_validator -from aegis_gitea_mcp.authz import reset_authz_caches -from aegis_gitea_mcp.config import reset_settings -from aegis_gitea_mcp.oauth import reset_oauth_validator -from aegis_gitea_mcp.oauth_flow import reset_oauth_client_registry -from aegis_gitea_mcp.observability import reset_metrics_registry -from aegis_gitea_mcp.policy import reset_policy_engine +from aegis_gitea_mcp.config import Settings, reset_settings from aegis_gitea_mcp.rate_limit import reset_rate_limiter from aegis_gitea_mcp.request_context import clear_gitea_auth_context -from aegis_gitea_mcp.server import reset_repo_authz_cache + + +def _reset_singletons() -> None: + """Reset every module-level singleton.""" + reset_settings() + reset_audit_logger() + reset_rate_limiter() + clear_gitea_auth_context() @pytest.fixture(autouse=True) def reset_globals(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Generator[None, None, None]: - """Reset global singletons between tests and set up temp audit log.""" - # Reset singletons before each test to ensure clean state - reset_settings() - reset_audit_logger() - reset_validator() - reset_oauth_validator() - reset_oauth_client_registry() - reset_repo_authz_cache() - reset_authz_caches() - reset_policy_engine() - reset_rate_limiter() - reset_metrics_registry() - clear_gitea_auth_context() + """Isolate each test: clean env, tmp cwd, tmp state paths, fresh singletons.""" + for field_name in Settings.model_fields: + monkeypatch.delenv(field_name.upper(), raising=False) + monkeypatch.chdir(tmp_path) - # Use temporary directory for audit logs in tests - audit_log_path = tmp_path / "audit.log" - monkeypatch.setenv("AUDIT_LOG_PATH", str(audit_log_path)) + data_dir = tmp_path / "data" + monkeypatch.setenv("DATA_DIR", str(data_dir)) + monkeypatch.setenv("AUDIT_LOG_PATH", str(data_dir / "audit.log")) + _reset_singletons() yield - - # Also reset after test for cleanup - reset_settings() - reset_audit_logger() - reset_validator() - reset_oauth_validator() - reset_oauth_client_registry() - reset_repo_authz_cache() - reset_authz_caches() - reset_policy_engine() - reset_rate_limiter() - reset_metrics_registry() - clear_gitea_auth_context() + _reset_singletons() @pytest.fixture def mock_env(monkeypatch: pytest.MonkeyPatch) -> None: - """Set up mock environment variables for testing (standard API key mode).""" + """Minimal valid environment: a Gitea URL and a token.""" monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") monkeypatch.setenv("GITEA_TOKEN", "test-token-12345") - monkeypatch.setenv("ENVIRONMENT", "test") - monkeypatch.setenv("MCP_HOST", "127.0.0.1") - monkeypatch.setenv("MCP_PORT", "8080") - monkeypatch.setenv("LOG_LEVEL", "DEBUG") - monkeypatch.setenv("MCP_API_KEYS", "a" * 64) - monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "false") - - -@pytest.fixture -def mock_env_oauth(monkeypatch: pytest.MonkeyPatch) -> None: - """Set up mock environment variables for OAuth mode testing.""" - monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") - monkeypatch.setenv("ENVIRONMENT", "test") - monkeypatch.setenv("MCP_HOST", "127.0.0.1") - monkeypatch.setenv("MCP_PORT", "8080") - monkeypatch.setenv("LOG_LEVEL", "DEBUG") - monkeypatch.setenv("OAUTH_MODE", "true") - monkeypatch.setenv("GITEA_OAUTH_CLIENT_ID", "test-client-id") - monkeypatch.setenv("GITEA_OAUTH_CLIENT_SECRET", "test-client-secret") - monkeypatch.setenv("OAUTH_STATE_SECRET", "test-state-secret-0123456789abcdef") - monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "false") diff --git a/tests/test_audit.py b/tests/test_audit.py index 26c970c..479fed9 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -11,8 +11,6 @@ from aegis_gitea_mcp.audit import AuditLogger, validate_audit_log_integrity def test_audit_log_integrity_valid(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: """Fresh audit log should validate with intact hash chain.""" monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") - monkeypatch.setenv("GITEA_TOKEN", "token-123") - monkeypatch.setenv("MCP_API_KEYS", "a" * 64) log_path = tmp_path / "audit.log" logger = AuditLogger(log_path=log_path) @@ -30,8 +28,6 @@ def test_audit_log_integrity_detects_tamper( ) -> None: """Integrity validation should fail when entries are modified.""" monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") - monkeypatch.setenv("GITEA_TOKEN", "token-123") - monkeypatch.setenv("MCP_API_KEYS", "a" * 64) log_path = tmp_path / "audit.log" logger = AuditLogger(log_path=log_path) @@ -48,3 +44,37 @@ def test_audit_log_integrity_detects_tamper( valid, errors = validate_audit_log_integrity(log_path) assert not valid assert errors + + +def test_audit_logger_uses_configured_path(mock_env: None, tmp_path: Path) -> None: + """Without an explicit path the logger writes to AUDIT_LOG_PATH (tmp in tests).""" + from aegis_gitea_mcp.audit import get_audit_logger + + logger = get_audit_logger() + logger.log_access_denied("tool", reason="nope") + logger.log_security_event("evt", "description", metadata={"k": "v"}) + + assert logger.log_path == tmp_path / "data" / "audit.log" + assert len(logger.log_path.read_text(encoding="utf-8").splitlines()) == 2 + valid, errors = validate_audit_log_integrity(logger.log_path) + assert valid, errors + + +def test_audit_masks_secrets_and_continues_chain(mock_env: None, tmp_path: Path) -> None: + """Secrets are masked before persistence and a reopened log continues the chain.""" + log_path = tmp_path / "chain.log" + first = AuditLogger(log_path=log_path) + first.log_tool_invocation("t", params={"note": "key AKIA1234567890ABCDEF"}) + first.close() + second = AuditLogger(log_path=log_path) + second.log_tool_invocation("t", error="boom", result_status="error") + second.close() + + text = log_path.read_text(encoding="utf-8") + assert "AKIA1234567890ABCDEF" not in text + valid, errors = validate_audit_log_integrity(log_path) + assert valid, errors + + +def test_validate_missing_log_is_valid(tmp_path: Path) -> None: + assert validate_audit_log_integrity(tmp_path / "absent.log") == (True, []) diff --git a/tests/test_catalog_model.py b/tests/test_catalog_model.py new file mode 100644 index 0000000..12de2c4 --- /dev/null +++ b/tests/test_catalog_model.py @@ -0,0 +1,117 @@ +"""Tests for the catalog data model.""" + +from __future__ import annotations + +import json +from typing import Any + +import pytest + +from aegis_gitea_mcp.catalog import Operation, Parameter + + +def sample_dict() -> dict[str, Any]: + return { + "tool_name": "gitea_issue_create", + "domain": "issue", + "operation_id": "issueCreateIssue", + "method": "POST", + "path": "/repos/{owner}/{repo}/issues", + "summary": "Create an issue", + "description": "", + "parameters": [ + { + "name": "owner", + "arg_name": "owner", + "location": "path", + "required": True, + "schema": {"type": "string"}, + "is_file": False, + }, + { + "name": "body", + "arg_name": "body", + "location": "body", + "required": False, + "schema": {"type": "object"}, + }, + ], + "input_schema": {"type": "object", "additionalProperties": False}, + "consumes": ["application/json"], + "produces": ["application/json"], + "read_only": False, + "destructive": False, + "idempotent": False, + "paginated": False, + } + + +def test_operation_round_trip_through_json() -> None: + data = sample_dict() + operation = Operation.from_dict(data) + + assert operation.method == "POST" + assert isinstance(operation.parameters, tuple) + assert isinstance(operation.consumes, tuple) + assert operation.parameters[0].location == "path" + assert operation.parameters[1].is_file is False + + rebuilt = Operation.from_dict(json.loads(json.dumps(operation.to_dict()))) + assert rebuilt == operation + + +def test_to_dict_uses_lists_and_is_json_serializable() -> None: + dumped = Operation.from_dict(sample_dict()).to_dict() + assert isinstance(dumped["parameters"], list) + assert isinstance(dumped["consumes"], list) + json.dumps(dumped) + + +def test_method_is_upper_cased() -> None: + data = sample_dict() + data["method"] = "post" + assert Operation.from_dict(data).method == "POST" + + +def test_optional_fields_default() -> None: + data = sample_dict() + del data["summary"], data["description"], data["consumes"], data["produces"] + operation = Operation.from_dict(data) + assert operation.summary == "" + assert operation.consumes == () + assert operation.produces == () + + +def test_invalid_location_rejected() -> None: + with pytest.raises(ValueError, match="location"): + Parameter.from_dict( + {"name": "x", "arg_name": "x", "location": "header", "required": True, "schema": {}} + ) + + +def test_missing_required_key_raises() -> None: + data = sample_dict() + del data["tool_name"] + with pytest.raises(KeyError): + Operation.from_dict(data) + + +def test_operation_is_frozen() -> None: + operation = Operation.from_dict(sample_dict()) + with pytest.raises(AttributeError): + operation.method = "GET" # type: ignore[misc] + + +def test_file_parameter() -> None: + parameter = Parameter.from_dict( + { + "name": "attachment", + "arg_name": "attachment", + "location": "formData", + "required": True, + "schema": {"type": "string", "contentEncoding": "base64"}, + "is_file": True, + } + ) + assert parameter.is_file is True + assert parameter.to_dict()["location"] == "formData" diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..894a025 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,37 @@ +"""Tests for the command line skeleton.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from aegis_gitea_mcp import __version__ +from aegis_gitea_mcp.cli import build_parser, main + + +def test_version_prints_package_version(capsys: pytest.CaptureFixture[str]) -> None: + assert main(["version"]) == 0 + assert capsys.readouterr().out.strip() == __version__ + + +@pytest.mark.parametrize("argv", [[], ["serve"], ["serve", "--transport", "http"], ["setup"]]) +def test_unbuilt_commands_exit_2(argv: list[str], capsys: pytest.CaptureFixture[str]) -> None: + assert main(argv) == 2 + captured = capsys.readouterr() + assert "not implemented yet" in captured.err + assert captured.out == "" + + +def test_serve_transport_choices() -> None: + parser = build_parser() + assert parser.parse_args(["serve", "--transport", "stdio"]).transport == "stdio" + with pytest.raises(SystemExit): + parser.parse_args(["serve", "--transport", "sse"]) + + +def test_package_version_matches_pyproject() -> None: + tomllib = pytest.importorskip("tomllib") # Python 3.11+ + + data = tomllib.loads((Path(__file__).parent.parent / "pyproject.toml").read_text()) + assert data["project"]["version"] == __version__ diff --git a/tests/test_config.py b/tests/test_config.py index 8680193..5b5b7ae 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,138 +1,391 @@ """Tests for configuration management.""" +from __future__ import annotations + import pytest from pydantic import ValidationError -from aegis_gitea_mcp.config import get_settings, reset_settings +from aegis_gitea_mcp.config import ( + DEFAULT_OAUTH_SCOPES, + GITEA_SCOPES, + Settings, + get_settings, + is_loopback_host, + reset_settings, +) -def test_settings_from_env(mock_env: None) -> None: - """Test loading settings from environment variables.""" +def make(monkeypatch: pytest.MonkeyPatch, **env: str) -> Settings: + """Build fresh settings from the given environment.""" + for key, value in env.items(): + monkeypatch.setenv(key, value) + reset_settings() + return get_settings() + + +def test_defaults(mock_env: None, tmp_path) -> None: settings = get_settings() - assert settings.gitea_base_url == "https://gitea.example.com" - assert settings.gitea_token == "test-token-12345" + assert settings.gitea_url == "https://gitea.example.com" + assert settings.gitea_api_url == "https://gitea.example.com/api/v1" + assert settings.gitea_token is not None + assert settings.gitea_token.get_secret_value() == "test-token-12345" + assert settings.mcp_transport == "stdio" assert settings.mcp_host == "127.0.0.1" assert settings.mcp_port == 8080 - assert settings.log_level == "DEBUG" - - -def test_settings_defaults(monkeypatch: pytest.MonkeyPatch) -> None: - """Test default values when not specified.""" - monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") - monkeypatch.setenv("GITEA_TOKEN", "test-token") - monkeypatch.setenv("MCP_API_KEYS", "a" * 64) - - settings = get_settings() - - assert settings.mcp_host == "127.0.0.1" - assert settings.mcp_port == 8080 - assert settings.log_level == "INFO" - assert settings.max_file_size_bytes == 1_048_576 + assert settings.mcp_path == "/mcp" + assert settings.mcp_bearer_token is None + assert settings.public_base_url is None + assert settings.allowed_hosts == [] + assert settings.oauth_enabled is False + assert settings.read_only is False + assert settings.toolsets == ["all"] + assert settings.disabled_tools == [] assert settings.request_timeout_seconds == 30 + assert settings.max_tool_response_items == 200 + assert settings.max_tool_response_chars == 60000 + assert settings.max_binary_bytes == 5_242_880 + assert settings.secret_detection_mode == "mask" + assert settings.log_level == "INFO" + assert settings.expose_error_details is False + assert settings.rate_limit_per_minute == 120 + assert ",".join(settings.oauth_scopes) == DEFAULT_OAUTH_SCOPES -def test_settings_validation_missing_required(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: - """Test that missing required fields raise validation errors.""" +def test_data_dir_defaults_and_audit_path(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("DATA_DIR") + monkeypatch.delenv("AUDIT_LOG_PATH") + settings = make(monkeypatch, GITEA_URL="https://g.example.com") - monkeypatch.delenv("GITEA_URL", raising=False) - monkeypatch.delenv("GITEA_TOKEN", raising=False) - - # Change to tmp directory so .env file won't be found - monkeypatch.chdir(tmp_path) - - reset_settings() - - with pytest.raises(ValidationError): - get_settings() + assert str(settings.data_dir).endswith(".local/state/aegis-gitea-mcp") + assert "~" not in str(settings.data_dir) + assert settings.resolved_audit_log_path == settings.data_dir / "audit.log" -def test_settings_invalid_log_level(monkeypatch: pytest.MonkeyPatch) -> None: - """Test that invalid log levels are rejected.""" - monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") - monkeypatch.setenv("GITEA_TOKEN", "test-token") - monkeypatch.setenv("MCP_API_KEYS", "a" * 64) - monkeypatch.setenv("LOG_LEVEL", "INVALID") +def test_audit_path_follows_data_dir(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: + monkeypatch.delenv("AUDIT_LOG_PATH") + settings = make(monkeypatch, GITEA_URL="https://g.example.com", DATA_DIR=str(tmp_path / "d")) - reset_settings() - - with pytest.raises(ValidationError): - get_settings() + assert settings.resolved_audit_log_path == tmp_path / "d" / "audit.log" -def test_settings_empty_token(monkeypatch: pytest.MonkeyPatch) -> None: - """Test that empty tokens are rejected.""" - monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") - monkeypatch.setenv("GITEA_TOKEN", " ") - monkeypatch.setenv("MCP_API_KEYS", "a" * 64) +def test_blank_values_are_unset(monkeypatch: pytest.MonkeyPatch) -> None: + settings = make( + monkeypatch, + GITEA_URL="https://g.example.com", + GITEA_TOKEN="", + MCP_BEARER_TOKEN=" ", + PUBLIC_BASE_URL="", + AUDIT_LOG_PATH="", + ) - reset_settings() - - with pytest.raises(ValidationError): - get_settings() + assert settings.gitea_token is None + assert settings.mcp_bearer_token is None + assert settings.public_base_url is None + assert settings.audit_log_path is None -def test_settings_singleton(mock_env: None) -> None: - """Test that get_settings returns same instance.""" - settings1 = get_settings() - settings2 = get_settings() - - assert settings1 is settings2 +def test_env_names_are_case_insensitive(monkeypatch: pytest.MonkeyPatch) -> None: + settings = make(monkeypatch, gitea_url="https://g.example.com", mcp_port="9000") + assert settings.mcp_port == 9000 -def test_write_mode_requires_whitelist_or_allow_all(monkeypatch: pytest.MonkeyPatch) -> None: - """Write mode without whitelist must be rejected unless allow-all is enabled.""" - monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") - monkeypatch.setenv("GITEA_TOKEN", "test-token") - monkeypatch.setenv("MCP_API_KEYS", "a" * 64) - monkeypatch.setenv("WRITE_MODE", "true") - monkeypatch.delenv("WRITE_REPOSITORY_WHITELIST", raising=False) - monkeypatch.setenv("WRITE_ALLOW_ALL_TOKEN_REPOS", "false") - - reset_settings() - with pytest.raises(ValidationError): - get_settings() - - -def test_write_mode_allows_all_token_repos(monkeypatch: pytest.MonkeyPatch) -> None: - """Allow-all mode should pass validation without explicit repository whitelist.""" - monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") - monkeypatch.setenv("GITEA_TOKEN", "test-token") - monkeypatch.setenv("MCP_API_KEYS", "a" * 64) - monkeypatch.setenv("WRITE_MODE", "true") - monkeypatch.delenv("WRITE_REPOSITORY_WHITELIST", raising=False) - monkeypatch.setenv("WRITE_ALLOW_ALL_TOKEN_REPOS", "true") - +def test_reads_dotenv_file(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: + (tmp_path / ".env").write_text("GITEA_URL=https://dotenv.example.com\nREAD_ONLY=true\n") reset_settings() settings = get_settings() - assert settings.write_allow_all_token_repos is True + + assert settings.gitea_url == "https://dotenv.example.com" + assert settings.read_only is True -def _oauth_env(monkeypatch: pytest.MonkeyPatch) -> None: - """Apply a minimal, valid OAuth-mode environment.""" - monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") - monkeypatch.setenv("OAUTH_MODE", "true") - monkeypatch.setenv("GITEA_OAUTH_CLIENT_ID", "test-client-id") - monkeypatch.setenv("GITEA_OAUTH_CLIENT_SECRET", "test-client-secret") - monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "false") - - -def test_oauth_state_secret_too_short_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None: - """OAUTH_STATE_SECRET shorter than 32 characters must fail validation.""" - _oauth_env(monkeypatch) - monkeypatch.setenv("OAUTH_STATE_SECRET", "short-secret") - - reset_settings() - with pytest.raises(ValidationError, match="at least 32 characters"): +def test_missing_gitea_url_is_rejected() -> None: + with pytest.raises(ValidationError): get_settings() -def test_oauth_state_secret_minimum_length_accepted(monkeypatch: pytest.MonkeyPatch) -> None: - """A 32+ character OAUTH_STATE_SECRET passes validation.""" - _oauth_env(monkeypatch) - monkeypatch.setenv("OAUTH_STATE_SECRET", "x" * 32) - +def test_settings_singleton_and_reset(mock_env: None) -> None: + first = get_settings() + assert get_settings() is first reset_settings() - settings = get_settings() - assert settings.oauth_mode is True - assert len(settings.oauth_state_secret) == 32 + assert get_settings() is not first + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("https://gitea.example.com", "https://gitea.example.com"), + ("https://gitea.example.com/", "https://gitea.example.com"), + ("http://localhost:3000", "http://localhost:3000"), + ("https://gitea.example.com/api/v1", "https://gitea.example.com"), + ("https://gitea.example.com/api/v1/", "https://gitea.example.com"), + ("https://example.com/gitea", "https://example.com/gitea"), + ("https://example.com/gitea/api/v1", "https://example.com/gitea"), + ], +) +def test_gitea_url_normalised(monkeypatch: pytest.MonkeyPatch, raw: str, expected: str) -> None: + assert make(monkeypatch, GITEA_URL=raw).gitea_url == expected + + +@pytest.mark.parametrize( + "raw", + [ + "ftp://gitea.example.com", + "gitea.example.com", + "https://", + "https://user:pw@gitea.example.com", + "https://user@gitea.example.com", + "https://gitea.example.com?x=1", + "https://gitea.example.com/#frag", + "https://gitea.example.com:99999", + "file:///etc/passwd", + ], +) +def test_gitea_url_rejected(monkeypatch: pytest.MonkeyPatch, raw: str) -> None: + monkeypatch.setenv("GITEA_URL", raw) + with pytest.raises(ValidationError): + get_settings() + + +def test_bearer_token_minimum_length(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITEA_URL", "https://g.example.com") + monkeypatch.setenv("MCP_BEARER_TOKEN", "short") + with pytest.raises(ValidationError, match="at least 32"): + get_settings() + + settings = make(monkeypatch, MCP_BEARER_TOKEN="x" * 32) + assert settings.mcp_bearer_token is not None + + +def test_encryption_key_minimum_length(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITEA_URL", "https://g.example.com") + monkeypatch.setenv("TOKEN_ENCRYPTION_KEY", "short") + with pytest.raises(ValidationError, match="TOKEN_ENCRYPTION_KEY"): + get_settings() + + +@pytest.mark.parametrize( + "raw", + [ + "https://mcp.example.com", + "https://mcp.example.com/", + "http://localhost:8080", + "http://127.0.0.1:8080", + "http://[::1]:8080", + ], +) +def test_public_base_url_accepted(monkeypatch: pytest.MonkeyPatch, raw: str) -> None: + settings = make(monkeypatch, GITEA_URL="https://g.example.com", PUBLIC_BASE_URL=raw) + assert settings.public_base_url == raw.rstrip("/") + + +@pytest.mark.parametrize( + "raw", + [ + "http://mcp.example.com", + "mcp.example.com", + "https://mcp.example.com?a=b", + "https://u:p@mcp.example.com", + ], +) +def test_public_base_url_rejected(monkeypatch: pytest.MonkeyPatch, raw: str) -> None: + monkeypatch.setenv("GITEA_URL", "https://g.example.com") + monkeypatch.setenv("PUBLIC_BASE_URL", raw) + with pytest.raises(ValidationError): + get_settings() + + +def test_mcp_path_validation(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITEA_URL", "https://g.example.com") + assert make(monkeypatch, MCP_PATH="/api/mcp/").mcp_path == "/api/mcp" + for bad in ("mcp", "//evil", "/a?b", "/", "/a b"): + monkeypatch.setenv("MCP_PATH", bad) + reset_settings() + with pytest.raises(ValidationError): + get_settings() + + +def test_port_and_transport_validation(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITEA_URL", "https://g.example.com") + monkeypatch.setenv("MCP_PORT", "70000") + with pytest.raises(ValidationError): + get_settings() + monkeypatch.setenv("MCP_PORT", "8080") + monkeypatch.setenv("MCP_TRANSPORT", "sse") + with pytest.raises(ValidationError): + get_settings() + assert make(monkeypatch, MCP_TRANSPORT="http").mcp_transport == "http" + + +def test_comma_separated_lists(monkeypatch: pytest.MonkeyPatch) -> None: + settings = make( + monkeypatch, + GITEA_URL="https://g.example.com", + ALLOWED_HOSTS="mcp.example.com, other.example.com:8443,mcp.example.com", + TOOLSETS="Repo, issue", + DISABLED_TOOLS="Gitea_Repo_Delete, gitea_admin_delete_user", + ) + + assert settings.allowed_hosts == ["mcp.example.com", "other.example.com:8443"] + assert settings.toolsets == ["repo", "issue"] + assert settings.disabled_tools == ["gitea_repo_delete", "gitea_admin_delete_user"] + + +def test_unknown_toolset_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITEA_URL", "https://g.example.com") + monkeypatch.setenv("TOOLSETS", "repo,bogus") + with pytest.raises(ValidationError, match="bogus"): + get_settings() + + +def test_empty_toolsets_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITEA_URL", "https://g.example.com") + monkeypatch.setenv("TOOLSETS", " , ") + with pytest.raises(ValidationError): + get_settings() + + +def test_default_oauth_scopes_are_valid() -> None: + assert set(DEFAULT_OAUTH_SCOPES.split(",")) <= GITEA_SCOPES + assert len(GITEA_SCOPES) == 18 + + +def test_oauth_scopes_custom_and_dedup(monkeypatch: pytest.MonkeyPatch) -> None: + settings = make( + monkeypatch, + GITEA_URL="https://g.example.com", + OAUTH_SCOPES="read:repository write:repository,read:repository", + ) + assert settings.oauth_scopes == ["read:repository", "write:repository"] + + +@pytest.mark.parametrize( + "scope", ["all", "repo", "read:bogus", "write:repository,admin", "public-only"] +) +def test_unknown_oauth_scope_rejected(monkeypatch: pytest.MonkeyPatch, scope: str) -> None: + monkeypatch.setenv("GITEA_URL", "https://g.example.com") + monkeypatch.setenv("OAUTH_SCOPES", scope) + with pytest.raises(ValidationError, match="unknown scope"): + get_settings() + + +def test_empty_oauth_scopes_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITEA_URL", "https://g.example.com") + monkeypatch.setenv("OAUTH_SCOPES", ",") + with pytest.raises(ValidationError, match="must not be empty"): + get_settings() + + +def test_log_level_normalised_and_validated(monkeypatch: pytest.MonkeyPatch) -> None: + assert ( + make(monkeypatch, GITEA_URL="https://g.example.com", LOG_LEVEL="debug").log_level == "DEBUG" + ) + monkeypatch.setenv("LOG_LEVEL", "loud") + reset_settings() + with pytest.raises(ValidationError): + get_settings() + + +def test_secret_detection_mode_validated(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITEA_URL", "https://g.example.com") + monkeypatch.setenv("SECRET_DETECTION_MODE", "block") + assert get_settings().secret_detection_mode == "block" + reset_settings() + monkeypatch.setenv("SECRET_DETECTION_MODE", "weird") + with pytest.raises(ValidationError): + get_settings() + + +@pytest.mark.parametrize( + ("host", "expected"), + [ + ("127.0.0.1", True), + ("127.5.6.7", True), + ("::1", True), + ("[::1]", True), + ("localhost", True), + ("LOCALHOST", True), + ("0.0.0.0", False), # noqa: S104 + ("::", False), + ("192.168.1.5", False), + ("gitea.example.com", False), + ], +) +def test_is_loopback(monkeypatch: pytest.MonkeyPatch, host: str, expected: bool) -> None: + assert is_loopback_host(host) is expected + settings = make(monkeypatch, GITEA_URL="https://g.example.com", MCP_HOST=host) + assert settings.is_loopback_bind() is expected + + +def test_oauth_config_errors(monkeypatch: pytest.MonkeyPatch) -> None: + settings = make(monkeypatch, GITEA_URL="https://g.example.com") + assert settings.oauth_config_errors() == [] + + settings = make(monkeypatch, OAUTH_ENABLED="true") + errors = settings.oauth_config_errors() + assert len(errors) == 4 + assert any("GITEA_OAUTH_CLIENT_ID" in e for e in errors) + assert any("GITEA_OAUTH_CLIENT_SECRET" in e for e in errors) + assert any("TOKEN_ENCRYPTION_KEY" in e for e in errors) + assert any("PUBLIC_BASE_URL" in e for e in errors) + + settings = make( + monkeypatch, + GITEA_OAUTH_CLIENT_ID="cid", + GITEA_OAUTH_CLIENT_SECRET="csecret", + TOKEN_ENCRYPTION_KEY="k" * 32, + PUBLIC_BASE_URL="https://mcp.example.com", + ) + assert settings.oauth_config_errors() == [] + + +def test_http_auth_configured(monkeypatch: pytest.MonkeyPatch) -> None: + assert make(monkeypatch, GITEA_URL="https://g.example.com").http_auth_configured() is False + assert make(monkeypatch, MCP_BEARER_TOKEN="b" * 32).http_auth_configured() is True + + monkeypatch.delenv("MCP_BEARER_TOKEN") + assert make(monkeypatch, OAUTH_ENABLED="true").http_auth_configured() is False + settings = make( + monkeypatch, + GITEA_OAUTH_CLIENT_ID="cid", + GITEA_OAUTH_CLIENT_SECRET="csecret", + TOKEN_ENCRYPTION_KEY="k" * 32, + PUBLIC_BASE_URL="https://mcp.example.com", + ) + assert settings.http_auth_configured() is True + + +def test_secrets_never_appear_in_repr_or_dump(monkeypatch: pytest.MonkeyPatch) -> None: + secrets = { + "GITEA_TOKEN": "gitea-token-SECRET-aaaaaaaaaaaa", + "MCP_BEARER_TOKEN": "bearer-SECRET-bbbbbbbbbbbbbbbbbbbbbbbbbb", + "GITEA_OAUTH_CLIENT_SECRET": "client-SECRET-cccccccccccc", + "TOKEN_ENCRYPTION_KEY": "encryption-SECRET-dddddddddddddddddddd", + } + settings = make(monkeypatch, GITEA_URL="https://g.example.com", **secrets) + + rendered = [ + repr(settings), + str(settings), + str(settings.model_dump()), + settings.model_dump_json(), + str(settings.model_dump(mode="json")), + ] + for text in rendered: + for secret in secrets.values(): + assert secret not in text + assert settings.gitea_token is not None + assert settings.gitea_token.get_secret_value() == secrets["GITEA_TOKEN"] + + +def test_env_example_is_valid() -> None: + """The documented example file must load as-is (blank values mean unset).""" + from pathlib import Path + + example = Path(__file__).parent.parent / ".env.example" + settings = Settings(_env_file=example) # type: ignore[call-arg] + + assert settings.gitea_url == "https://gitea.example.com" + assert settings.gitea_token is None + assert settings.toolsets == ["all"] + assert ",".join(settings.oauth_scopes) == DEFAULT_OAUTH_SCOPES diff --git a/tests/test_errors.py b/tests/test_errors.py new file mode 100644 index 0000000..1d9f6b6 --- /dev/null +++ b/tests/test_errors.py @@ -0,0 +1,73 @@ +"""Tests for error types and message sanitization.""" + +from __future__ import annotations + +import pytest + +from aegis_gitea_mcp.errors import GiteaAPIError, ToolError, kind_for_status, sanitize_message + + +def test_tool_error_defaults() -> None: + error = ToolError("nope") + assert str(error) == "nope" + assert error.detail == "nope" + assert error.status_code == 400 + assert ToolError("x", status_code=403).status_code == 403 + + +@pytest.mark.parametrize( + ("status", "kind"), + [ + (400, "bad_request"), + (401, "unauthorized"), + (403, "forbidden"), + (404, "not_found"), + (409, "conflict"), + (422, "validation_failed"), + (429, "rate_limited"), + (418, "client_error"), + (500, "server_error"), + (503, "server_error"), + (0, "error"), + ], +) +def test_kind_for_status(status: int, kind: str) -> None: + assert kind_for_status(status) == kind + + +def test_gitea_api_error_fields() -> None: + error = GiteaAPIError(404, "repository does not exist") + + assert isinstance(error, ToolError) + assert error.status == 404 + assert error.status_code == 404 + assert error.kind == "not_found" + assert error.message == "repository does not exist" + assert error.to_dict() == { + "status": 404, + "kind": "not_found", + "message": "repository does not exist", + } + + +def test_gitea_api_error_explicit_kind() -> None: + assert GiteaAPIError(0, "connection refused", kind="network_error").kind == "network_error" + + +def test_gitea_api_error_sanitizes_message() -> None: + error = GiteaAPIError(401, "bad header Authorization: Bearer abcdef1234567890abcdef\nretry") + + assert "abcdef1234567890abcdef" not in error.message + assert "\n" not in error.message + assert error.detail == error.message + + +def test_sanitize_message_masks_secret_patterns() -> None: + message = sanitize_message("failed with key AKIA1234567890ABCDEF in body") + assert "AKIA1234567890ABCDEF" not in message + + +def test_sanitize_message_truncates() -> None: + message = sanitize_message("x" * 2000, max_chars=100) + assert len(message) == 100 + assert message.endswith("...") diff --git a/tests/test_identity.py b/tests/test_identity.py new file mode 100644 index 0000000..9e97ef7 --- /dev/null +++ b/tests/test_identity.py @@ -0,0 +1,31 @@ +"""Tests for the identity type.""" + +from __future__ import annotations + +import dataclasses + +import pytest + +from aegis_gitea_mcp.identity import Identity + + +def test_metadata_never_contains_token() -> None: + identity = Identity(login="alice", mode="oauth", token="super-secret-token") + + assert identity.metadata() == {"login": "alice", "mode": "oauth", "sudo": None} + assert identity.metadata(sudo="bob") == {"login": "alice", "mode": "oauth", "sudo": "bob"} + assert "super-secret-token" not in str(identity.metadata(sudo="bob")) + + +def test_token_hidden_from_repr() -> None: + identity = Identity(login=None, mode="admin-token", token="super-secret-token") + + assert "super-secret-token" not in repr(identity) + assert "super-secret-token" not in str(identity) + assert "login=None" in repr(identity) + + +def test_identity_is_frozen() -> None: + identity = Identity(login="a", mode="admin-token", token="t") + with pytest.raises(dataclasses.FrozenInstanceError): + identity.login = "b" # type: ignore[misc] diff --git a/tests/test_logging_utils.py b/tests/test_logging_utils.py index f5a62fa..6b71d9f 100644 --- a/tests/test_logging_utils.py +++ b/tests/test_logging_utils.py @@ -1,4 +1,4 @@ -"""Tests for structured logging helpers and get_issue instrumentation (#14).""" +"""Tests for structured logging helpers.""" from __future__ import annotations @@ -7,26 +7,12 @@ import logging import pytest -from aegis_gitea_mcp.config import reset_settings from aegis_gitea_mcp.logging_utils import ( JsonLogFormatter, log_event, log_nullable_field, sanitize_context, ) -from aegis_gitea_mcp.tools.read_tools import get_issue_tool - -READ_TOOLS_LOGGER = "aegis_gitea_mcp.tools.read_tools" - - -@pytest.fixture(autouse=True) -def tool_env(monkeypatch: pytest.MonkeyPatch) -> None: - """Provide minimal settings environment for response limit helpers.""" - reset_settings() - monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") - monkeypatch.setenv("GITEA_TOKEN", "test-token") - monkeypatch.setenv("MCP_API_KEYS", "a" * 64) - monkeypatch.setenv("ENVIRONMENT", "test") def test_sanitize_context_masks_sensitive_keys() -> None: @@ -82,39 +68,3 @@ def test_log_nullable_field_characterizes_value(caplog: pytest.LogCaptureFixture "is_none": False, "value_type": "list", } - - -class _NullIssueGitea: - """Minimal stub returning an issue with null collection fields.""" - - async def get_issue(self, owner: str, repo: str, index: int) -> dict[str, object]: - return { - "number": index, - "title": "Issue", - "body": "Body", - "state": "open", - "labels": None, - "assignees": None, - "user": None, - } - - -@pytest.mark.asyncio -async def test_get_issue_tool_emits_debug_events(caplog: pytest.LogCaptureFixture) -> None: - """get_issue_tool emits start/shape/field-check debug events and still parses.""" - with caplog.at_level(logging.DEBUG, logger=READ_TOOLS_LOGGER): - result = await get_issue_tool( - _NullIssueGitea(), {"owner": "acme", "repo": "app", "issue_number": 7} - ) - - events = [r.getMessage() for r in caplog.records] - assert "get_issue.start" in events - assert "get_issue.payload_shape" in events - - field_checks = [r.context for r in caplog.records if r.getMessage() == "get_issue.field_check"] - assert {"field": "labels", "is_none": True, "value_type": None} in field_checks - assert {"field": "user", "is_none": True, "value_type": None} in field_checks - - # Logging must not change behaviour: null collections still parse to empties. - assert result["labels"] == [] - assert result["assignees"] == [] diff --git a/tests/test_no_host_paths.py b/tests/test_no_host_paths.py new file mode 100644 index 0000000..46f0674 --- /dev/null +++ b/tests/test_no_host_paths.py @@ -0,0 +1,19 @@ +"""Guard: the test environment must never point state at system directories.""" + +from __future__ import annotations + +from pathlib import Path + +from aegis_gitea_mcp.audit import get_audit_logger +from aegis_gitea_mcp.config import Settings + + +def test_state_paths_are_inside_tmp(mock_env: None, tmp_path: Path) -> None: + settings = Settings() # type: ignore[call-arg] + assert settings.data_dir.is_relative_to(tmp_path) + assert settings.resolved_audit_log_path.is_relative_to(tmp_path) + assert get_audit_logger().log_path.is_relative_to(tmp_path) + + +def test_cwd_is_isolated(tmp_path: Path) -> None: + assert Path.cwd() == tmp_path diff --git a/tests/test_rate_limit.py b/tests/test_rate_limit.py new file mode 100644 index 0000000..42e317a --- /dev/null +++ b/tests/test_rate_limit.py @@ -0,0 +1,55 @@ +"""Tests for the request rate limiter.""" + +from __future__ import annotations + +import pytest + +from aegis_gitea_mcp.audit import validate_audit_log_integrity +from aegis_gitea_mcp.config import get_settings +from aegis_gitea_mcp.rate_limit import ( + RequestRateLimiter, + SlidingWindowLimiter, + get_rate_limiter, + reset_rate_limiter, +) + + +def test_sliding_window_blocks_after_limit(monkeypatch: pytest.MonkeyPatch) -> None: + now = [1000.0] + monkeypatch.setattr("aegis_gitea_mcp.rate_limit.time.time", lambda: now[0]) + limiter = SlidingWindowLimiter(max_requests=2, window_seconds=60) + + assert limiter.allow("k") + assert limiter.allow("k") + assert not limiter.allow("k") + assert limiter.allow("other") + + now[0] += 61 + assert limiter.allow("k") + + +def test_request_limiter_per_ip_and_token(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITEA_URL", "https://g.example.com") + monkeypatch.setenv("RATE_LIMIT_PER_MINUTE", "2") + limiter = RequestRateLimiter() + + assert limiter.check("1.1.1.1", "tok").allowed + assert limiter.check("1.1.1.1", None).allowed + assert not limiter.check("1.1.1.1", "tok").allowed + + # A different IP with the same token trips the per-token limiter. + other = limiter.check("2.2.2.2", "tok") + assert other.allowed or other.reason == "Per-token rate limit exceeded" + third = limiter.check("3.3.3.3", "tok") + assert not third.allowed + assert third.reason == "Per-token rate limit exceeded" + + valid, errors = validate_audit_log_integrity(get_settings().resolved_audit_log_path) + assert valid, errors + + +def test_global_limiter_singleton(mock_env: None) -> None: + first = get_rate_limiter() + assert get_rate_limiter() is first + reset_rate_limiter() + assert get_rate_limiter() is not first diff --git a/tests/test_request_context.py b/tests/test_request_context.py new file mode 100644 index 0000000..321001f --- /dev/null +++ b/tests/test_request_context.py @@ -0,0 +1,35 @@ +"""Tests for request-scoped context variables.""" + +from __future__ import annotations + +from aegis_gitea_mcp.request_context import ( + clear_gitea_auth_context, + get_gitea_user_login, + get_gitea_user_scopes, + get_gitea_user_token, + get_request_id, + set_gitea_user_login, + set_gitea_user_scopes, + set_gitea_user_token, + set_request_id, +) + + +def test_request_id_roundtrip() -> None: + set_request_id("req-1") + assert get_request_id() == "req-1" + + +def test_auth_context_set_and_clear() -> None: + set_gitea_user_token("tok") + set_gitea_user_login("alice") + set_gitea_user_scopes([" write:repository", "read:user", "read:user", ""]) + + assert get_gitea_user_token() == "tok" + assert get_gitea_user_login() == "alice" + assert get_gitea_user_scopes() == ("read:user", "write:repository") + + clear_gitea_auth_context() + assert get_gitea_user_token() is None + assert get_gitea_user_login() is None + assert get_gitea_user_scopes() == () diff --git a/tests/test_response_limits.py b/tests/test_response_limits.py new file mode 100644 index 0000000..85d0b70 --- /dev/null +++ b/tests/test_response_limits.py @@ -0,0 +1,46 @@ +"""Tests for response size bounds.""" + +from __future__ import annotations + +import pytest + +from aegis_gitea_mcp.config import reset_settings +from aegis_gitea_mcp.response_limits import ResponseLimitError, limit_items, limit_text + + +def test_limit_items_uses_configured_default( + mock_env: None, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("MAX_TOOL_RESPONSE_ITEMS", "3") + reset_settings() + items = [{"i": n} for n in range(5)] + + trimmed, omitted = limit_items(items) + + assert trimmed == items[:3] + assert omitted == 2 + + +def test_limit_items_within_bound(mock_env: None) -> None: + items = [{"i": 1}] + assert limit_items(items) == (items, 0) + assert limit_items(items, configured_limit=1) == (items, 0) + assert limit_items([{"i": 1}, {"i": 2}], configured_limit=1) == ([{"i": 1}], 1) + + +def test_limit_text(mock_env: None) -> None: + assert limit_text("abcdef", configured_limit=3) == "abc" + assert limit_text("abc", configured_limit=10) == "abc" + + +def test_limit_text_default_from_settings(mock_env: None, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("MAX_TOOL_RESPONSE_CHARS", "5") + reset_settings() + assert limit_text("abcdefghij") == "abcde" + + +def test_invalid_limits_raise(mock_env: None) -> None: + with pytest.raises(ResponseLimitError): + limit_items([], configured_limit=-1) + with pytest.raises(ResponseLimitError): + limit_text("x", configured_limit=-1) diff --git a/tests/test_validate_audit_script.py b/tests/test_validate_audit_script.py new file mode 100644 index 0000000..1fbadfc --- /dev/null +++ b/tests/test_validate_audit_script.py @@ -0,0 +1,56 @@ +"""Tests for scripts/validate_audit_log.py.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path +from types import ModuleType + +import pytest + +from aegis_gitea_mcp.audit import AuditLogger + +SCRIPT = Path(__file__).parent.parent / "scripts" / "validate_audit_log.py" + + +def load_script() -> ModuleType: + spec = importlib.util.spec_from_file_location("validate_audit_log", SCRIPT) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_default_path_from_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + script = load_script() + monkeypatch.setenv("AUDIT_LOG_PATH", str(tmp_path / "x.log")) + assert script.default_audit_path() == tmp_path / "x.log" + + monkeypatch.delenv("AUDIT_LOG_PATH") + monkeypatch.setenv("DATA_DIR", str(tmp_path / "d")) + assert script.default_audit_path() == tmp_path / "d" / "audit.log" + + monkeypatch.delenv("DATA_DIR") + assert script.default_audit_path().name == "audit.log" + assert "aegis-gitea-mcp" in str(script.default_audit_path()) + + +def test_main_reports_ok_and_failure( + mock_env: None, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + script = load_script() + log_path = tmp_path / "audit.log" + logger = AuditLogger(log_path=log_path) + logger.log_tool_invocation("t") + logger.close() + + monkeypatch.setattr("sys.argv", ["x", "--path", str(log_path)]) + assert script.main() == 0 + assert "integrity OK" in capsys.readouterr().out + + log_path.write_text(log_path.read_text().replace('"t"', '"u"'), encoding="utf-8") + assert script.main() == 1 + assert "FAILED" in capsys.readouterr().out