feat!: full-access Gitea MCP server (1.0.0): full API coverage, stdio + streamable HTTP, admin and per-user OAuth #71
@@ -0,0 +1,125 @@
|
|||||||
|
"""The ``aegis-gitea-mcp setup`` command against the live instance."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
import stat
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from aegis_gitea_mcp.config import Settings
|
||||||
|
from aegis_gitea_mcp.setup_cmd import OAUTH_APP_NAME, parse_env_text
|
||||||
|
from tests.integration.conftest import RegularUser
|
||||||
|
from tests.integration.harness import GiteaInstance
|
||||||
|
|
||||||
|
PUBLIC_URL = "https://mcp.example.test"
|
||||||
|
REDIRECT_URI = f"{PUBLIC_URL}/oauth/callback"
|
||||||
|
SECRET_KEYS = (
|
||||||
|
"GITEA_TOKEN",
|
||||||
|
"MCP_BEARER_TOKEN",
|
||||||
|
"TOKEN_ENCRYPTION_KEY",
|
||||||
|
"GITEA_OAUTH_CLIENT_SECRET",
|
||||||
|
)
|
||||||
|
_PASSTHROUGH_ENV = ("PATH", "HOME", "LANG", "LC_ALL", "TMPDIR")
|
||||||
|
|
||||||
|
|
||||||
|
def run_setup(
|
||||||
|
gitea: GiteaInstance, token: str, env_file: Path, cwd: Path
|
||||||
|
) -> subprocess.CompletedProcess[str]:
|
||||||
|
"""Run the setup command in a subprocess; the Gitea token goes in the environment only."""
|
||||||
|
env = {name: os.environ[name] for name in _PASSTHROUGH_ENV if name in os.environ}
|
||||||
|
env["GITEA_TOKEN"] = token
|
||||||
|
return subprocess.run( # noqa: S603
|
||||||
|
[
|
||||||
|
sys.executable,
|
||||||
|
"-m",
|
||||||
|
"aegis_gitea_mcp.cli",
|
||||||
|
"setup",
|
||||||
|
"--gitea-url",
|
||||||
|
gitea.base_url,
|
||||||
|
"--public-base-url",
|
||||||
|
PUBLIC_URL,
|
||||||
|
"--env-file",
|
||||||
|
str(env_file),
|
||||||
|
"--yes",
|
||||||
|
],
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
timeout=120,
|
||||||
|
check=False,
|
||||||
|
env=env,
|
||||||
|
cwd=cwd,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def oauth_apps(gitea: GiteaInstance) -> list[dict[str, object]]:
|
||||||
|
"""The admin's OAuth2 applications named like the one setup registers."""
|
||||||
|
response = gitea.api("GET", "/user/applications/oauth2", params={"limit": 50})
|
||||||
|
assert response.status_code == 200
|
||||||
|
return [app for app in response.json() if app["name"] == OAUTH_APP_NAME]
|
||||||
|
|
||||||
|
|
||||||
|
def read_env(path: Path) -> dict[str, str]:
|
||||||
|
"""Key/value pairs of an env file."""
|
||||||
|
return {key: entry.value for key, entry in parse_env_text(path.read_text("utf-8")).items()}
|
||||||
|
|
||||||
|
|
||||||
|
def test_setup_registers_the_oauth_app_and_writes_a_private_env_file(
|
||||||
|
gitea: GiteaInstance, tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
env_file = tmp_path / ".env"
|
||||||
|
assert oauth_apps(gitea) == []
|
||||||
|
|
||||||
|
first = run_setup(gitea, gitea.admin_token, env_file, tmp_path)
|
||||||
|
assert first.returncode == 0, first.stderr
|
||||||
|
assert "setup failed" not in first.stderr
|
||||||
|
|
||||||
|
# The OAuth2 application exists with the callback of the public URL.
|
||||||
|
apps = oauth_apps(gitea)
|
||||||
|
assert len(apps) == 1
|
||||||
|
assert apps[0]["redirect_uris"] == [REDIRECT_URI]
|
||||||
|
assert apps[0]["confidential_client"] is True
|
||||||
|
|
||||||
|
# The env file is private, complete and loads without errors.
|
||||||
|
assert stat.S_IMODE(env_file.stat().st_mode) == 0o600
|
||||||
|
values = read_env(env_file)
|
||||||
|
assert values["GITEA_URL"] == gitea.base_url
|
||||||
|
assert values["PUBLIC_BASE_URL"] == PUBLIC_URL
|
||||||
|
assert values["OAUTH_ENABLED"] == "true"
|
||||||
|
assert values["GITEA_OAUTH_CLIENT_ID"] == apps[0]["client_id"]
|
||||||
|
settings = Settings(_env_file=str(env_file)) # type: ignore[call-arg]
|
||||||
|
assert settings.oauth_enabled
|
||||||
|
assert settings.oauth_config_errors() == []
|
||||||
|
assert settings.gitea_token is not None
|
||||||
|
assert settings.gitea_token.get_secret_value() == gitea.admin_token
|
||||||
|
assert settings.mcp_bearer_token is not None
|
||||||
|
|
||||||
|
# No secret value reaches stdout or stderr.
|
||||||
|
for key in SECRET_KEYS:
|
||||||
|
assert values[key], key
|
||||||
|
assert values[key] not in first.stdout, f"{key} leaked to stdout"
|
||||||
|
assert values[key] not in first.stderr, f"{key} leaked to stderr"
|
||||||
|
assert gitea.admin_token not in first.stdout + first.stderr
|
||||||
|
|
||||||
|
# Running it again is idempotent: same app, same secrets, no duplicate.
|
||||||
|
second = run_setup(gitea, gitea.admin_token, env_file, tmp_path)
|
||||||
|
assert second.returncode == 0, second.stderr
|
||||||
|
assert len(oauth_apps(gitea)) == 1
|
||||||
|
assert read_env(env_file) == values
|
||||||
|
assert stat.S_IMODE(env_file.stat().st_mode) == 0o600
|
||||||
|
for key in SECRET_KEYS:
|
||||||
|
assert values[key] not in second.stdout + second.stderr
|
||||||
|
|
||||||
|
|
||||||
|
def test_setup_with_a_non_admin_token_fails_with_an_actionable_message(
|
||||||
|
gitea: GiteaInstance, regular_user: RegularUser, tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
env_file = tmp_path / "non-admin.env"
|
||||||
|
result = run_setup(gitea, regular_user.token, env_file, tmp_path)
|
||||||
|
|
||||||
|
assert result.returncode == 1
|
||||||
|
assert "not a site administrator" in result.stderr
|
||||||
|
assert "Generate New Token" in result.stderr
|
||||||
|
assert not env_file.exists()
|
||||||
|
assert regular_user.token not in result.stdout + result.stderr
|
||||||
Reference in new issue
Block a user