297 lines
11 KiB
Python
297 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
import unittest
|
|
|
|
from tea_mcp.persimmons import list_connections
|
|
from tea_mcp.persimmons_web import PersimmonsWebRequester, _repo_results, parse_secret_page, parse_variable_page
|
|
|
|
|
|
class PersimmonsWebSessionParserTests(unittest.TestCase):
|
|
def test_gitea_127_secret_markup_is_safe_and_identifiable(self) -> None:
|
|
page = """
|
|
<div class="item">
|
|
<button data-modal-secret-name.value="CLOUDFLARE_TOKEN"></button>
|
|
<button data-url="/user/settings/actions/secrets/delete?id=41"></button>
|
|
</div>
|
|
<div class="item">
|
|
<button data-modal-secret-name.value="DEPLOY_PAT"></button>
|
|
<button data-url="/user/settings/actions/secrets/delete?id=42"></button>
|
|
</div>
|
|
"""
|
|
items = parse_secret_page(page)
|
|
self.assertEqual(
|
|
[(item["name"], item["id"]) for item in items],
|
|
[("CLOUDFLARE_TOKEN", 41), ("DEPLOY_PAT", 42)],
|
|
)
|
|
self.assertNotIn("data", items[0])
|
|
|
|
def test_gitea_127_variable_markup_yields_non_secret_metadata(self) -> None:
|
|
page = """
|
|
<button
|
|
data-modal-form.url="/user/settings/actions/variables/9/edit"
|
|
data-modal-dialog-variable-name="PERSIM_META_ABCDEF012345"
|
|
data-modal-dialog-variable-data="{"provider":"Gitea"}"
|
|
data-modal-dialog-variable-description="Persimmons metadata"
|
|
></button>
|
|
"""
|
|
items = parse_variable_page(page)
|
|
self.assertEqual(len(items), 1)
|
|
self.assertEqual(items[0]["id"], 9)
|
|
self.assertEqual(items[0]["name"], "PERSIM_META_ABCDEF012345")
|
|
self.assertEqual(items[0]["data"], '{"provider":"Gitea"}')
|
|
self.assertEqual(items[0]["description"], "Persimmons metadata")
|
|
|
|
|
|
def test_gitea_127_web_repo_search_unwraps_nested_repository(self) -> None:
|
|
payload = {
|
|
"ok": True,
|
|
"data": [
|
|
{
|
|
"repository": {
|
|
"full_name": "acme/example",
|
|
"private": True,
|
|
},
|
|
"latest_commit_status": None,
|
|
}
|
|
],
|
|
}
|
|
self.assertEqual(
|
|
_repo_results(payload),
|
|
[{"full_name": "acme/example", "private": True}],
|
|
)
|
|
|
|
def test_gitea_web_repo_search_rejects_unknown_shape(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
_repo_results({"ok": True})
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, status_code: int, payload=None) -> None:
|
|
self.status_code = status_code
|
|
self._payload = payload
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
|
|
class _RepoDiscoveryRequester(PersimmonsWebRequester):
|
|
async def _get(self, path: str, *, params=None, accept_json: bool = False):
|
|
if path == "/repo/search":
|
|
return _FakeResponse(
|
|
200,
|
|
{
|
|
"ok": True,
|
|
"data": [
|
|
{"repository": {"full_name": "matcha/owned", "private": True}},
|
|
{"repository": {"full_name": "shared/collab", "private": True}},
|
|
{"repository": {"full_name": "other/hidden", "private": True}},
|
|
],
|
|
},
|
|
)
|
|
if path in {
|
|
"/matcha/owned/settings/actions/secrets",
|
|
"/shared/collab/settings/actions/secrets",
|
|
}:
|
|
return _FakeResponse(200)
|
|
if path == "/other/hidden/settings/actions/secrets":
|
|
return _FakeResponse(404)
|
|
raise AssertionError(path)
|
|
|
|
|
|
class PersimmonsRepositoryDiscoveryTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_discovery_keeps_writable_collaborator_repo_and_filters_inaccessible(self) -> None:
|
|
requester = _RepoDiscoveryRequester("https://example.invalid", "cookie", "actor")
|
|
repos = await requester._repositories({})
|
|
self.assertEqual(
|
|
[repo["full_name"] for repo in repos],
|
|
["matcha/owned", "shared/collab"],
|
|
)
|
|
|
|
|
|
class _ActorScopedRequester(PersimmonsWebRequester):
|
|
ALL_REPOS = [
|
|
"teahub-admin/bootstrap-persistence-check",
|
|
"matcha/greentea",
|
|
"matcha/persimmons",
|
|
"chai/pointtest",
|
|
"matcha/teahub",
|
|
]
|
|
|
|
def __init__(self, actor_name: str, allowed_repos: set[str], user_secret: str) -> None:
|
|
super().__init__("https://example.invalid", f"cookie-{actor_name}", actor_name)
|
|
self.allowed_repos = allowed_repos
|
|
self.user_secret = user_secret
|
|
self.posts: list[tuple[str, dict[str, str]]] = []
|
|
|
|
async def _get(self, path: str, *, params=None, accept_json: bool = False):
|
|
if path == "/repo/search":
|
|
return _FakeResponse(
|
|
200,
|
|
{
|
|
"ok": True,
|
|
"data": [
|
|
{"repository": {"full_name": name, "private": True}}
|
|
for name in self.ALL_REPOS
|
|
],
|
|
},
|
|
)
|
|
if path == "/user/settings/actions/secrets":
|
|
return _HtmlResponse(
|
|
200,
|
|
f'<button data-modal-secret-name.value="{self.user_secret}"></button>'
|
|
'<button data-url="/user/settings/actions/secrets/delete?id=1"></button>',
|
|
)
|
|
if path == "/user/settings/actions/variables":
|
|
return _HtmlResponse(200, "")
|
|
match = re.fullmatch(r"/([^/]+/[^/]+)/settings/actions/(secrets|variables)", path)
|
|
if match:
|
|
repository, kind = match.groups()
|
|
if repository not in self.allowed_repos:
|
|
return _HtmlResponse(404, "")
|
|
if kind == "variables":
|
|
return _HtmlResponse(200, "")
|
|
secret = "POINTTEST_TOKEN" if repository == "chai/pointtest" else "REPO_TOKEN"
|
|
return _HtmlResponse(
|
|
200,
|
|
f'<button data-modal-secret-name.value="{secret}"></button>'
|
|
f'<button data-url="/{repository}/settings/actions/secrets/delete?id=2"></button>',
|
|
)
|
|
raise AssertionError(path)
|
|
|
|
async def _post(self, path: str, data: dict[str, str], *, params=None):
|
|
self.posts.append((path, data))
|
|
return _HtmlResponse(302, "")
|
|
|
|
|
|
class _MutationRequester(PersimmonsWebRequester):
|
|
def __init__(self) -> None:
|
|
super().__init__("https://example.invalid", "session-cookie", "actor")
|
|
self.posts: list[tuple[str, dict[str, str], object]] = []
|
|
|
|
async def _get(self, path: str, *, params=None, accept_json: bool = False):
|
|
if path.endswith("/secrets"):
|
|
return _HtmlResponse(
|
|
200,
|
|
'<button data-modal-secret-name.value="TEST_FIXTURE"></button>'
|
|
'<button data-url="/user/settings/actions/secrets/delete?id=7"></button>',
|
|
)
|
|
if path.endswith("/variables"):
|
|
return _HtmlResponse(200, "")
|
|
raise AssertionError(path)
|
|
|
|
async def _post(self, path: str, data: dict[str, str], *, params=None):
|
|
self.posts.append((path, data, params))
|
|
return _HtmlResponse(302, "")
|
|
|
|
|
|
class PersimmonsMutationContractTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_secret_write_does_not_require_legacy_csrf_field(self) -> None:
|
|
requester = _MutationRequester()
|
|
await requester._secret_put(
|
|
"/user/settings/actions",
|
|
"TEST_FIXTURE",
|
|
"fixture-value",
|
|
"fixture",
|
|
)
|
|
path, data, params = requester.posts[-1]
|
|
self.assertEqual(path, "/user/settings/actions/secrets")
|
|
self.assertEqual(data["name"], "TEST_FIXTURE")
|
|
self.assertEqual(data["data"], "fixture-value")
|
|
self.assertNotIn("_csrf", data)
|
|
self.assertIsNone(params)
|
|
|
|
async def test_secret_delete_does_not_require_legacy_csrf_field(self) -> None:
|
|
requester = _MutationRequester()
|
|
await requester._secret_delete("/user/settings/actions", "TEST_FIXTURE")
|
|
path, data, params = requester.posts[-1]
|
|
self.assertEqual(path, "/user/settings/actions/secrets/delete")
|
|
self.assertEqual(data, {"id": "7"})
|
|
self.assertEqual(params, {"id": 7})
|
|
|
|
|
|
class _HtmlResponse(_FakeResponse):
|
|
def __init__(self, status_code: int, text: str) -> None:
|
|
super().__init__(status_code)
|
|
self.text = text
|
|
|
|
|
|
class PersimmonsMultiUserIsolationTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_distinct_callers_get_distinct_repository_sets(self) -> None:
|
|
matcha = _ActorScopedRequester(
|
|
"matcha",
|
|
set(_ActorScopedRequester.ALL_REPOS),
|
|
"MATCHA_USER_TOKEN",
|
|
)
|
|
chai = _ActorScopedRequester(
|
|
"chai",
|
|
{"chai/pointtest"},
|
|
"CHAI_USER_TOKEN",
|
|
)
|
|
|
|
matcha_repos = await matcha._repositories({})
|
|
chai_repos = await chai._repositories({})
|
|
|
|
self.assertEqual(
|
|
[repo["full_name"] for repo in matcha_repos],
|
|
_ActorScopedRequester.ALL_REPOS,
|
|
)
|
|
self.assertEqual(
|
|
[repo["full_name"] for repo in chai_repos],
|
|
["chai/pointtest"],
|
|
)
|
|
|
|
async def test_user_vault_listing_is_bound_to_session(self) -> None:
|
|
matcha = _ActorScopedRequester("matcha", set(), "MATCHA_USER_TOKEN")
|
|
chai = _ActorScopedRequester("chai", set(), "CHAI_USER_TOKEN")
|
|
|
|
matcha_vault = await list_connections(matcha, include_legacy=False)
|
|
chai_vault = await list_connections(chai, include_legacy=False)
|
|
|
|
self.assertEqual(
|
|
[item["name"] for item in matcha_vault["connections"]],
|
|
["MATCHA_USER_TOKEN"],
|
|
)
|
|
self.assertEqual(
|
|
[item["name"] for item in chai_vault["connections"]],
|
|
["CHAI_USER_TOKEN"],
|
|
)
|
|
|
|
async def test_guessed_repository_cannot_bypass_caller_authorization(self) -> None:
|
|
chai = _ActorScopedRequester("chai", {"chai/pointtest"}, "CHAI_USER_TOKEN")
|
|
|
|
with self.assertRaises(RuntimeError):
|
|
await list_connections(
|
|
chai,
|
|
repository="matcha/teahub",
|
|
include_user=False,
|
|
include_legacy=False,
|
|
)
|
|
|
|
with self.assertRaises(RuntimeError):
|
|
await chai(
|
|
"PUT",
|
|
"repos/matcha/teahub/actions/secrets/TEST_FIXTURE",
|
|
json={"data": "fixture-only"},
|
|
)
|
|
|
|
self.assertEqual(chai.posts, [])
|
|
|
|
async def test_authorized_repository_projects_safe_secret_metadata_only(self) -> None:
|
|
chai = _ActorScopedRequester("chai", {"chai/pointtest"}, "CHAI_USER_TOKEN")
|
|
listed = await list_connections(
|
|
chai,
|
|
repository="chai/pointtest",
|
|
include_user=False,
|
|
include_legacy=False,
|
|
)
|
|
item = listed["connections"][0]
|
|
self.assertEqual(item["name"], "POINTTEST_TOKEN")
|
|
self.assertEqual(item["repository"], "chai/pointtest")
|
|
self.assertTrue(item["available"])
|
|
self.assertNotIn("credential", item)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|