277 lines
10 KiB
Python
277 lines
10 KiB
Python
"""Scenarios: releases and attachments, Action secrets and variables, webhooks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from tests.integration.conftest import McpSession, RepoRef, b64, drop_org
|
|
from tests.integration.harness import GiteaInstance
|
|
|
|
HOOK_CONFIG = {"url": "http://127.0.0.1:9/hook", "content_type": "json"}
|
|
|
|
|
|
async def test_release_with_attachment(
|
|
mcp: McpSession, gitea: GiteaInstance, fresh_repo: RepoRef
|
|
) -> None:
|
|
repo = fresh_repo.args
|
|
release = await mcp.ok(
|
|
"gitea_release_create_release",
|
|
status=201,
|
|
body={
|
|
"tag_name": "v1.0.0",
|
|
"target_commitish": "main",
|
|
"name": "One point oh",
|
|
"body": "first release",
|
|
},
|
|
**repo,
|
|
)
|
|
release_id = release.data["id"]
|
|
assert release.data["tag_name"] == "v1.0.0"
|
|
assert (await mcp.ok("gitea_release_get_latest_release", **repo)).data["id"] == release_id
|
|
assert (await mcp.ok("gitea_release_get_release_by_tag", tag="v1.0.0", **repo)).data["id"] == (
|
|
release_id
|
|
)
|
|
|
|
# Upload an attachment through the file argument (base64), then list and download it.
|
|
payload = b"release asset " + secrets.token_bytes(24)
|
|
asset = await mcp.ok(
|
|
"gitea_release_create_release_attachment",
|
|
status=201,
|
|
id=release_id,
|
|
name="notes.bin",
|
|
attachment={
|
|
"filename": "notes.bin",
|
|
"content_base64": b64(payload),
|
|
"content_type": "application/octet-stream",
|
|
},
|
|
**repo,
|
|
)
|
|
attachment_id = asset.data["id"]
|
|
assert asset.data["name"] == "notes.bin"
|
|
assert asset.data["size"] == len(payload)
|
|
listed = await mcp.ok("gitea_release_list_release_attachments", id=release_id, **repo)
|
|
assert [item["name"] for item in listed.data] == ["notes.bin"]
|
|
meta = await mcp.ok(
|
|
"gitea_release_get_release_attachment", id=release_id, attachment_id=attachment_id, **repo
|
|
)
|
|
download = gitea.request("GET", meta.data["browser_download_url"].removeprefix(gitea.base_url))
|
|
assert download.status_code == 200
|
|
assert download.content == payload
|
|
|
|
renamed = await mcp.ok(
|
|
"gitea_release_edit_release_attachment",
|
|
status=(200, 201),
|
|
id=release_id,
|
|
attachment_id=attachment_id,
|
|
body={"name": "renamed.bin"},
|
|
**repo,
|
|
)
|
|
assert renamed.data["name"] == "renamed.bin"
|
|
edited = await mcp.ok(
|
|
"gitea_release_edit_release",
|
|
id=release_id,
|
|
body={"name": "Renamed", "draft": False},
|
|
**repo,
|
|
)
|
|
assert edited.data["name"] == "Renamed"
|
|
assert [r["id"] for r in (await mcp.ok("gitea_release_list_releases", **repo)).data] == [
|
|
release_id
|
|
]
|
|
|
|
await mcp.ok(
|
|
"gitea_release_delete_release_attachment",
|
|
id=release_id,
|
|
attachment_id=attachment_id,
|
|
**repo,
|
|
)
|
|
assert (
|
|
await mcp.ok("gitea_release_list_release_attachments", id=release_id, **repo)
|
|
).data == []
|
|
await mcp.ok("gitea_release_delete_release", id=release_id, **repo)
|
|
await mcp.fail("gitea_release_get_release", 404, id=release_id, **repo)
|
|
|
|
# A second release is removed through its tag.
|
|
await mcp.ok(
|
|
"gitea_release_create_release",
|
|
status=201,
|
|
body={"tag_name": "v1.1.0", "target_commitish": "main"},
|
|
**repo,
|
|
)
|
|
await mcp.ok("gitea_release_delete_release_by_tag", tag="v1.1.0", **repo)
|
|
|
|
|
|
def _names(result: Any) -> list[str]:
|
|
return [item["name"] for item in result.data]
|
|
|
|
|
|
async def test_secrets_and_variables_never_leak_values(
|
|
mcp: McpSession, gitea: GiteaInstance, fresh_repo: RepoRef
|
|
) -> None:
|
|
repo = fresh_repo.args
|
|
org = gitea.unique("it-secrets")
|
|
values = [f"S3cr3t-{secrets.token_hex(12)}" for _ in range(4)]
|
|
repo_secret, org_secret, user_secret, probe = values
|
|
try:
|
|
await mcp.ok("gitea_org_create", status=201, body={"username": org})
|
|
|
|
# Repository, organization and user-level secrets: set, list (names only), delete.
|
|
await mcp.ok(
|
|
"gitea_actions_update_repo_secret",
|
|
secretname="REPO_KEY",
|
|
body={"data": repo_secret, "description": "repo key"},
|
|
**repo,
|
|
)
|
|
await mcp.ok(
|
|
"gitea_actions_update_repo_secret",
|
|
secretname="REPO_KEY",
|
|
body={"data": repo_secret + "x"},
|
|
**repo,
|
|
)
|
|
listed = await mcp.ok("gitea_actions_repo_list_secrets", **repo)
|
|
assert _names(listed) == ["REPO_KEY"]
|
|
assert set(listed.data[0]) <= {"name", "description", "created_at"}
|
|
await mcp.ok(
|
|
"gitea_actions_update_org_secret",
|
|
org=org,
|
|
secretname="ORG_KEY",
|
|
body={"data": org_secret},
|
|
)
|
|
org_listed = await mcp.ok("gitea_actions_org_list_secrets", org=org)
|
|
assert _names(org_listed) == ["ORG_KEY"]
|
|
assert set(org_listed.data[0]) <= {"name", "description", "created_at"}
|
|
await mcp.ok(
|
|
"gitea_actions_update_user_secret", secretname="USER_KEY", body={"data": user_secret}
|
|
)
|
|
|
|
# A failing call carrying a secret must not echo it either.
|
|
denied = await mcp.fail(
|
|
"gitea_actions_update_repo_secret",
|
|
404,
|
|
owner=gitea.admin_login,
|
|
repo="does-not-exist-" + secrets.token_hex(3),
|
|
secretname="NOPE",
|
|
body={"data": probe},
|
|
)
|
|
assert probe not in denied.text
|
|
|
|
await mcp.ok("gitea_actions_delete_repo_secret", secretname="REPO_KEY", **repo)
|
|
await mcp.ok("gitea_actions_delete_org_secret", org=org, secretname="ORG_KEY")
|
|
await mcp.ok("gitea_actions_delete_user_secret", secretname="USER_KEY")
|
|
assert (await mcp.ok("gitea_actions_repo_list_secrets", **repo)).data == []
|
|
|
|
# Variables are not secret: create, get, update, list, delete at every level.
|
|
scopes: list[tuple[str, dict[str, str]]] = [
|
|
("repo", repo),
|
|
("org", {"org": org}),
|
|
("user", {}),
|
|
]
|
|
for level, scope in scopes:
|
|
create = f"gitea_actions_create_{level}_variable"
|
|
await mcp.ok(
|
|
create,
|
|
status=(200, 201, 204),
|
|
variablename="GREETING",
|
|
body={"value": "one"},
|
|
**scope,
|
|
)
|
|
got = await mcp.ok(
|
|
f"gitea_actions_get_{level}_variable", variablename="GREETING", **scope
|
|
)
|
|
assert got.data["data"] == "one"
|
|
await mcp.ok(
|
|
f"gitea_actions_update_{level}_variable",
|
|
variablename="GREETING",
|
|
body={"value": "two"},
|
|
**scope,
|
|
)
|
|
changed = await mcp.ok(
|
|
f"gitea_actions_get_{level}_variable", variablename="GREETING", **scope
|
|
)
|
|
assert changed.data["data"] == "two"
|
|
names = await mcp.ok(f"gitea_actions_get_{level}_variables_list", **scope)
|
|
assert "GREETING" in _names(names)
|
|
await mcp.ok(f"gitea_actions_delete_{level}_variable", variablename="GREETING", **scope)
|
|
await mcp.fail(
|
|
f"gitea_actions_get_{level}_variable", 404, variablename="GREETING", **scope
|
|
)
|
|
finally:
|
|
drop_org(gitea, org)
|
|
gitea.api("DELETE", "/user/actions/secrets/USER_KEY")
|
|
|
|
# No secret value in any tool result, and none in the audit log.
|
|
for value in values:
|
|
assert all(value not in text for text in mcp.texts)
|
|
assert value not in mcp.audit_text()
|
|
audited = [entry["tool_name"] for entry in mcp.audit_entries()]
|
|
assert "gitea_actions_update_repo_secret" in audited
|
|
|
|
|
|
@pytest.mark.parametrize("level", ["repo", "org", "user", "admin"])
|
|
async def test_webhook_lifecycle(
|
|
mcp: McpSession, gitea: GiteaInstance, fresh_repo: RepoRef, level: str
|
|
) -> None:
|
|
org = gitea.unique("it-hooks")
|
|
scope: dict[str, str] = {}
|
|
prefix = f"gitea_hook_{level}"
|
|
try:
|
|
if level == "repo":
|
|
scope = fresh_repo.args
|
|
elif level == "org":
|
|
await mcp.ok("gitea_org_create", status=201, body={"username": org})
|
|
scope = {"org": org}
|
|
if level == "admin":
|
|
create, get, listing, edit, delete = (
|
|
"gitea_admin_create_hook",
|
|
"gitea_admin_get_hook",
|
|
"gitea_admin_list_hooks",
|
|
"gitea_admin_edit_hook",
|
|
"gitea_admin_delete_hook",
|
|
)
|
|
else:
|
|
create, get, listing, edit, delete = (
|
|
f"{prefix}_create_hook",
|
|
f"{prefix}_get_hook",
|
|
f"{prefix}_list_hooks",
|
|
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": config, "events": ["push"], "active": True},
|
|
**scope,
|
|
)
|
|
hook_id = hook.data["id"]
|
|
assert hook.data["type"] == "gitea"
|
|
# Hooks created through /admin/hooks count as "default" hooks, so list both kinds.
|
|
query = {"type": "all"} if level == "admin" else {}
|
|
hooks = await mcp.ok(listing, **scope, **query)
|
|
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"], "config": config},
|
|
**scope,
|
|
)
|
|
assert edited.data["active"] is False
|
|
assert "issues" in edited.data["events"]
|
|
if level == "repo":
|
|
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)
|