103 lines
5.8 KiB
Python
103 lines
5.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
|
|
from tea_mcp.persimmons import create_connection, create_juice, list_juice, revoke_connection, revoke_juice
|
|
from tests.test_persimmons import FakeGitea
|
|
|
|
|
|
class JuiceTests(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
self.store = FakeGitea()
|
|
self.source = await create_connection(self.store.request, {
|
|
"provider": "Example", "credential_type": "Token", "credential": "source-secret-value",
|
|
"vault_kind": "user",
|
|
})
|
|
self.data = {"source_id": self.source["id"], "subject": "agent:example", "scopes": ["read"]}
|
|
self.calls = []
|
|
|
|
async def request(self, method, path, **kwargs):
|
|
self.calls.append((method, path))
|
|
# Juice must not read individual secrets or mutate any source secret.
|
|
self.assertNotIn("/secrets/", path)
|
|
return await self.store.request(method, path, **kwargs)
|
|
|
|
async def test_lifecycle_metadata_only_and_source_survives_revoke(self):
|
|
created = await create_juice(self.request, self.data)
|
|
self.assertEqual(created["kind"], "juice")
|
|
self.assertEqual(created["status"], "active")
|
|
self.assertEqual(created["source"], "Example")
|
|
self.assertEqual(created["vault"], {"kind": "user"})
|
|
self.assertIsNone(created["expires_at"])
|
|
self.assertTrue(created["revocable"])
|
|
self.assertIn("PERSIM_JUICE_" + created["id"], self.store.user_variables)
|
|
self.calls.clear()
|
|
listed = await list_juice(self.request, source_id=self.source["id"])
|
|
self.assertEqual(listed, {"juice": [created]})
|
|
self.assertTrue(all("/secrets" not in path for _, path in self.calls))
|
|
self.assertEqual(await list_juice(self.request, source_id="AAAAAAAAAAAA"), {"juice": []})
|
|
revoked = await revoke_juice(self.request, created["id"])
|
|
self.assertEqual(revoked["status"], "revoked")
|
|
self.assertEqual(await revoke_juice(self.request, created["id"]), revoked)
|
|
self.assertEqual(await list_juice(self.request), {"juice": [revoked]})
|
|
self.assertNotIn("source-secret-value", json.dumps([created, listed, revoked, self.store.user_variables]))
|
|
self.assertEqual(self.store.user_secrets, {"EXAMPLE": "source-secret-value"})
|
|
|
|
async def test_missing_revoked_and_unavailable_sources_rejected(self):
|
|
with self.assertRaisesRegex(ValueError, "not found"):
|
|
await create_juice(self.request, {**self.data, "source_id": "AAAAAAAAAAAA"})
|
|
self.store.user_secrets.clear()
|
|
with self.assertRaisesRegex(ValueError, "unavailable"):
|
|
await create_juice(self.request, self.data)
|
|
await revoke_connection(self.store.request, self.source["id"], {"vault_kind": "user"})
|
|
self.store.user_secrets["EXAMPLE"] = "source-secret-value"
|
|
with self.assertRaisesRegex(ValueError, "revoked"):
|
|
await create_juice(self.request, self.data)
|
|
|
|
async def test_repository_isolation_and_mismatched_metadata(self):
|
|
repo = {"repository": "matcha/example"}
|
|
with self.assertRaisesRegex(ValueError, "not found"):
|
|
await create_juice(self.request, {**self.data, **repo})
|
|
key = "PERSIM_META_" + self.source["id"]
|
|
self.store.repo_variables[repo["repository"]][key] = self.store.user_variables[key]
|
|
with self.assertRaisesRegex(ValueError, "vault does not match"):
|
|
await create_juice(self.request, {**self.data, **repo})
|
|
source = await create_connection(self.store.request, {
|
|
"provider": "Example", "credential_type": "Token", "credential": "repo-secret",
|
|
"vault_kind": "repository", **repo,
|
|
})
|
|
juice = await create_juice(self.request, {**self.data, **repo, "source_id": source["id"]})
|
|
self.assertEqual(await list_juice(self.request), {"juice": []})
|
|
self.assertEqual(await list_juice(self.request, **repo), {"juice": [juice]})
|
|
with self.assertRaisesRegex(ValueError, "not found"):
|
|
await revoke_juice(self.request, juice["id"])
|
|
copied = "PERSIM_JUICE_" + juice["id"]
|
|
self.store.user_variables[copied] = self.store.repo_variables[repo["repository"]][copied]
|
|
with self.assertRaisesRegex(ValueError, "vault does not match"):
|
|
await revoke_juice(self.request, juice["id"])
|
|
self.assertEqual(await list_juice(self.request), {"juice": []})
|
|
await revoke_juice(self.request, juice["id"], repo)
|
|
self.assertEqual(self.store.repo_secrets[repo["repository"]]["EXAMPLE"], "repo-secret")
|
|
|
|
async def test_validation_rejects_unbounded_or_ambiguous_policy(self):
|
|
for override in (
|
|
{"subject": ""}, {"subject": "x" * 201}, {"scopes": []}, {"scopes": "read"},
|
|
{"scopes": ["x"] * 41}, {"scopes": ["x" * 101]}, {"scopes": [{}]},
|
|
{"expires_at": "tomorrow"}, {"expires_at": "2000-01-01T00:00:00Z"},
|
|
{"expires_at": "2099-01-01T00:00:00"}, {"source_id": "../secret"},
|
|
{"vault_kind": "user", "repository": "matcha/example"},
|
|
):
|
|
with self.subTest(override=override), self.assertRaises(ValueError):
|
|
await create_juice(self.request, {**self.data, **override})
|
|
created = await create_juice(self.request, {**self.data, "expires_at": "2099-01-01T00:00:00Z"})
|
|
self.assertEqual(created["expires_at"], "2099-01-01T00:00:00Z")
|
|
|
|
async def test_listing_projects_only_juice_fields(self):
|
|
created = await create_juice(self.request, self.data)
|
|
key = "PERSIM_JUICE_" + created["id"]
|
|
metadata = json.loads(self.store.user_variables[key])
|
|
metadata["credential"] = "unexpected-private-field"
|
|
self.store.user_variables[key] = json.dumps(metadata)
|
|
self.assertNotIn("unexpected-private-field", json.dumps(await list_juice(self.request)))
|