Dev #72

Merged
Latte merged 71 commits from dev into main 2026-10-14 17:52:23 +00:00
7 changed files with 1798 additions and 3 deletions
Showing only changes of commit 1d223e6799 - Show all commits

No files matched your search

+5 -2
View File
@@ -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)
+853
View File
@@ -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
+143
View File
@@ -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 ``<your-gitea-token>``, 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 = "<your-gitea-token>" # noqa: S105
BEARER_PLACEHOLDER = "<MCP_BEARER_TOKEN>"
#: 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-"
+71
View File
@@ -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
"""
+1 -1
View File
@@ -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()
+713
View File
@@ -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=<your-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 <MCP_BEARER_TOKEN>"'
) 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="<html>login</html>")
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
+12
View File
@@ -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