From 8902c4f642c8a8f4783402336818201c38731e25 Mon Sep 17 00:00:00 2001 From: Latte Date: Sat, 27 Jun 2026 10:56:16 +0200 Subject: [PATCH] feat: add local stdio adapter and uv-installable package with extras Add aegis_gitea_mcp.stdio_app: a single-user, local MCP server over stdio (official mcp SDK) that serves the same tools from the shared registry, resolves the PAT owner via GET /user and pins request context to it, and runs policy + WRITE_MODE + secret sanitization + audit while skipping the per-user repo probe (the operator is the trusted token owner). Audit log falls back to a per-user state path when the container default is unwritable. Packaging: split deps into core (httpx/pydantic/mcp/...) and a [server] extra (fastapi/uvicorn/PyJWT/python-multipart); add console scripts aegis-gitea-mcp (stdio) and aegis-gitea-mcp-server (guarded HTTP entry); bump to 0.2.0 and fix repo URLs. mcp added to requirements for CI. Co-Authored-By: Claude Opus 4.8 (1M context) --- pyproject.toml | 34 ++-- requirements.txt | 1 + src/aegis_gitea_mcp/server_entry.py | 34 ++++ src/aegis_gitea_mcp/stdio_app.py | 231 ++++++++++++++++++++++++++++ 4 files changed, 290 insertions(+), 10 deletions(-) create mode 100644 src/aegis_gitea_mcp/server_entry.py create mode 100644 src/aegis_gitea_mcp/stdio_app.py diff --git a/pyproject.toml b/pyproject.toml index c90aae6..679815b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "aegis-gitea-mcp" -version = "0.1.0" -description = "Private, security-first MCP server for controlled AI access to self-hosted Gitea" +version = "0.2.0" +description = "Security-first MCP server for controlled AI access to self-hosted Gitea (local stdio + public HTTP/OAuth)" authors = [ {name = "AegisGitea MCP Contributors"} ] @@ -19,20 +19,27 @@ classifiers = [ "Programming Language :: Python :: 3.12", ] +# Core (default install) powers the local stdio transport. It deliberately +# excludes the web/OAuth stack so `uvx aegis-gitea-mcp` stays light; the HTTP +# server pulls those in via the [server] extra. dependencies = [ - "fastapi>=0.109.0", - "uvicorn[standard]>=0.27.0", "httpx>=0.26.0", "pydantic>=2.5.0", "pydantic-settings>=2.1.0", "PyYAML>=6.0.1", "python-dotenv>=1.0.0", "structlog>=24.1.0", - "python-multipart>=0.0.9", - "PyJWT[crypto]>=2.9.0", + "mcp>=1.2.0", ] [project.optional-dependencies] +# The public HTTP/OAuth server (aegis-gitea-mcp-server) needs the web stack. +server = [ + "fastapi>=0.109.0", + "uvicorn[standard]>=0.27.0", + "PyJWT[crypto]>=2.9.0", + "python-multipart>=0.0.9", +] dev = [ "pytest>=7.4.0", "pytest-asyncio>=0.23.0", @@ -44,11 +51,18 @@ dev = [ "pre-commit>=3.6.0", ] +[project.scripts] +# Local stdio MCP server (default install, no web stack required). +aegis-gitea-mcp = "aegis_gitea_mcp.stdio_app:main" +# Public HTTP/OAuth server; requires the [server] extra. The entry point guards +# against a missing web stack with an actionable message. +aegis-gitea-mcp-server = "aegis_gitea_mcp.server_entry:main" + [project.urls] -Homepage = "https://github.com/your-org/AegisGitea-MCP" -Documentation = "https://github.com/your-org/AegisGitea-MCP/blob/main/README.md" -Repository = "https://github.com/your-org/AegisGitea-MCP.git" -Issues = "https://github.com/your-org/AegisGitea-MCP/issues" +Homepage = "https://git.hiddenden.cafe/Hiddenden/AegisGitea-MCP" +Documentation = "https://git.hiddenden.cafe/Hiddenden/AegisGitea-MCP/src/branch/main/README.md" +Repository = "https://git.hiddenden.cafe/Hiddenden/AegisGitea-MCP.git" +Issues = "https://git.hiddenden.cafe/Hiddenden/AegisGitea-MCP/issues" [build-system] requires = ["setuptools>=68.0.0", "wheel"] diff --git a/requirements.txt b/requirements.txt index a614de2..1003e8d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,3 +8,4 @@ python-dotenv>=1.0.0 python-multipart>=0.0.9 structlog>=24.1.0 PyJWT[crypto]>=2.9.0 +mcp>=1.2.0 diff --git a/src/aegis_gitea_mcp/server_entry.py b/src/aegis_gitea_mcp/server_entry.py new file mode 100644 index 0000000..6e07202 --- /dev/null +++ b/src/aegis_gitea_mcp/server_entry.py @@ -0,0 +1,34 @@ +"""Guarded console-script entry point for the HTTP/OAuth server. + +The HTTP server (``aegis_gitea_mcp.server``) imports FastAPI/uvicorn at module +load. Those live in the optional ``[server]`` extra, so a default (local-only) +install would crash with a bare ``ModuleNotFoundError`` traceback if the +``aegis-gitea-mcp-server`` script were invoked. This thin wrapper imports nothing +from the web stack at module scope and degrades to an actionable message. +""" + +from __future__ import annotations + +import sys + + +def main() -> None: + """Run the HTTP server, or explain how to install the web stack.""" + try: + import fastapi # noqa: F401 + import uvicorn # noqa: F401 + except ModuleNotFoundError as exc: + print( + "aegis-gitea-mcp-server requires the web stack, which is not installed.\n" + "Install it with: pip install 'aegis-gitea-mcp[server]'", + file=sys.stderr, + ) + raise SystemExit(1) from exc + + from aegis_gitea_mcp.server import main as server_main + + server_main() + + +if __name__ == "__main__": + main() diff --git a/src/aegis_gitea_mcp/stdio_app.py b/src/aegis_gitea_mcp/stdio_app.py new file mode 100644 index 0000000..6d79a87 --- /dev/null +++ b/src/aegis_gitea_mcp/stdio_app.py @@ -0,0 +1,231 @@ +"""Local stdio transport adapter (``aegis-gitea-mcp``). + +This is the second transport for the shared core: a single-user, local MCP +server spoken over stdio using the official ``mcp`` SDK. It is meant to be run +like ``uvx aegis-gitea-mcp`` and wired into Claude Desktop / Claude Code, mirror- +ing the ergonomics of other local MCP servers. + +Trust model +----------- +The local operator owns the Gitea Personal Access Token supplied via +``GITEA_TOKEN``; there is no per-user OAuth. At startup the adapter resolves the +PAT owner (``GET /user``) and pins the request context to that single login. +Because the caller *is* the token owner, the per-user repository-permission +probe used by the public HTTP server is intentionally skipped — but the policy +engine, ``WRITE_MODE`` gate, secret sanitization and the tamper-evident audit +log all run exactly as they do on the server. The same tools (including +``gitea_request``) are served from the shared :mod:`aegis_gitea_mcp.registry`. +""" + +from __future__ import annotations + +import asyncio +import os +import sys +from pathlib import Path +from typing import Any + +from aegis_gitea_mcp.errors import ToolError + + +class StdioConfigError(RuntimeError): + """Raised when the local environment is missing required configuration.""" + + +def _default_audit_log_path() -> Path: + """Return a writable per-user audit-log path for local runs. + + The server's container default (``/var/log/aegis-mcp/audit.log``) is not + writable on a typical workstation, so fall back to an OS-appropriate user + state directory. + """ + if sys.platform == "win32": + base = os.environ.get("LOCALAPPDATA") or str(Path.home() / "AppData" / "Local") + return Path(base) / "aegis-gitea-mcp" / "audit.log" + xdg_state = os.environ.get("XDG_STATE_HOME") + base_dir = Path(xdg_state) if xdg_state else (Path.home() / ".local" / "state") + return base_dir / "aegis-gitea-mcp" / "audit.log" + + +def _bootstrap_env() -> None: + """Apply local-mode defaults to the environment before settings load. + + Local mode has no OAuth and no API-key gate (the operator is the trusted PAT + owner), and writes its audit log to a per-user path when one is not set. User + overrides via real env vars or ``.env`` always win for everything else. + """ + # python-dotenv: load a local .env so GITEA_URL/GITEA_TOKEN can live there. + try: + from dotenv import load_dotenv + + load_dotenv() + except Exception: # pragma: no cover - dotenv is a core dep, defensive only + pass + + # Local mode is single-user PAT auth: force OAuth off and disable the API-key + # requirement so the server's API-key/OAuth config validation does not apply. + os.environ["OAUTH_MODE"] = "false" + os.environ.setdefault("AUTH_ENABLED", "false") + os.environ.setdefault("STARTUP_VALIDATE_GITEA", "false") + + if not os.environ.get("AUDIT_LOG_PATH", "").strip(): + os.environ["AUDIT_LOG_PATH"] = str(_default_audit_log_path()) + + +def _check_required_env() -> None: + """Fail with an actionable message when required env vars are missing.""" + missing = [name for name in ("GITEA_URL", "GITEA_TOKEN") if not os.environ.get(name, "").strip()] + if missing: + raise StdioConfigError( + "Missing required environment variable(s): " + + ", ".join(missing) + + ".\nSet them in your environment or a local .env file, e.g.:\n" + " GITEA_URL=https://gitea.example.com\n" + " GITEA_TOKEN=\n" + ) + + +# The PAT owner login, resolved once at startup and pinned onto every dispatch. +_owner_login: str | None = None + + +async def _resolve_owner_login() -> str: + """Resolve and cache the Gitea login that owns the configured PAT.""" + from aegis_gitea_mcp.config import get_settings + from aegis_gitea_mcp.gitea_client import GiteaClient + + settings = get_settings() + async with GiteaClient(token=settings.gitea_token) as gitea: + user = await gitea.get_current_user() + login = str(user.get("login", "")).strip() + if not login: + raise StdioConfigError( + "Could not resolve the Gitea user for the supplied GITEA_TOKEN. " + "Verify the token is valid and has API access." + ) + return login + + +async def _dispatch(tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]: + """Execute a tool with the same policy/audit/sanitize guarantees as the server. + + The per-user repository-permission probe is intentionally omitted: the local + operator is the PAT owner. Everything else — policy engine, ``WRITE_MODE``, + the ``gitea_request`` per-method authorization, secret sanitization and audit + logging — runs identically to the HTTP adapter. + """ + from aegis_gitea_mcp.audit import get_audit_logger + from aegis_gitea_mcp.config import get_settings + from aegis_gitea_mcp.gitea_client import GiteaClient + from aegis_gitea_mcp.policy import get_policy_engine + from aegis_gitea_mcp.registry import get_tool_by_name, get_tool_handler + from aegis_gitea_mcp.request_context import set_gitea_user_login + from aegis_gitea_mcp.security import sanitize_data + from aegis_gitea_mcp.tools.arguments import extract_repository, extract_target_path + + # Pin identity to the trusted PAT owner for every call (e.g. list_repositories + # scopes its results to this login in service-PAT mode). + if _owner_login: + set_gitea_user_login(_owner_login) + + settings = get_settings() + audit = get_audit_logger() + + tool_def = get_tool_by_name(tool_name) + if tool_def is None: + raise ToolError(f"Tool '{tool_name}' not found", status_code=404) + handler = get_tool_handler(tool_name) + if handler is None: + raise ToolError(f"Tool '{tool_name}' has no handler implementation", status_code=500) + + repository = extract_repository(arguments) + target_path = extract_target_path(arguments) + decision = get_policy_engine().authorize( + tool_name=tool_name, + is_write=tool_def.write_operation, + repository=repository, + target_path=target_path, + ) + if not decision.allowed: + audit.log_access_denied(tool_name=tool_name, repository=repository, reason=decision.reason) + raise ToolError(f"Policy denied request: {decision.reason}", status_code=403) + + correlation_id = audit.log_tool_invocation(tool_name=tool_name, params=arguments) + async with GiteaClient(token=settings.gitea_token) as gitea: + result = await handler(gitea, arguments) + + if settings.secret_detection_mode != "off": + result = sanitize_data(result, mode=settings.secret_detection_mode) + + audit.log_tool_invocation( + tool_name=tool_name, correlation_id=correlation_id, result_status="success" + ) + return result + + +async def _serve() -> None: + """Build the stdio MCP server from the shared registry and serve it.""" + import mcp.types as mcp_types + from mcp.server import Server + from mcp.server.stdio import stdio_server + + from aegis_gitea_mcp.config import get_settings + from aegis_gitea_mcp.policy import get_policy_engine + from aegis_gitea_mcp.registry import list_tool_definitions + + # Fail fast on bad settings/policy before opening the transport. + get_settings() + get_policy_engine() + + global _owner_login + _owner_login = await _resolve_owner_login() + + server: Server = Server("aegis-gitea-mcp") + + @server.list_tools() + async def list_tools() -> list[mcp_types.Tool]: + return [ + mcp_types.Tool( + name=tool.name, + description=tool.description, + inputSchema=tool.input_schema, + ) + for tool in list_tool_definitions() + ] + + @server.call_tool() + async def call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]: + # Returning a dict yields structured content plus a JSON text block. + return await _dispatch(name, arguments) + + async with stdio_server() as (read_stream, write_stream): + await server.run(read_stream, write_stream, server.create_initialization_options()) + + +def main() -> None: + """Console-script entry point for the local stdio MCP server.""" + _bootstrap_env() + try: + _check_required_env() + except StdioConfigError as exc: + print(f"aegis-gitea-mcp: {exc}", file=sys.stderr) + raise SystemExit(2) from exc + + try: + from aegis_gitea_mcp.config import get_settings + + get_settings() + except Exception as exc: # pydantic ValidationError or PolicyError + print(f"aegis-gitea-mcp: invalid configuration: {exc}", file=sys.stderr) + raise SystemExit(2) from exc + + try: + asyncio.run(_serve()) + except StdioConfigError as exc: + print(f"aegis-gitea-mcp: {exc}", file=sys.stderr) + raise SystemExit(2) from exc + except KeyboardInterrupt: # pragma: no cover - interactive shutdown + pass + + +__all__ = ["main", "StdioConfigError"]