feat: scope list_repositories to the authenticated user in service-PAT mode
Previously list_repositories was blocked in service-PAT mode because it has no repository target for the per-user permission check, so users could not list their repositories at all (the connector surfaced a generic error). list_repositories now returns only the repositories the signed-in user owns or contributes to, instead of everything the bot token can see: - gitea_client.py: add list_user_repositories(login) — resolves the user id and queries /api/v1/repos/search with the uid filter. - repository.py: list_repositories_tool uses the user-scoped path when a service PAT is configured and a user login is present; pure-OAuth mode still uses the user's own /user/repos. - server.py: allow list_repositories through the service-PAT guard (it is scoped to the user in the handler); all other tools still require a repository target. - README.md: document the new user-scoped behavior and its visibility caveat. Tests: user-scoped client method (uid resolution + unknown user), PAT-mode tool scoping, and conftest now clears the request context between tests to prevent contextvar login leakage across files. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -173,6 +173,57 @@ class GiteaClient:
|
||||
)
|
||||
raise
|
||||
|
||||
async def list_user_repositories(self, login: str) -> list[dict[str, Any]]:
|
||||
"""List repositories the given user owns or contributes to.
|
||||
|
||||
Used in service-PAT mode so ``list_repositories`` returns repositories
|
||||
scoped to the authenticated user instead of everything the service token
|
||||
can see. Resolves the user id, then queries Gitea's repo search with the
|
||||
``uid`` filter. Visibility of private repos still depends on what the
|
||||
service token itself can see.
|
||||
"""
|
||||
correlation_id = self.audit.log_tool_invocation(
|
||||
tool_name="list_user_repositories",
|
||||
result_status="pending",
|
||||
)
|
||||
try:
|
||||
user = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/users/{quote(login, safe='')}",
|
||||
correlation_id=correlation_id,
|
||||
)
|
||||
uid = user.get("id") if isinstance(user, dict) else None
|
||||
if not isinstance(uid, int):
|
||||
return []
|
||||
|
||||
result = await self._request(
|
||||
"GET",
|
||||
"/api/v1/repos/search",
|
||||
params={"uid": uid, "limit": 50, "page": 1},
|
||||
correlation_id=correlation_id,
|
||||
)
|
||||
repositories: list[dict[str, Any]] = []
|
||||
if isinstance(result, dict):
|
||||
data = result.get("data", [])
|
||||
if isinstance(data, list):
|
||||
repositories = [item for item in data if isinstance(item, dict)]
|
||||
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="list_user_repositories",
|
||||
correlation_id=correlation_id,
|
||||
result_status="success",
|
||||
params={"login": login, "count": len(repositories)},
|
||||
)
|
||||
return repositories
|
||||
except Exception as exc:
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="list_user_repositories",
|
||||
correlation_id=correlation_id,
|
||||
result_status="error",
|
||||
error=str(exc),
|
||||
)
|
||||
raise
|
||||
|
||||
async def get_repository(self, owner: str, repo: str) -> dict[str, Any]:
|
||||
"""Get repository metadata."""
|
||||
repo_id = f"{owner}/{repo}"
|
||||
|
||||
Reference in New Issue
Block a user