didericis-claude
314dc03b0d
Moves the orchestrator into bot_bottle/orchestrator/ so one install gets everything. Entry point is now `python -m bot_bottle.orchestrator run`. - Add bot_bottle/orchestrator/ with all 14 modules (verbatim move; internal imports were already relative, so no changes inside orchestrator modules) - Rewrite bootstrap.py: remove the lazy bot_bottle import guard, use direct relative imports from ..contrib.* - Add bot_bottle/contrib/forge/base.py: ScopedForge (read-anywhere / write-scoped) - Add bot_bottle/contrib/gitea/client.py: GiteaClient + GiteaForge (urllib.request only) - Add bot_bottle/contrib/gitea/forge_state.py: ForgeState + SqliteForgeStateStore - Add tests/unit/orchestrator/ (82 tests: 63 migrated + 19 new for contrib modules) Closes #321
53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
"""Scoped forge wrapper: read-anywhere / write-scoped access control.
|
|
|
|
`ScopedForge` wraps any forge object and restricts write operations to
|
|
the set of issue/PR numbers the agent is explicitly assigned to. Read
|
|
operations always pass through unconditionally.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
class ScopedForge:
|
|
"""Delegates all forge calls to an inner forge, raising `PermissionError`
|
|
on write calls for numbers outside the assigned scope."""
|
|
|
|
def __init__(
|
|
self,
|
|
forge: Any,
|
|
*,
|
|
assigned_issue: int,
|
|
assigned_prs: list[int],
|
|
) -> None:
|
|
self._forge = forge
|
|
self._allowed_writes: frozenset[int] = frozenset({assigned_issue, *assigned_prs})
|
|
|
|
def _check_write(self, number: int) -> None:
|
|
if number not in self._allowed_writes:
|
|
raise PermissionError(
|
|
f"write to #{number} is outside the assigned scope "
|
|
f"(allowed: {sorted(self._allowed_writes)})"
|
|
)
|
|
|
|
def is_org_member(self, org: str, username: str) -> bool:
|
|
return self._forge.is_org_member(org, username)
|
|
|
|
def read_issue(self, number: int) -> dict[str, Any]:
|
|
return self._forge.read_issue(number)
|
|
|
|
def read_pr(self, number: int) -> dict[str, Any]:
|
|
return self._forge.read_pr(number)
|
|
|
|
def read_comments(self, number: int) -> list[dict[str, Any]]:
|
|
return self._forge.read_comments(number)
|
|
|
|
def post_comment(self, number: int, body: str) -> None:
|
|
self._check_write(number)
|
|
self._forge.post_comment(number, body)
|
|
|
|
def update_description(self, number: int, body: str) -> None:
|
|
self._check_write(number)
|
|
self._forge.update_description(number, body)
|