diff --git a/tests/integration/test_failure_paths.py b/tests/integration/test_failure_paths.py new file mode 100644 index 0000000..11ce73a --- /dev/null +++ b/tests/integration/test_failure_paths.py @@ -0,0 +1,79 @@ +"""Write tools that Gitea refuses: the status and error kind reach the client intact.""" + +from __future__ import annotations + +import secrets + +from tests.integration.conftest import McpSession, RepoRef, b64 +from tests.integration.harness import GiteaInstance + + +async def test_creating_in_a_missing_repository_is_not_found( + mcp: McpSession, gitea: GiteaInstance +) -> None: + missing = {"owner": gitea.admin_login, "repo": "missing-" + secrets.token_hex(4)} + issue = await mcp.fail("gitea_issue_create_issue", 404, body={"title": "x"}, **missing) + assert issue.error_kind == "not_found" + label = await mcp.fail( + "gitea_issue_create_label", 404, body={"name": "x", "color": "#ffffff"}, **missing + ) + assert label.error_kind == "not_found" + # The audit log records the failure with the same kind. + failed = [e for e in mcp.audit_entries() if e["tool_name"] == "gitea_issue_create_issue"] + assert failed and failed[-1]["outcome"] == "not_found" + assert failed[-1]["result_status"] == "error" + + +async def test_invalid_values_are_validation_errors(mcp: McpSession, fresh_repo: RepoRef) -> None: + repo = fresh_repo.args + bad_color = await mcp.call( + "gitea_issue_create_label", body={"name": "bad", "color": "not-a-color"}, **repo + ) + assert bad_color.is_error + assert bad_color.error_kind in ("validation_failed", "bad_request") + assert bad_color.status in (400, 422) + + # Gitea itself refuses a missing body. + missing_body = await mcp.call("gitea_issue_create_issue", **repo) + assert missing_body.is_error and missing_body.status == 422 + assert missing_body.error_kind == "validation_failed" + + # A field of the wrong type never reaches Gitea: the schema rejects it first. + wrong_type = await mcp.call( + "gitea_issue_create_issue", body={"title": ["not", "a", "string"]}, **repo + ) + assert wrong_type.is_error and wrong_type.status is None + assert wrong_type.error_kind == "invalid_argument" + + +async def test_conflicting_writes_are_conflicts( + mcp: McpSession, gitea: GiteaInstance, fresh_repo: RepoRef +) -> None: + repo = fresh_repo.args + duplicate = await mcp.fail( + "gitea_user_current_create_repo", 409, body={"name": fresh_repo.name} + ) + assert duplicate.error_kind == "conflict" + + await mcp.ok( + "gitea_repo_create_file", + status=201, + filepath="a.txt", + body={"content": b64("one"), "message": "one"}, + **repo, + ) + again = await mcp.call( + "gitea_repo_create_file", + filepath="a.txt", + body={"content": b64("two"), "message": "two"}, + **repo, + ) + assert again.is_error and again.status in (409, 422) + + stale = await mcp.call( + "gitea_repo_update_file", + filepath="a.txt", + body={"content": b64("three"), "message": "three", "sha": "0" * 40}, + **repo, + ) + assert stale.is_error and stale.status in (409, 422) diff --git a/tests/integration/test_per_user_oauth.py b/tests/integration/test_per_user_oauth.py index 6e7cb3e..13aadf8 100644 --- a/tests/integration/test_per_user_oauth.py +++ b/tests/integration/test_per_user_oauth.py @@ -47,6 +47,7 @@ class OAuthStack: bearer: str user_login: str user_access_token: str + expired_access_token: str secrets: list[str] @@ -96,6 +97,15 @@ async def build_stack( scopes=[MCP_SCOPE], expires_at=time.time() + ACCESS_TOKEN_TTL, ) + # A token of the same valid session whose lifetime ended a minute ago. + expired = TOKEN_PREFIX + secrets.token_urlsafe(32) + store.tokens[token_hash(expired)] = TokenRecord( + kind="access", + session_id=session_id, + client_id=OAUTH_CLIENT_ID, + scopes=[MCP_SCOPE], + expires_at=time.time() - 60, + ) await store.save() audit = AuditLogger(settings.resolved_audit_log_path) @@ -117,9 +127,11 @@ async def build_stack( bearer=bearer, user_login=user.login, user_access_token=access, + expired_access_token=expired, secrets=[ bearer, access, + expired, user.token, gitea.admin_token, oauth_client_secret, @@ -192,11 +204,32 @@ async def test_calls_run_as_the_signed_in_user( ) ).status in (403, 404) - # Admin tools and sudo are refused by Gitea with 403, not by the server. - admin_only = await mcp.call("gitea_admin_cron_list") - assert admin_only.is_error and admin_only.status == 403 - assert admin_only.error_kind == "forbidden" - assert admin_only.identity["mode"] == "oauth" + # Admin tools and sudo are refused by Gitea with 403, not by the server: every + # admin list endpoint and an admin create, all for this non-admin user. + for tool in ( + "gitea_admin_cron_list", + "gitea_admin_search_users", + "gitea_admin_get_all_orgs", + "gitea_admin_get_all_emails", + "gitea_admin_list_hooks", + "gitea_admin_list_packages", + "gitea_admin_get_runners", + "gitea_admin_unadopted_list", + ): + refused = await mcp.call(tool) + assert refused.is_error and refused.status == 403, tool + assert refused.error_kind == "forbidden", tool + assert refused.identity["mode"] == "oauth" + created = await mcp.call( + "gitea_admin_create_user", + body={ + "username": gitea.unique("it-nope"), + "email": f"{gitea.unique('nope')}@example.test", + "password": secrets.token_urlsafe(24), + }, + ) + assert created.is_error and created.status == 403 + assert created.error_kind == "forbidden" sudo = await mcp.call("gitea_user_get_current", sudo=gitea.admin_login) assert sudo.is_error and sudo.status == 403 @@ -206,6 +239,7 @@ async def test_calls_run_as_the_signed_in_user( assert {e["tool_name"] for e in entries} >= { "gitea_user_get_current", "gitea_admin_cron_list", + "gitea_admin_create_user", } gitea.api("DELETE", f"/repos/{regular_user.login}/{own}") @@ -254,6 +288,7 @@ async def test_unknown_and_expired_tokens_never_reach_gitea( ): for token in ( TOKEN_PREFIX + secrets.token_urlsafe(32), # looks like ours, unknown + stack.expired_access_token, # ours and known, but past its expiry regular_user.token, # the user's own Gitea token is not an MCP token secrets.token_urlsafe(36), ): @@ -264,3 +299,10 @@ async def test_unknown_and_expired_tokens_never_reach_gitea( assert token not in response.text anonymous = await http_client.post("/mcp", json=body, headers=headers) assert anonymous.status_code == 401 + # The expiry is what excluded the token above: the live one of the same session works. + live = await http_client.post( + "/mcp", + json=body, + headers={**headers, "Authorization": f"Bearer {stack.user_access_token}"}, + ) + assert live.status_code == 200 diff --git a/tests/integration/test_scenario_release_hooks.py b/tests/integration/test_scenario_release_hooks.py index 49943c7..eb2c9ff 100644 --- a/tests/integration/test_scenario_release_hooks.py +++ b/tests/integration/test_scenario_release_hooks.py @@ -238,10 +238,15 @@ async def test_webhook_lifecycle( f"{prefix}_edit_hook", f"{prefix}_delete_hook", ) + # The signing secret and the Authorization header are credentials: neither may come + # back in any result or reach the audit log (Gitea itself omits the secret). + secret = f"hook-secret-{secrets.token_hex(12)}" + header = f"Bearer hook-auth-{secrets.token_hex(12)}" + config = {**HOOK_CONFIG, "secret": secret, "authorization_header": header} hook = await mcp.ok( create, status=201, - body={"type": "gitea", "config": HOOK_CONFIG, "events": ["push"], "active": True}, + body={"type": "gitea", "config": config, "events": ["push"], "active": True}, **scope, ) hook_id = hook.data["id"] @@ -252,7 +257,10 @@ async def test_webhook_lifecycle( assert hook_id in [item["id"] for item in hooks.data] assert (await mcp.ok(get, id=hook_id, **scope)).data["config"]["url"] == HOOK_CONFIG["url"] edited = await mcp.ok( - edit, id=hook_id, body={"active": False, "events": ["push", "issues"]}, **scope + edit, + id=hook_id, + body={"active": False, "events": ["push", "issues"], "config": config}, + **scope, ) assert edited.data["active"] is False assert "issues" in edited.data["events"] @@ -260,6 +268,9 @@ async def test_webhook_lifecycle( await mcp.ok("gitea_hook_repo_test_hook", id=hook_id, **scope) await mcp.ok(delete, id=hook_id, **scope) await mcp.fail(get, 404, id=hook_id, **scope) + for credential in (secret, header, header.removeprefix("Bearer ")): + assert all(credential not in text for text in mcp.texts), "a webhook credential leaked" + assert credential not in mcp.audit_text(), "a webhook credential reached the audit log" finally: if level == "org": drop_org(gitea, org)