diff --git a/src/aegis_gitea_mcp/cli.py b/src/aegis_gitea_mcp/cli.py index 0626802..c27da20 100644 --- a/src/aegis_gitea_mcp/cli.py +++ b/src/aegis_gitea_mcp/cli.py @@ -10,7 +10,7 @@ import argparse import sys from collections.abc import Sequence -from aegis_gitea_mcp import __version__ +from aegis_gitea_mcp import __version__, setup_cmd def build_parser() -> argparse.ArgumentParser: @@ -29,7 +29,8 @@ def build_parser() -> argparse.ArgumentParser: help="transport to serve (default: MCP_TRANSPORT, else stdio)", ) - subparsers.add_parser("setup", help="interactive setup against a Gitea instance") + setup = subparsers.add_parser("setup", help="verify a Gitea instance and write the env file") + setup_cmd.add_arguments(setup) subparsers.add_parser("version", help="print the version and exit") return parser @@ -48,6 +49,8 @@ def main(argv: Sequence[str] | None = None) -> int: if command == "version": print(__version__) return 0 + if command == "setup": + return setup_cmd.run_setup(args) return _not_implemented(command) diff --git a/src/aegis_gitea_mcp/setup_cmd.py b/src/aegis_gitea_mcp/setup_cmd.py new file mode 100644 index 0000000..5ca60eb --- /dev/null +++ b/src/aegis_gitea_mcp/setup_cmd.py @@ -0,0 +1,853 @@ +"""``aegis-gitea-mcp setup``: verify a Gitea instance, register the OAuth app, write ``.env``. + +The command talks to Gitea over its public REST API only, so it behaves the same +against a mocked server and a real instance. It never prints or logs a secret: +the Gitea token, the generated MCP bearer token, the token encryption key and the +OAuth client secret only ever go into the env file (mode 0600). +""" + +from __future__ import annotations + +import argparse +import getpass +import os +import re +import secrets +import socket +import ssl +import stat +import sys +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any +from urllib.parse import urlsplit + +import httpx + +from aegis_gitea_mcp import setup_snippets as snippets +from aegis_gitea_mcp.config import ( + _MIN_SECRET_CHARS, + GITEA_SCOPE_CATEGORIES, + Settings, + is_loopback_host, +) +from aegis_gitea_mcp.setup_template import ENV_TEMPLATE + +OAUTH_APP_NAME = "aegis-gitea-mcp" +OAUTH_CALLBACK_PATH = "/oauth/callback" +REQUEST_TIMEOUT_SECONDS = 15.0 +PAGE_SIZE = 50 +MAX_PAGES = 100 +PLACEHOLDER_GITEA_HOSTS = frozenset({"gitea.example.com"}) + +#: Scopes a token needs so every Gitea API operation works. +REQUIRED_SCOPES = tuple(f"write:{category}" for category in GITEA_SCOPE_CATEGORIES) + +_ASSIGNMENT = re.compile(r"^(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)\s*=(.*)$") +_SAFE_VALUE = re.compile(r"^[^\s\"'#$\\]+$") + + +class SetupError(Exception): + """A setup step failed; the message is safe to print (it never contains a secret).""" + + +# --------------------------------------------------------------------------- +# env file handling +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class EnvEntry: + """One assignment of an existing env file.""" + + key: str + raw_value: str + value: str + + +def _unquote(raw: str) -> str: + """Return the effective value of a dotenv right-hand side.""" + text = raw.strip() + if text[:1] in {'"', "'"}: + quote = text[0] + end = text.find(quote, 1) + if end != -1: + return text[1:end] + return text[1:] + match = re.search(r"\s#", text) + if match: + text = text[: match.start()] + return text.strip() + + +def parse_env_text(text: str) -> dict[str, EnvEntry]: + """Parse env file text into entries keyed by upper-cased name (last one wins).""" + entries: dict[str, EnvEntry] = {} + for line in text.splitlines(): + if line.lstrip().startswith("#"): + continue + match = _ASSIGNMENT.match(line.strip()) + if not match: + continue + key = match.group(1).upper() + entries[key] = EnvEntry(key=key, raw_value=match.group(2), value=_unquote(match.group(2))) + return entries + + +def render_env(template: str, values: dict[str, str], existing: dict[str, EnvEntry]) -> str: + """Fill ``template`` keeping its comments and order. + + Precedence per key: value computed by setup, then the existing file's value + (verbatim), then the template default. Existing keys the template does not + know are preserved in a trailing block. + """ + lines: list[str] = [] + seen: set[str] = set() + for line in template.splitlines(): + match = None if line.lstrip().startswith("#") else _ASSIGNMENT.match(line) + if match is None: + lines.append(line) + continue + key = match.group(1).upper() + seen.add(key) + if key in values: + lines.append(f"{key}={values[key]}") + elif key in existing: + lines.append(f"{key}={existing[key].raw_value}") + else: + lines.append(line) + + extra_values = [f"{key}={value}" for key, value in values.items() if key not in seen] + extra_existing = [ + f"{entry.key}={entry.raw_value}" + for key, entry in existing.items() + if key not in seen and key not in values + ] + if extra_values or extra_existing: + lines.append("") + lines.append("# --- Added by setup or preserved from the previous file ---") + lines.extend(extra_values) + lines.extend(extra_existing) + return "\n".join(lines) + "\n" + + +def check_env_target(path: Path) -> None: + """Refuse a symlink, a non-regular file or an unwritable location.""" + if path.is_symlink(): + raise SetupError( + f"{path} is a symlink; refusing to write through it. " + "Point --env-file at a regular file or remove the link." + ) + if path.exists() and not path.is_file(): + raise SetupError(f"{path} exists but is not a regular file") + if not path.parent.is_dir(): + raise SetupError(f"directory {path.parent} does not exist") + if not os.access(path.parent, os.W_OK): + raise SetupError(f"directory {path.parent} is not writable") + + +def write_env_atomic(path: Path, content: str) -> None: + """Write ``content`` to ``path`` atomically with mode 0600 (set before the content).""" + check_env_target(path) + fd, tmp_name = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp") + tmp = Path(tmp_name) + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(content) + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp, path) + except BaseException: + tmp.unlink(missing_ok=True) + raise + + +# --------------------------------------------------------------------------- +# Gitea API +# --------------------------------------------------------------------------- + + +@dataclass +class OAuthApp: + """The fields of a Gitea OAuth2 application that setup cares about.""" + + id: int + name: str + client_id: str + redirect_uris: list[str] + confidential_client: bool + client_secret: str = field(default="", repr=False) + + +class GiteaApi: + """Tiny synchronous Gitea client. Never puts the token into errors or output.""" + + def __init__(self, base_url: str, token: str, client: httpx.Client) -> None: + self._api = f"{base_url}/api/v1" + self._base_url = base_url + self._client = client + self._headers = {"Authorization": f"token {token}", "Accept": "application/json"} + self._secrets: list[str] = [token] + + def add_secret(self, value: str) -> None: + """Register a value that must be redacted from any error text.""" + if value and value not in self._secrets: + self._secrets.append(value) + + def redact(self, text: str) -> str: + """Replace every known secret in ``text``.""" + for value in self._secrets: + text = text.replace(value, "***") + return text + + def request( + self, + method: str, + path: str, + *, + params: dict[str, Any] | None = None, + json: Any = None, + ) -> httpx.Response: + """Send one request; transport errors become :class:`SetupError`.""" + url = f"{self._api}{path}" + try: + response = self._client.request( + method, url, params=params, json=json, headers=self._headers + ) + except httpx.HTTPError as exc: + raise SetupError(self._explain_transport_error(exc)) from None + if response.is_redirect: + location = self.redact(response.headers.get("location", "?"))[:200] + raise SetupError( + f"Gitea answered {response.status_code} with a redirect to {location}. " + "Redirects are not followed so the token cannot leave the host you named. " + "Use the final URL (for example https instead of http) as the Gitea URL." + ) + return response + + def detail(self, response: httpx.Response) -> str: + """Return a short, redacted, printable description of an error response.""" + message = "" + try: + body = response.json() + if isinstance(body, dict) and isinstance(body.get("message"), str): + message = body["message"] + except ValueError: + message = "" + message = re.sub(r"[\x00-\x1f\x7f]", " ", self.redact(message)).strip()[:200] + return f"HTTP {response.status_code}" + (f": {message}" if message else "") + + def json_object(self, response: httpx.Response, what: str) -> dict[str, Any]: + """Decode a JSON object body or raise a helpful error.""" + try: + body = response.json() + except ValueError: + body = None + if not isinstance(body, dict): + raise SetupError( + f"{what} did not return Gitea API JSON. Check that the URL is the Gitea " + "root (not a login page or a proxy in front of it)." + ) + return body + + def _explain_transport_error(self, exc: httpx.HTTPError) -> str: + host = urlsplit(self._base_url).hostname or self._base_url + chain: list[BaseException] = [] + cursor: BaseException | None = exc + while cursor is not None and len(chain) < 8: + chain.append(cursor) + cursor = cursor.__cause__ or cursor.__context__ + if any(isinstance(item, socket.gaierror) for item in chain): + hint = f"DNS lookup for {host} failed. Check the host name in the URL." + elif any(isinstance(item, ssl.SSLError) for item in chain): + hint = ( + f"the TLS handshake with {host} failed (certificate not trusted, expired or " + "for another name, or the server speaks plain http). Check the URL scheme " + "and the certificate." + ) + elif isinstance(exc, httpx.TimeoutException): + hint = f"{host} did not answer within {int(REQUEST_TIMEOUT_SECONDS)} seconds." + elif any(isinstance(item, ConnectionRefusedError) for item in chain): + hint = f"connection to {host} was refused. Check the port and that Gitea is running." + else: + hint = f"could not talk to {host} ({type(exc).__name__})." + return f"Gitea is not reachable: {hint}" + + +def missing_scopes(scopes: list[str]) -> list[str]: + """Return the required scopes a token lacks (``all`` grants everything).""" + granted = {scope.strip().lower() for scope in scopes} + if "all" in granted: + return [] + return [scope for scope in REQUIRED_SCOPES if scope not in granted] + + +# --------------------------------------------------------------------------- +# setup steps +# --------------------------------------------------------------------------- + + +@dataclass +class Report: + """Non-secret facts collected for the final summary.""" + + gitea_url: str = "" + version: str = "" + login: str = "" + scope_note: str = "" + oauth_note: str = "" + oauth_app_id: int | None = None + oauth_client_id: str = "" + redirect_uri: str = "" + warnings: list[str] = field(default_factory=list) + notes: list[str] = field(default_factory=list) + + +def _say(message: str = "") -> None: + print(message) + + +def _validate_gitea_url(value: str) -> str: + try: + return Settings._validate_gitea_url(value) + except ValueError as exc: + raise SetupError(f"invalid Gitea URL: {exc}") from None + + +def _validate_public_url(value: str) -> str: + try: + validated = Settings._validate_public_base_url(value) + except ValueError as exc: + raise SetupError(f"invalid public base URL: {exc}") from None + assert validated is not None # noqa: S101 (value is a non-empty string here) + return validated + + +def _validate_token_text(token: str) -> str: + token = token.strip() + if not token: + raise SetupError("no Gitea token given") + if not _SAFE_VALUE.match(token): + raise SetupError( + "the Gitea token contains whitespace or one of the characters " + "\" ' # $ \\ that cannot be stored safely in an env file" + ) + return token + + +def _check_value_text(name: str, value: str) -> str: + if not _SAFE_VALUE.match(value): + raise SetupError(f"{name} contains characters that cannot be stored safely in an env file") + return value + + +def _resolve_gitea_url(args: argparse.Namespace, existing: dict[str, EnvEntry]) -> str: + candidate = args.gitea_url or os.environ.get("GITEA_URL", "").strip() + if not candidate and "GITEA_URL" in existing: + stored = existing["GITEA_URL"].value + host = urlsplit(stored).hostname or "" + if stored and host not in PLACEHOLDER_GITEA_HOSTS: + candidate = stored + if not candidate: + if args.yes: + raise SetupError("no Gitea URL given. Pass --gitea-url or set GITEA_URL.") + candidate = input("Gitea URL (for example https://git.example.com): ").strip() + if not candidate: + raise SetupError("no Gitea URL given") + return _validate_gitea_url(candidate) + + +def _resolve_token(args: argparse.Namespace, existing: dict[str, EnvEntry]) -> str: + token = os.environ.get("GITEA_TOKEN", "").strip() + if not token and "GITEA_TOKEN" in existing: + token = existing["GITEA_TOKEN"].value.strip() + if not token: + if args.yes: + raise SetupError("no Gitea token. Set GITEA_TOKEN in the environment or the env file.") + token = getpass.getpass("Gitea admin token (input is hidden): ") + return _validate_token_text(token) + + +def _resolve_public_url(args: argparse.Namespace, existing: dict[str, EnvEntry]) -> str | None: + candidate = args.public_base_url or os.environ.get("PUBLIC_BASE_URL", "").strip() + if not candidate and "PUBLIC_BASE_URL" in existing: + candidate = existing["PUBLIC_BASE_URL"].value.strip() + if not candidate and not args.yes and not args.no_oauth and sys.stdin.isatty(): + candidate = input("Public base URL of this MCP server (blank skips OAuth): ").strip() + return _validate_public_url(candidate) if candidate else None + + +def verify_instance(api: GiteaApi, report: Report) -> None: + """Check reachability, token validity, admin rights and scopes (raises SetupError).""" + response = api.request("GET", "/version") + if response.status_code != 200: + raise SetupError( + f"GET /api/v1/version failed ({api.detail(response)}). " + "Check that the URL is the root of a Gitea instance." + ) + version = api.json_object(response, "GET /api/v1/version").get("version") + if not isinstance(version, str) or not version: + raise SetupError("GET /api/v1/version did not report a Gitea version") + report.version = version[:64] + _say(f"Gitea reachable: version {report.version}") + + response = api.request("GET", "/user") + if response.status_code in (401, 403): + raise SetupError( + f"token rejected by Gitea ({api.detail(response)}). " + "Check that the token is correct, not expired and not revoked." + ) + if response.status_code != 200: + raise SetupError(f"GET /api/v1/user failed ({api.detail(response)})") + user = api.json_object(response, "GET /api/v1/user") + login = user.get("login") + report.login = login if isinstance(login, str) else "?" + if user.get("is_admin") is not True: + raise SetupError( + f"the token belongs to '{report.login}', who is not a site administrator. " + "The admin features need a site-admin token. Sign in to Gitea as a site admin, " + "open Settings > Applications > Generate New Token, set every scope to Read and " + "Write (or choose all), then run setup again with that token." + ) + _say(f"Token valid: {report.login} (site administrator)") + _check_scopes(api, report) + + +def _check_scopes(api: GiteaApi, report: Report) -> None: + response = api.request("GET", "/token") + if response.status_code != 200: + report.scope_note = "not verified" + report.warnings.append( + f"could not read the token scopes (GET /api/v1/token: {api.detail(response)}). " + "This happens for OAuth tokens or older Gitea versions. Make sure the token has " + "the 'all' scope or every write scope." + ) + return + try: + body = response.json() + except ValueError: + body = None + scopes = body.get("scopes") if isinstance(body, dict) else None + if not isinstance(scopes, list) or not scopes or not all(isinstance(s, str) for s in scopes): + report.scope_note = "not verified" + report.warnings.append( + "Gitea did not report scopes for this token, so they could not be verified." + ) + return + missing = missing_scopes(scopes) + if missing: + raise SetupError( + "the token lacks required scopes: " + + ", ".join(missing) + + ". Create a new token with the 'all' scope (or every write scope) under " + "Settings > Applications and run setup again." + ) + lowered = {s.lower() for s in scopes} + report.scope_note = "all" if "all" in lowered else "all write scopes" + if "public-only" in lowered: + report.warnings.append( + "the token has the public-only scope, so private resources stay unreachable." + ) + _say(f"Token scopes ok: {report.scope_note}") + + +def _parse_app(item: Any) -> OAuthApp | None: + if not isinstance(item, dict): + return None + app_id, name, client_id = item.get("id"), item.get("name"), item.get("client_id") + if not isinstance(app_id, int) or not isinstance(name, str) or not isinstance(client_id, str): + return None + uris = item.get("redirect_uris") + secret = item.get("client_secret") + return OAuthApp( + id=app_id, + name=name, + client_id=client_id, + redirect_uris=[u for u in uris if isinstance(u, str)] if isinstance(uris, list) else [], + confidential_client=item.get("confidential_client") is True, + client_secret=secret if isinstance(secret, str) else "", + ) + + +def list_oauth_apps(api: GiteaApi) -> list[OAuthApp]: + """List the admin's OAuth2 applications (all pages).""" + apps: list[OAuthApp] = [] + for page in range(1, MAX_PAGES + 1): + response = api.request( + "GET", "/user/applications/oauth2", params={"page": page, "limit": PAGE_SIZE} + ) + if response.status_code != 200: + raise SetupError(f"could not list OAuth applications ({api.detail(response)})") + try: + body = response.json() + except ValueError: + body = None + if not isinstance(body, list): + raise SetupError("could not list OAuth applications: unexpected response") + batch = [app for app in (_parse_app(item) for item in body) if app is not None] + apps.extend(batch) + total = response.headers.get("x-total-count", "") + if not body or (total.isdigit() and len(apps) >= int(total)): + break + return apps + + +def _app_payload(redirect_uri: str) -> dict[str, Any]: + return { + "name": OAUTH_APP_NAME, + "redirect_uris": [redirect_uri], + "confidential_client": True, + "skip_secondary_authorization": False, + } + + +def _app_from_write(api: GiteaApi, response: httpx.Response, ok: int, action: str) -> OAuthApp: + if response.status_code == 403: + raise SetupError( + f"Gitea refused to {action} the OAuth application ({api.detail(response)}). " + "The token needs the write:user scope (or all)." + ) + if response.status_code != ok: + raise SetupError(f"could not {action} the OAuth application ({api.detail(response)})") + try: + body = response.json() + except ValueError: + body = None + app = _parse_app(body) + if app is None or not app.client_secret: + raise SetupError(f"Gitea did not return a client id and secret after the {action}") + api.add_secret(app.client_secret) + return app + + +def _confirm(question: str, *, yes: bool) -> bool: + if yes: + return True + if not sys.stdin.isatty(): + return False + return input(f"{question} [y/N] ").strip().lower() in {"y", "yes"} + + +def ensure_oauth_app( + api: GiteaApi, + redirect_uri: str, + existing: dict[str, EnvEntry], + *, + yes: bool, + report: Report, +) -> tuple[str, str]: + """Create, reuse or rotate the ``aegis-gitea-mcp`` app. Returns (client id, client secret).""" + known_id = ( + existing["GITEA_OAUTH_CLIENT_ID"].value if "GITEA_OAUTH_CLIENT_ID" in existing else "" + ) + known_secret = ( + existing["GITEA_OAUTH_CLIENT_SECRET"].value + if "GITEA_OAUTH_CLIENT_SECRET" in existing + else "" + ) + if known_secret: + api.add_secret(known_secret) + + matches = [app for app in list_oauth_apps(api) if app.name == OAUTH_APP_NAME] + chosen = next((app for app in matches if known_id and app.client_id == known_id), None) + if chosen is None and matches: + chosen = matches[0] + if len(matches) > 1: + report.warnings.append( + f"{len(matches)} OAuth applications are named {OAUTH_APP_NAME}; using id " + f"{chosen.id if chosen else '?'}. Delete the extra ones in Gitea." + ) + + if chosen is None: + app = _app_from_write( + api, + api.request("POST", "/user/applications/oauth2", json=_app_payload(redirect_uri)), + 201, + "create", + ) + report.oauth_note = "created" + else: + secret_known = bool(known_id and known_secret and chosen.client_id == known_id) + up_to_date = chosen.redirect_uris == [redirect_uri] and chosen.confidential_client + if secret_known and up_to_date: + app = chosen + app.client_secret = known_secret + report.oauth_note = "reused" + else: + reason = ( + "redirect URIs differ" + if secret_known + else "its client secret is not in the env file" + ) + if not _confirm( + f"The OAuth app {OAUTH_APP_NAME} (id {chosen.id}) must be updated because " + f"{reason}. Gitea regenerates the client secret on update, so anything using " + "the old secret stops working. Continue?", + yes=yes, + ): + raise SetupError( + "the existing OAuth application needs its client secret regenerated. " + "Re-run with --yes to allow it, or restore the old client secret in the " + "env file." + ) + app = _app_from_write( + api, + api.request( + "PATCH", + f"/user/applications/oauth2/{chosen.id}", + json=_app_payload(redirect_uri), + ), + 200, + "update", + ) + report.oauth_note = "updated, secret regenerated" if secret_known else "secret rotated" + dropped = [u for u in chosen.redirect_uris if u != redirect_uri] + if dropped: + report.notes.append( + f"removed {len(dropped)} other redirect URI(s) from the OAuth application" + ) + report.oauth_app_id = app.id + report.oauth_client_id = app.client_id + report.redirect_uri = redirect_uri + _say(f"OAuth application {OAUTH_APP_NAME}: {report.oauth_note} (id {app.id})") + return app.client_id, app.client_secret + + +# --------------------------------------------------------------------------- +# output +# --------------------------------------------------------------------------- + + +def _print_summary( + report: Report, + *, + env_file: Path, + created: bool, + tightened: bool, + generated: list[str], + oauth_enabled: bool, + public_url: str | None, +) -> None: + env_name = str(env_file) + _say() + _say("Setup complete") + _say(f" Instance: {report.gitea_url} (Gitea {report.version})") + _say(f" Admin: {report.login}") + _say(f" Scopes: {report.scope_note or 'not verified'}") + if oauth_enabled: + _say( + f" OAuth app: {OAUTH_APP_NAME}, id {report.oauth_app_id}, " + f"client id {report.oauth_client_id} ({report.oauth_note})" + ) + _say(f" Redirect: {report.redirect_uri}") + else: + _say(" OAuth: disabled (OAUTH_ENABLED=false)") + _say(f" Env file: {env_name} ({'created' if created else 'updated'}, mode 0600)") + if generated: + _say(f" Generated: {', '.join(generated)}") + if tightened: + _say(" The existing env file was readable by others; permissions are now 0600.") + for note in report.notes: + _say(f" Note: {note}") + for warning in report.warnings: + _say(f" Warning: {warning}") + _say() + _say("Secrets are stored in the env file only. To read one:") + _say(" MCP_BEARER_TOKEN: " + snippets.read_secret_command(env_name, "MCP_BEARER_TOKEN")) + _say(" Use the value of MCP_BEARER_TOKEN in the env file wherever a bearer token is asked.") + + _say() + _say("== Local stdio clients (the server runs on your machine) ==") + _say("Claude Code (replace the token placeholder with your Gitea token):") + _say(f" {snippets.claude_code_stdio(report.gitea_url)}") + _say() + _say("Claude Desktop, Cursor and other clients (mcpServers JSON):") + _say(snippets.mcp_servers_json(report.gitea_url)) + + base = public_url or snippets.LOCAL_BASE_URL + _say() + _say(f"== Remote streamable HTTP ({base}) ==") + _say("Claude Code (static bearer, the MCP_BEARER_TOKEN value from the env file):") + _say(f" {snippets.claude_code_http(base)}") + _say() + _say("claude.ai custom connector:") + _say(snippets.claude_ai_connector(base, oauth=oauth_enabled)) + _say() + _say("Mistral Le Chat custom connector:") + _say(snippets.le_chat_connector(base, oauth=oauth_enabled)) + + _say() + _say("== docker-compose.yml ==") + _say(snippets.docker_compose()) + if public_url: + _say("== Caddyfile (terminates TLS) ==") + _say(snippets.caddy_snippet(public_url)) + _say(snippets.NGINX_NOTE) + _say() + _say("== Start ==") + _say(" stdio: aegis-gitea-mcp serve") + _say(" HTTP: MCP_TRANSPORT=http aegis-gitea-mcp serve") + _say(f" Check: curl {base}/health") + + +# --------------------------------------------------------------------------- +# entry points +# --------------------------------------------------------------------------- + + +def add_arguments(parser: argparse.ArgumentParser) -> None: + """Add the ``setup`` options to ``parser``.""" + parser.add_argument("--gitea-url", default=None, help="Gitea base URL") + parser.add_argument( + "--public-base-url", + default=None, + help="public https URL of this MCP server (enables the OAuth app)", + ) + parser.add_argument( + "--env-file", default=".env", help="env file to create or update (default: ./.env)" + ) + parser.add_argument( + "--template", + default=None, + help="env template (default: ./.env.example, else the copy built into the package)", + ) + parser.add_argument( + "--no-oauth", action="store_true", help="do not create the Gitea OAuth application" + ) + parser.add_argument( + "--yes", + action="store_true", + help="never prompt; accept actions such as regenerating the OAuth client secret", + ) + + +def _load_template(arg: str | None) -> str: + if arg: + path = Path(arg) + try: + return path.read_text(encoding="utf-8") + except OSError as exc: + raise SetupError(f"cannot read template {path}: {exc.strerror or exc}") from None + default = Path(".env.example") + try: + if default.is_file(): + return default.read_text(encoding="utf-8") + except OSError: + pass + return ENV_TEMPLATE + + +def _read_existing(path: Path) -> tuple[dict[str, EnvEntry], int | None]: + """Return the existing entries and the file's permission bits (None when absent).""" + if not path.exists(): + return {}, None + try: + mode = stat.S_IMODE(path.stat().st_mode) + return parse_env_text(path.read_text(encoding="utf-8")), mode + except OSError as exc: + raise SetupError(f"cannot read {path}: {exc.strerror or exc}") from None + + +def _generated(existing: dict[str, EnvEntry], key: str, values: dict[str, str]) -> bool: + """Add a fresh random value for ``key`` unless one exists. Returns True when generated.""" + current = existing[key].value if key in existing else "" + if current: + return False + values[key] = secrets.token_urlsafe(48) + return True + + +def _run(args: argparse.Namespace) -> int: + env_file = Path(args.env_file) + check_env_target(env_file) + template = _load_template(args.template) + existing, old_mode = _read_existing(env_file) + + report = Report() + report.gitea_url = _resolve_gitea_url(args, existing) + token = _resolve_token(args, existing) + public_url = _resolve_public_url(args, existing) + if urlsplit(report.gitea_url).scheme == "http" and not _is_loopback(report.gitea_url): + report.warnings.append("the Gitea URL uses plain http, so the token travels unencrypted.") + + oauth_wanted = not args.no_oauth and public_url is not None + _say(f"Using {report.gitea_url}") + + with httpx.Client(timeout=REQUEST_TIMEOUT_SECONDS, follow_redirects=False) as client: + api = GiteaApi(report.gitea_url, token, client) + verify_instance(api, report) + values: dict[str, str] = {"GITEA_URL": report.gitea_url, "GITEA_TOKEN": token} + if public_url: + values["PUBLIC_BASE_URL"] = public_url + if oauth_wanted: + assert public_url is not None # noqa: S101 + client_id, client_secret = ensure_oauth_app( + api, + f"{public_url}{OAUTH_CALLBACK_PATH}", + existing, + yes=args.yes, + report=report, + ) + values["GITEA_OAUTH_CLIENT_ID"] = _check_value_text("client id", client_id) + values["GITEA_OAUTH_CLIENT_SECRET"] = _check_value_text("client secret", client_secret) + else: + if not args.no_oauth: + report.notes.append( + "no public base URL given, so the OAuth application was skipped" + ) + else: + report.notes.append("OAuth skipped (--no-oauth)") + values["OAUTH_ENABLED"] = "true" if oauth_wanted else "false" + + generated = [ + key + for key in ("MCP_BEARER_TOKEN", "TOKEN_ENCRYPTION_KEY") + if _generated(existing, key, values) + ] + for key in ("MCP_BEARER_TOKEN", "TOKEN_ENCRYPTION_KEY"): + stored = existing[key].value if key in existing else "" + if stored and len(stored) < _MIN_SECRET_CHARS: + report.warnings.append( + f"{key} in the env file is shorter than {_MIN_SECRET_CHARS} characters; " + "the server will refuse it. Replace it with a longer value." + ) + + content = render_env(template, values, existing) + try: + write_env_atomic(env_file, content) + except OSError as exc: + raise SetupError( + f"could not write {env_file}: {exc.strerror or exc}. " + "If the OAuth application was just created or updated, run setup again; " + "that regenerates its client secret." + ) from None + + tightened = old_mode is not None and bool(old_mode & 0o077) + _print_summary( + report, + env_file=env_file, + created=old_mode is None, + tightened=tightened, + generated=generated, + oauth_enabled=oauth_wanted, + public_url=public_url, + ) + return 0 + + +def _is_loopback(url: str) -> bool: + return is_loopback_host(urlsplit(url).hostname or "") + + +def run_setup(args: argparse.Namespace) -> int: + """Run ``aegis-gitea-mcp setup`` and return the process exit code.""" + try: + return _run(args) + except SetupError as exc: + print(f"setup failed: {exc}", file=sys.stderr) + return 1 + except (KeyboardInterrupt, EOFError): + print("setup aborted", file=sys.stderr) + return 1 diff --git a/src/aegis_gitea_mcp/setup_snippets.py b/src/aegis_gitea_mcp/setup_snippets.py new file mode 100644 index 0000000..6a0a0f6 --- /dev/null +++ b/src/aegis_gitea_mcp/setup_snippets.py @@ -0,0 +1,143 @@ +"""Client and deployment snippets printed by ``aegis-gitea-mcp setup``. + +Every function returns plain text and takes only non-secret inputs. Secrets are +always shown as placeholders such as ````, so these templates +can be printed, logged and reused in the documentation without leaking anything. +""" + +from __future__ import annotations + +import json +import shlex +from urllib.parse import urlsplit + +#: Name used for the server in client configurations. +SERVER_NAME = "aegis-gitea" +#: Container image used by the docker-compose snippet. +DOCKER_IMAGE = "git.hiddenden.cafe/hiddenden/aegisgitea-mcp:dev" +TOKEN_PLACEHOLDER = "" # noqa: S105 +BEARER_PLACEHOLDER = "" +#: Loopback base URL used when no public base URL is configured. +LOCAL_BASE_URL = "http://127.0.0.1:8080" + + +def claude_code_stdio(gitea_url: str) -> str: + """Return the ``claude mcp add`` command for the local stdio server.""" + return ( + "claude mcp add --transport stdio " + f"--env GITEA_URL={gitea_url} --env GITEA_TOKEN={TOKEN_PLACEHOLDER} " + f"{SERVER_NAME} -- uvx aegis-gitea-mcp" + ) + + +def mcp_servers_json(gitea_url: str) -> str: + """Return a generic ``mcpServers`` JSON block (Claude Desktop, Cursor, and similar).""" + block = { + "mcpServers": { + SERVER_NAME: { + "command": "uvx", + "args": ["aegis-gitea-mcp"], + "env": {"GITEA_URL": gitea_url, "GITEA_TOKEN": TOKEN_PLACEHOLDER}, + } + } + } + return json.dumps(block, indent=2) + + +def mcp_endpoint(base_url: str, mcp_path: str = "/mcp") -> str: + """Return the full MCP endpoint URL for a base URL.""" + return f"{base_url.rstrip('/')}{mcp_path}" + + +def claude_code_http(base_url: str, mcp_path: str = "/mcp") -> str: + """Return the ``claude mcp add`` command for the streamable-HTTP endpoint.""" + return ( + f"claude mcp add --transport http {SERVER_NAME} {mcp_endpoint(base_url, mcp_path)} " + f'--header "Authorization: Bearer {BEARER_PLACEHOLDER}"' + ) + + +def claude_ai_connector(base_url: str, mcp_path: str = "/mcp", *, oauth: bool) -> str: + """Return setup instructions for a claude.ai custom connector.""" + lines = [ + "Settings > Connectors > Add custom connector", + f" URL: {mcp_endpoint(base_url, mcp_path)}", + ] + if oauth: + lines.append( + " Authentication: leave the OAuth client fields empty. Claude registers itself " + "(dynamic client registration) and sends you to Gitea to sign in." + ) + else: + lines.append( + " Authentication: OAuth is off. Use a static bearer token where your plan " + "supports it, or enable OAuth and re-run setup." + ) + return "\n".join(lines) + + +def le_chat_connector(base_url: str, mcp_path: str = "/mcp", *, oauth: bool) -> str: + """Return setup instructions for a Mistral Le Chat custom connector.""" + lines = [ + "Intelligence > Connectors > Add Connector > Custom MCP Connector", + f" URL: {mcp_endpoint(base_url, mcp_path)} (HTTPS with a valid certificate)", + f" Authentication: HTTP Bearer Token, with the {BEARER_PLACEHOLDER} value", + ] + if oauth: + lines.append( + " or OAuth 2.1 with dynamic client registration (users sign in with their " + "own Gitea account)." + ) + return "\n".join(lines) + + +def docker_compose(port: int = 8080) -> str: + """Return a hardened docker-compose service for the HTTP transport.""" + return f"""\ +services: + aegis-gitea-mcp: + image: {DOCKER_IMAGE} + restart: unless-stopped + env_file: .env + environment: + MCP_TRANSPORT: http + MCP_HOST: 0.0.0.0 # allowed because authentication is configured + DATA_DIR: /var/lib/aegis-mcp + ports: + - "127.0.0.1:{port}:8080" + volumes: + - aegis-mcp-data:/var/lib/aegis-mcp + read_only: true + tmpfs: + - /tmp + cap_drop: + - ALL + security_opt: + - no-new-privileges:true + +volumes: + aegis-mcp-data: +""" + + +def caddy_snippet(public_base_url: str, port: int = 8080) -> str: + """Return a Caddyfile site block that terminates TLS and proxies to the container.""" + host = urlsplit(public_base_url).netloc or "gitea-mcp.example.com" + return f"""\ +{host} {{ + reverse_proxy 127.0.0.1:{port} {{ + flush_interval -1 + }} +}} +""" + + +NGINX_NOTE = ( + "nginx: inside the proxying location block set `proxy_buffering off;` so " + "streamed responses are not held back." +) + + +def read_secret_command(env_file: str, key: str) -> str: + """Return a shell command that prints one value of the env file.""" + return f"grep {shlex.quote(f'^{key}=')} {shlex.quote(env_file)} | cut -d= -f2-" diff --git a/src/aegis_gitea_mcp/setup_template.py b/src/aegis_gitea_mcp/setup_template.py new file mode 100644 index 0000000..20d19ad --- /dev/null +++ b/src/aegis_gitea_mcp/setup_template.py @@ -0,0 +1,71 @@ +"""Embedded copy of ``.env.example`` used when the template file is not on disk. + +``aegis-gitea-mcp setup`` prefers ``./.env.example`` but must also work from an +installed wheel. ``tests/test_setup_template.py`` keeps this copy identical to the +repository file. +""" + +from __future__ import annotations + +ENV_TEMPLATE = """\ +# AegisGitea MCP configuration. Copy to .env and fill in. All names are +# case-insensitive. Never commit a real .env. + +# --- Gitea ------------------------------------------------------------------- +# Base URL of your Gitea instance. A trailing /api/v1 is accepted and removed. +GITEA_URL=https://gitea.example.com +# Token used for stdio and for admin-token mode on HTTP. Create it in Gitea under +# Settings > Applications. Its scopes bound what this server can do. +GITEA_TOKEN= + +# --- Transport --------------------------------------------------------------- +# stdio (default) or http +MCP_TRANSPORT=stdio +MCP_HOST=127.0.0.1 +MCP_PORT=8080 +MCP_PATH=/mcp +# Static bearer token for the HTTP endpoint (admin identity). At least 32 +# characters. Generate with: openssl rand -hex 32 +MCP_BEARER_TOKEN= +# Public base URL of this server behind a reverse proxy (https unless loopback). +PUBLIC_BASE_URL= +# Extra Host header values accepted behind a reverse proxy (comma separated). +ALLOWED_HOSTS= + +# --- Per-user OAuth (HTTP only) ----------------------------------------------- +OAUTH_ENABLED=false +GITEA_OAUTH_CLIENT_ID= +GITEA_OAUTH_CLIENT_SECRET= +# Encrypts stored user tokens. At least 32 characters. Generate with: +# openssl rand -hex 32 +TOKEN_ENCRYPTION_KEY= +# Granular Gitea scopes only (read:/write: activitypub, admin, issue, misc, +# notification, organization, package, repository, user). Unknown values are rejected. +OAUTH_SCOPES=write:activitypub,write:admin,write:issue,write:misc,write:notification,write:organization,write:package,write:repository,write:user + +# --- State -------------------------------------------------------------------- +# Token store and audit log live here. +DATA_DIR=~/.local/state/aegis-gitea-mcp +# Defaults to DATA_DIR/audit.log +AUDIT_LOG_PATH= + +# --- Tool exposure (everything is enabled by default) --------------------------- +# Expose only GET operations. +READ_ONLY=false +# Comma separated domains: admin, repo, issue, pull, release, wiki, org, team, +# user, actions, hook, package, notification, misc, or all. +TOOLSETS=all +# Comma separated tool names to hide, e.g. gitea_repo_delete +DISABLED_TOOLS= + +# --- Limits and safety ---------------------------------------------------------- +REQUEST_TIMEOUT_SECONDS=30 +MAX_TOOL_RESPONSE_ITEMS=200 +MAX_TOOL_RESPONSE_CHARS=60000 +MAX_BINARY_BYTES=5242880 +# off, mask or block +SECRET_DETECTION_MODE=mask +LOG_LEVEL=INFO +EXPOSE_ERROR_DETAILS=false +RATE_LIMIT_PER_MINUTE=120 +""" diff --git a/tests/test_cli.py b/tests/test_cli.py index 894a025..dfb42ed 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -15,7 +15,7 @@ def test_version_prints_package_version(capsys: pytest.CaptureFixture[str]) -> N assert capsys.readouterr().out.strip() == __version__ -@pytest.mark.parametrize("argv", [[], ["serve"], ["serve", "--transport", "http"], ["setup"]]) +@pytest.mark.parametrize("argv", [[], ["serve"], ["serve", "--transport", "http"]]) def test_unbuilt_commands_exit_2(argv: list[str], capsys: pytest.CaptureFixture[str]) -> None: assert main(argv) == 2 captured = capsys.readouterr() diff --git a/tests/test_setup.py b/tests/test_setup.py new file mode 100644 index 0000000..9104f0e --- /dev/null +++ b/tests/test_setup.py @@ -0,0 +1,713 @@ +"""Tests for ``aegis-gitea-mcp setup`` (HTTP mocked with pytest-httpx, files in tmp_path).""" + +from __future__ import annotations + +import json +import os +import re +import stat +from pathlib import Path +from typing import Any + +import httpx +import pytest +from pytest_httpx import HTTPXMock + +from aegis_gitea_mcp import setup_cmd +from aegis_gitea_mcp.cli import main +from aegis_gitea_mcp.config import Settings +from aegis_gitea_mcp.setup_cmd import ( + EnvEntry, + missing_scopes, + parse_env_text, + render_env, + write_env_atomic, +) +from aegis_gitea_mcp.setup_template import ENV_TEMPLATE + +GITEA = "https://git.test" +API = f"{GITEA}/api/v1" +PUBLIC = "https://mcp.example.com" +TOKEN = "tok_SECRET_0123456789abcdef" +CLIENT_SECRET = "gto_CLIENTSECRET_9876543210" +OLD_SECRET = "gto_OLDSECRET_1111111111" +ALL_WRITE = [f"write:{c}" for c in setup_cmd.GITEA_SCOPE_CATEGORIES] +LIST_URL = re.compile(re.escape(f"{API}/user/applications/oauth2") + r"\?page=\d+&limit=\d+") + + +def app_json( + app_id: int = 7, + client_id: str = "cid-7", + redirects: list[str] | None = None, + secret: str = "", + confidential: bool = True, + name: str = "aegis-gitea-mcp", +) -> dict[str, Any]: + return { + "id": app_id, + "name": name, + "client_id": client_id, + "client_secret": secret, + "redirect_uris": redirects if redirects is not None else [f"{PUBLIC}/oauth/callback"], + "confidential_client": confidential, + "skip_secondary_authorization": False, + } + + +def mock_instance( + httpx_mock: HTTPXMock, + *, + scopes: list[str] | None = None, + apps: list[dict[str, Any]] | None = None, + token_status: int = 200, +) -> None: + """Register the version, user (admin), token and app-list endpoints.""" + httpx_mock.add_response(url=f"{API}/version", json={"version": "1.99.0"}) + httpx_mock.add_response(url=f"{API}/user", json={"login": "root", "is_admin": True}) + if token_status == 200: + httpx_mock.add_response( + url=f"{API}/token", + json={"id": 1, "name": "setup", "scopes": ["all"] if scopes is None else scopes}, + ) + else: + httpx_mock.add_response(url=f"{API}/token", status_code=token_status, json={}) + if apps is not None: + httpx_mock.add_response(url=LIST_URL, json=apps, headers={"x-total-count": str(len(apps))}) + + +def mock_create(httpx_mock: HTTPXMock, **overrides: Any) -> None: + httpx_mock.add_response( + method="POST", + url=f"{API}/user/applications/oauth2", + status_code=201, + json=app_json(secret=CLIENT_SECRET, **overrides), + ) + + +def run_setup(*extra: str, public: bool = True, oauth: bool = True) -> int: + argv = ["setup", "--gitea-url", GITEA, "--yes", *extra] + if public: + argv += ["--public-base-url", PUBLIC] + if not oauth: + argv.append("--no-oauth") + return main(argv) + + +def read_env(path: Path) -> dict[str, str]: + return {k: e.value for k, e in parse_env_text(path.read_text()).items()} + + +def write_existing_env(path: Path, **values: str) -> None: + path.write_text("".join(f"{k}={v}\n" for k, v in values.items())) + path.chmod(0o600) + + +@pytest.fixture(autouse=True) +def token_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITEA_TOKEN", TOKEN) + + +def assert_no_secrets(out: str, *extra: str) -> None: + for secret in (TOKEN, CLIENT_SECRET, OLD_SECRET, *extra): + assert secret not in out + + +# --- happy path -------------------------------------------------------------- + + +def test_happy_path_creates_app_and_env( + httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + mock_instance(httpx_mock, apps=[]) + mock_create(httpx_mock) + assert run_setup() == 0 + + post = next(r for r in httpx_mock.get_requests() if r.method == "POST") + assert json.loads(post.content) == { + "name": "aegis-gitea-mcp", + "redirect_uris": [f"{PUBLIC}/oauth/callback"], + "confidential_client": True, + "skip_secondary_authorization": False, + } + assert all(r.headers["authorization"] == f"token {TOKEN}" for r in httpx_mock.get_requests()) + + env_path = tmp_path / ".env" + assert stat.S_IMODE(env_path.stat().st_mode) == 0o600 + env = read_env(env_path) + assert env["GITEA_URL"] == GITEA + assert env["GITEA_TOKEN"] == TOKEN + assert env["GITEA_OAUTH_CLIENT_ID"] == "cid-7" + assert env["GITEA_OAUTH_CLIENT_SECRET"] == CLIENT_SECRET + assert env["OAUTH_ENABLED"] == "true" + assert env["PUBLIC_BASE_URL"] == PUBLIC + assert len(env["MCP_BEARER_TOKEN"]) >= 32 + assert len(env["TOKEN_ENCRYPTION_KEY"]) >= 32 + assert env["MCP_BEARER_TOKEN"] != env["TOKEN_ENCRYPTION_KEY"] + text = env_path.read_text() + assert text.startswith("# AegisGitea MCP configuration.") + assert "# --- Gitea" in text + assert text.index("GITEA_URL=") < text.index("MCP_TRANSPORT=") < text.index("OAUTH_ENABLED=") + + captured = capsys.readouterr() + for stream in (captured.out, captured.err): + assert_no_secrets(stream, env["MCP_BEARER_TOKEN"], env["TOKEN_ENCRYPTION_KEY"]) + assert "Gitea reachable: version 1.99.0" in captured.out + assert "root (site administrator)" in captured.out + assert "cid-7" in captured.out + assert "grep '^MCP_BEARER_TOKEN=' " in captured.out + + +def test_written_env_is_accepted_by_settings(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + mock_instance(httpx_mock, apps=[]) + mock_create(httpx_mock) + assert run_setup() == 0 + settings = Settings(_env_file=str(tmp_path / ".env")) # type: ignore[call-arg] + assert settings.oauth_enabled + assert settings.oauth_config_errors() == [] + assert settings.http_auth_configured() + assert settings.gitea_url == GITEA + + +def test_printed_configs(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None: + mock_instance(httpx_mock, apps=[]) + mock_create(httpx_mock) + assert run_setup() == 0 + out = capsys.readouterr().out + assert f"--env GITEA_URL={GITEA} --env GITEA_TOKEN=" in out + assert "uvx aegis-gitea-mcp" in out + assert '"mcpServers"' in out + assert ( + f"claude mcp add --transport http aegis-gitea {PUBLIC}/mcp " + '--header "Authorization: Bearer "' + ) in out + assert "claude.ai custom connector" in out + assert "Mistral Le Chat" in out + assert "git.hiddenden.cafe/hiddenden/aegisgitea-mcp:dev" in out + assert "MCP_HOST: 0.0.0.0" in out + assert '"127.0.0.1:8080:8080"' in out + assert "no-new-privileges:true" in out + assert "mcp.example.com {" in out + assert "reverse_proxy 127.0.0.1:8080" in out + assert "proxy_buffering off;" in out + assert f"curl {PUBLIC}/health" in out + assert "aegis-gitea-mcp serve" in out + assert "MCP_TRANSPORT=http aegis-gitea-mcp serve" in out + + +# --- verification failures ----------------------------------------------------- + + +def test_non_admin_token_fails( + httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + httpx_mock.add_response(url=f"{API}/version", json={"version": "1.99.0"}) + httpx_mock.add_response(url=f"{API}/user", json={"login": "bob", "is_admin": False}) + assert run_setup() == 1 + err = capsys.readouterr().err + assert "not a site administrator" in err + assert "Settings > Applications > Generate New Token" in err + assert not (tmp_path / ".env").exists() + assert_no_secrets(err) + + +def test_rejected_token_fails(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None: + httpx_mock.add_response(url=f"{API}/version", json={"version": "1.99.0"}) + httpx_mock.add_response(url=f"{API}/user", status_code=401, json={"message": "bad token"}) + assert run_setup() == 1 + assert "token rejected" in capsys.readouterr().err + + +def test_missing_scopes_listed( + httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + mock_instance(httpx_mock, scopes=["write:repository", "read:user", "write:issue"]) + assert run_setup() == 1 + err = capsys.readouterr().err + listed = err.split("lacks required scopes: ")[1].split(". ")[0].split(", ") + assert "write:admin" in listed + assert "write:user" in listed + assert "write:repository" not in listed + assert "write:issue" not in listed + assert not (tmp_path / ".env").exists() + + +@pytest.mark.parametrize("scopes", [["all"], ["ALL", "read:user"], ALL_WRITE]) +def test_full_scopes_accepted(httpx_mock: HTTPXMock, scopes: list[str]) -> None: + mock_instance(httpx_mock, scopes=scopes) + assert run_setup(oauth=False) == 0 + + +def test_missing_scopes_helper() -> None: + assert missing_scopes(["all"]) == [] + assert missing_scopes(ALL_WRITE) == [] + assert missing_scopes(["read:user"]) == list(setup_cmd.REQUIRED_SCOPES) + + +def test_token_endpoint_missing_warns_and_continues( + httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str] +) -> None: + mock_instance(httpx_mock, token_status=404) + assert run_setup(oauth=False) == 0 + out = capsys.readouterr().out + assert "Warning: could not read the token scopes" in out + assert "not verified" in out + + +def test_empty_scope_list_warns(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None: + mock_instance(httpx_mock, scopes=[]) + assert run_setup(oauth=False) == 0 + assert "could not be verified" in capsys.readouterr().out + + +def test_public_only_scope_warns(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None: + mock_instance(httpx_mock, scopes=["all", "public-only"]) + assert run_setup(oauth=False) == 0 + assert "public-only" in capsys.readouterr().out + + +def test_unreachable_instance( + httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + httpx_mock.add_exception(httpx.ConnectError("boom")) + assert run_setup() == 1 + err = capsys.readouterr().err + assert "Gitea is not reachable" in err + assert "git.test" in err + assert not (tmp_path / ".env").exists() + + +def test_timeout_message(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None: + httpx_mock.add_exception(httpx.ReadTimeout("slow")) + assert run_setup() == 1 + assert "did not answer within 15 seconds" in capsys.readouterr().err + + +def test_not_gitea_response(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None: + httpx_mock.add_response(url=f"{API}/version", text="login") + assert run_setup() == 1 + assert "did not return Gitea API JSON" in capsys.readouterr().err + + +def test_redirect_is_not_followed( + httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str] +) -> None: + httpx_mock.add_response( + url=f"{API}/version", status_code=301, headers={"location": "https://other.test/x"} + ) + assert run_setup() == 1 + err = capsys.readouterr().err + assert "redirect to https://other.test/x" in err + assert len(httpx_mock.get_requests()) == 1 + + +def test_gitea_error_text_is_redacted( + httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str] +) -> None: + httpx_mock.add_response(url=f"{API}/version", json={"version": "1.99.0"}) + httpx_mock.add_response( + url=f"{API}/user", status_code=500, json={"message": f"boom {TOKEN} boom"} + ) + assert run_setup() == 1 + err = capsys.readouterr().err + assert "boom *** boom" in err + assert_no_secrets(err) + + +# --- OAuth app ------------------------------------------------------------------- + + +def test_oauth_app_reused_keeps_secret( + httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + write_existing_env( + tmp_path / ".env", + GITEA_OAUTH_CLIENT_ID="cid-7", + GITEA_OAUTH_CLIENT_SECRET=OLD_SECRET, + MCP_BEARER_TOKEN="b" * 40, + TOKEN_ENCRYPTION_KEY="k" * 40, + ) + mock_instance(httpx_mock, apps=[app_json(), app_json(app_id=8, client_id="x", name="other")]) + assert run_setup() == 0 + assert all(r.method == "GET" for r in httpx_mock.get_requests()) + env = read_env(tmp_path / ".env") + assert env["GITEA_OAUTH_CLIENT_SECRET"] == OLD_SECRET + assert env["GITEA_OAUTH_CLIENT_ID"] == "cid-7" + assert env["OAUTH_ENABLED"] == "true" + out = capsys.readouterr().out + assert "reused" in out + assert_no_secrets(out) + + +def test_oauth_app_redirect_differs_is_patched( + httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + write_existing_env( + tmp_path / ".env", GITEA_OAUTH_CLIENT_ID="cid-7", GITEA_OAUTH_CLIENT_SECRET=OLD_SECRET + ) + mock_instance(httpx_mock, apps=[app_json(redirects=["https://old.example.com/oauth/callback"])]) + httpx_mock.add_response( + method="PATCH", + url=f"{API}/user/applications/oauth2/7", + json=app_json(secret=CLIENT_SECRET), + ) + assert run_setup() == 0 + patch = next(r for r in httpx_mock.get_requests() if r.method == "PATCH") + assert json.loads(patch.content)["redirect_uris"] == [f"{PUBLIC}/oauth/callback"] + env = read_env(tmp_path / ".env") + assert env["GITEA_OAUTH_CLIENT_SECRET"] == CLIENT_SECRET + out = capsys.readouterr().out + assert "secret regenerated" in out + assert "removed 1 other redirect URI" in out + assert_no_secrets(out) + + +def test_oauth_app_unknown_secret_is_rotated( + httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + mock_instance(httpx_mock, apps=[app_json()]) + httpx_mock.add_response( + method="PATCH", + url=f"{API}/user/applications/oauth2/7", + json=app_json(secret=CLIENT_SECRET), + ) + assert run_setup() == 0 + env = read_env(tmp_path / ".env") + assert env["GITEA_OAUTH_CLIENT_SECRET"] == CLIENT_SECRET + assert env["GITEA_OAUTH_CLIENT_ID"] == "cid-7" + assert "secret rotated" in capsys.readouterr().out + + +def test_rotation_requires_yes_when_not_interactive( + httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + mock_instance(httpx_mock, apps=[app_json()]) + assert main(["setup", "--gitea-url", GITEA, "--public-base-url", PUBLIC]) == 1 + assert "Re-run with --yes" in capsys.readouterr().err + assert not (tmp_path / ".env").exists() + + +def test_stale_client_id_creates_new_app_when_missing( + httpx_mock: HTTPXMock, tmp_path: Path +) -> None: + write_existing_env( + tmp_path / ".env", GITEA_OAUTH_CLIENT_ID="gone", GITEA_OAUTH_CLIENT_SECRET=OLD_SECRET + ) + mock_instance(httpx_mock, apps=[]) + mock_create(httpx_mock, client_id="fresh") + assert run_setup() == 0 + env = read_env(tmp_path / ".env") + assert env["GITEA_OAUTH_CLIENT_ID"] == "fresh" + assert env["GITEA_OAUTH_CLIENT_SECRET"] == CLIENT_SECRET + + +def test_oauth_apps_are_paginated(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + mock_instance(httpx_mock) + httpx_mock.add_response( + url=f"{API}/user/applications/oauth2?page=1&limit=50", + json=[app_json(app_id=1, client_id="a", name="x")], + ) + httpx_mock.add_response( + url=f"{API}/user/applications/oauth2?page=2&limit=50", + json=[app_json(app_id=2, client_id="b", name="y")], + ) + httpx_mock.add_response(url=f"{API}/user/applications/oauth2?page=3&limit=50", json=[]) + mock_create(httpx_mock, app_id=3, client_id="c") + assert run_setup() == 0 + assert read_env(tmp_path / ".env")["GITEA_OAUTH_CLIENT_ID"] == "c" + + +def test_oauth_create_forbidden_explains_scope( + httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str] +) -> None: + mock_instance(httpx_mock, apps=[]) + httpx_mock.add_response( + method="POST", url=f"{API}/user/applications/oauth2", status_code=403, json={} + ) + assert run_setup() == 1 + assert "write:user" in capsys.readouterr().err + + +def test_no_oauth_flag(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + mock_instance(httpx_mock) + assert run_setup(oauth=False) == 0 + assert not any("applications" in str(r.url) for r in httpx_mock.get_requests()) + env = read_env(tmp_path / ".env") + assert env["OAUTH_ENABLED"] == "false" + assert env["PUBLIC_BASE_URL"] == PUBLIC + assert env["GITEA_OAUTH_CLIENT_ID"] == "" + + +def test_no_public_url_skips_oauth( + httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + mock_instance(httpx_mock) + assert run_setup(public=False) == 0 + assert read_env(tmp_path / ".env")["OAUTH_ENABLED"] == "false" + out = capsys.readouterr().out + assert "no public base URL given" in out + assert "http://127.0.0.1:8080/mcp" in out + assert "Caddyfile" not in out + + +# --- env file handling ---------------------------------------------------------------- + + +def test_existing_values_preserved(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + bearer, key = "b" * 40, "k" * 40 + (tmp_path / ".env").write_text( + "MAX_TOOL_RESPONSE_ITEMS=5\n" + f"MCP_BEARER_TOKEN={bearer}\n" + f"TOKEN_ENCRYPTION_KEY={key}\n" + 'MY_CUSTOM_VAR="hello world"\n' + "READ_ONLY=true # keep it safe\n" + ) + mock_instance(httpx_mock) + assert run_setup(oauth=False) == 0 + text = (tmp_path / ".env").read_text() + env = read_env(tmp_path / ".env") + assert env["MAX_TOOL_RESPONSE_ITEMS"] == "5" + assert env["MCP_BEARER_TOKEN"] == bearer + assert env["TOKEN_ENCRYPTION_KEY"] == key + assert env["MY_CUSTOM_VAR"] == "hello world" + assert 'MY_CUSTOM_VAR="hello world"' in text + assert "READ_ONLY=true # keep it safe" in text + assert text.count("MAX_TOOL_RESPONSE_ITEMS=") == 1 + + +def test_short_existing_secret_warns( + httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + write_existing_env(tmp_path / ".env", MCP_BEARER_TOKEN="short") + mock_instance(httpx_mock) + assert run_setup(oauth=False) == 0 + assert read_env(tmp_path / ".env")["MCP_BEARER_TOKEN"] == "short" + assert "MCP_BEARER_TOKEN in the env file is shorter" in capsys.readouterr().out + + +def test_url_and_token_taken_from_env_file( + httpx_mock: HTTPXMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv("GITEA_TOKEN") + write_existing_env(tmp_path / ".env", GITEA_URL=GITEA, GITEA_TOKEN=TOKEN) + mock_instance(httpx_mock) + assert main(["setup", "--yes", "--no-oauth"]) == 0 + assert read_env(tmp_path / ".env")["GITEA_TOKEN"] == TOKEN + + +def test_symlink_env_file_refused( + httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + target = tmp_path / "real.env" + target.write_text("KEEP=1\n") + (tmp_path / ".env").symlink_to(target) + assert run_setup() == 1 + assert "symlink" in capsys.readouterr().err + assert target.read_text() == "KEEP=1\n" + assert httpx_mock.get_requests() == [] + + +def test_loose_permissions_are_tightened( + httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + env_path = tmp_path / ".env" + env_path.write_text("READ_ONLY=true\n") + env_path.chmod(0o644) + mock_instance(httpx_mock) + assert run_setup(oauth=False) == 0 + assert stat.S_IMODE(env_path.stat().st_mode) == 0o600 + assert "permissions are now 0600" in capsys.readouterr().out + + +def test_custom_env_file_and_template(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + template = tmp_path / "tpl" + template.write_text("# my template\nGITEA_URL=\nOAUTH_ENABLED=false\nOTHER=default\n") + mock_instance(httpx_mock) + target = tmp_path / "out.env" + assert run_setup("--env-file", str(target), "--template", str(template), oauth=False) == 0 + text = target.read_text() + assert text.startswith("# my template\n") + assert "OTHER=default" in text + assert f"GITEA_URL={GITEA}" in text + + +def test_missing_explicit_template_fails(capsys: pytest.CaptureFixture[str]) -> None: + assert run_setup("--template", "/nonexistent/tpl") == 1 + assert "cannot read template" in capsys.readouterr().err + + +def test_falls_back_to_embedded_template(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + assert not (tmp_path / ".env.example").exists() + mock_instance(httpx_mock) + assert run_setup(oauth=False) == 0 + assert (tmp_path / ".env").read_text().startswith(ENV_TEMPLATE.splitlines()[0]) + + +def test_uses_local_env_example_when_present(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + (tmp_path / ".env.example").write_text("# local template\nGITEA_URL=\n") + mock_instance(httpx_mock) + assert run_setup(oauth=False) == 0 + assert (tmp_path / ".env").read_text().startswith("# local template\n") + + +def test_missing_env_dir_fails(capsys: pytest.CaptureFixture[str]) -> None: + assert run_setup("--env-file", "/nonexistent-dir-xyz/.env") == 1 + assert "does not exist" in capsys.readouterr().err + + +# --- inputs ---------------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("url", "needle"), + [ + ("ftp://git.test", "http:// or https://"), + ("https://user:hunter2@git.test", "credentials"), + ("https://git.test?x=1", "query string"), + ("git.test", "http:// or https://"), + ], +) +def test_invalid_gitea_url(url: str, needle: str, capsys: pytest.CaptureFixture[str]) -> None: + assert main(["setup", "--yes", "--gitea-url", url]) == 1 + err = capsys.readouterr().err + assert "invalid Gitea URL" in err + assert needle in err + assert "hunter2" not in err + + +def test_invalid_public_url(capsys: pytest.CaptureFixture[str]) -> None: + assert main(["setup", "--yes", "--gitea-url", GITEA, "--public-base-url", "http://mcp.x"]) == 1 + err = capsys.readouterr().err + assert "invalid public base URL" in err + assert "https" in err + + +def test_gitea_url_api_suffix_normalised(httpx_mock: HTTPXMock) -> None: + mock_instance(httpx_mock) + assert main(["setup", "--yes", "--no-oauth", "--gitea-url", f"{GITEA}/api/v1/"]) == 0 + + +def test_token_prompt_uses_getpass( + httpx_mock: HTTPXMock, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + monkeypatch.delenv("GITEA_TOKEN") + prompts: list[str] = [] + + def fake_getpass(prompt: str = "") -> str: + prompts.append(prompt) + return f" {TOKEN} " + + monkeypatch.setattr(setup_cmd.getpass, "getpass", fake_getpass) + mock_instance(httpx_mock) + assert main(["setup", "--gitea-url", GITEA, "--no-oauth"]) == 0 + assert len(prompts) == 1 + assert read_env(tmp_path / ".env")["GITEA_TOKEN"] == TOKEN + assert_no_secrets(capsys.readouterr().out) + + +def test_url_prompt(httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setattr("builtins.input", lambda _prompt="": GITEA) + mock_instance(httpx_mock) + assert main(["setup", "--no-oauth"]) == 0 + assert read_env(tmp_path / ".env")["GITEA_URL"] == GITEA + + +def test_missing_token_with_yes_fails( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.delenv("GITEA_TOKEN") + assert main(["setup", "--yes", "--gitea-url", GITEA]) == 1 + assert "no Gitea token" in capsys.readouterr().err + + +def test_missing_url_with_yes_fails(capsys: pytest.CaptureFixture[str]) -> None: + assert main(["setup", "--yes"]) == 1 + assert "no Gitea URL" in capsys.readouterr().err + + +def test_unsafe_token_rejected_without_echo( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setenv("GITEA_TOKEN", "abc def#ghi") + assert main(["setup", "--yes", "--gitea-url", GITEA]) == 1 + captured = capsys.readouterr() + assert "abc" not in captured.err + captured.out + + +def test_placeholder_url_in_env_file_is_ignored( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + write_existing_env(tmp_path / ".env", GITEA_URL="https://gitea.example.com") + assert main(["setup", "--yes"]) == 1 + assert "no Gitea URL" in capsys.readouterr().err + + +# --- env helpers -------------------------------------------------------------------------------- + + +def test_parse_env_text_handles_quotes_comments_and_case() -> None: + text = "# c\nA=1\nb = \"two words\"\nC='x'\nD=val # note\nexport E=5\nnot a line\n" + parsed = parse_env_text(text) + assert {k: e.value for k, e in parsed.items()} == { + "A": "1", + "B": "two words", + "C": "x", + "D": "val", + "E": "5", + } + + +def test_render_env_precedence_and_extras() -> None: + template = "# head\nA=tpl\nB=tpl\n# C=commented\nD=tpl\n" + existing = { + "B": EnvEntry("B", '"keep me"', "keep me"), + "X": EnvEntry("X", "9", "9"), + "A": EnvEntry("A", "old", "old"), + } + out = render_env(template, {"A": "new", "Z": "1"}, existing) + lines = out.splitlines() + assert lines[:5] == ["# head", "A=new", 'B="keep me"', "# C=commented", "D=tpl"] + assert "Z=1" in lines + assert "X=9" in lines + assert out.endswith("\n") + + +def test_write_env_atomic_mode_and_cleanup(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + path = tmp_path / "x.env" + write_env_atomic(path, "A=1\n") + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + assert path.read_text() == "A=1\n" + + def boom(*_args: Any, **_kwargs: Any) -> None: + raise OSError("disk full") + + monkeypatch.setattr(os, "replace", boom) + with pytest.raises(OSError, match="disk full"): + write_env_atomic(path, "A=2\n") + assert path.read_text() == "A=1\n" + assert [p.name for p in tmp_path.iterdir() if p.name != "x.env"] == [] + + +def test_write_env_failure_reports_rerun_hint( + httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + mock_instance(httpx_mock) + + def boom(*_args: Any, **_kwargs: Any) -> None: + raise OSError("nope") + + monkeypatch.setattr(os, "replace", boom) + assert run_setup(oauth=False) == 1 + assert "run setup again" in capsys.readouterr().err + + +def test_http_gitea_url_warns(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None: + httpx_mock.add_response(url="http://git.test/api/v1/version", json={"version": "1"}) + httpx_mock.add_response( + url="http://git.test/api/v1/user", json={"login": "root", "is_admin": True} + ) + httpx_mock.add_response(url="http://git.test/api/v1/token", json={"scopes": ["all"]}) + assert main(["setup", "--yes", "--no-oauth", "--gitea-url", "http://git.test"]) == 0 + assert "plain http" in capsys.readouterr().out diff --git a/tests/test_setup_template.py b/tests/test_setup_template.py new file mode 100644 index 0000000..1eed1e1 --- /dev/null +++ b/tests/test_setup_template.py @@ -0,0 +1,12 @@ +"""The embedded env template must stay identical to ``.env.example``.""" + +from __future__ import annotations + +from pathlib import Path + +from aegis_gitea_mcp.setup_template import ENV_TEMPLATE + + +def test_embedded_template_matches_env_example() -> None: + example = Path(__file__).parent.parent / ".env.example" + assert example.read_text(encoding="utf-8") == ENV_TEMPLATE