From 7d98df7055586606043a78ed7535695ca63fe668 Mon Sep 17 00:00:00 2001 From: Chris Christiansen Date: Mon, 21 Sep 2026 01:44:47 +0000 Subject: [PATCH] feat(opax-mcp): add guarded gitea change proposals --- opax-mcp.yaml | 2 +- opax-mcp/server.py | 236 ++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 236 insertions(+), 2 deletions(-) diff --git a/opax-mcp.yaml b/opax-mcp.yaml index 03714f1..a014f13 100644 --- a/opax-mcp.yaml +++ b/opax-mcp.yaml @@ -54,7 +54,7 @@ spec: key: latest name: gitea-token - name: GITEA_URL - value: "http://34.67.252.59:3000" + value: "https://git.vauco.no" - name: INTERNAL_API_KEY valueFrom: secretKeyRef: diff --git a/opax-mcp/server.py b/opax-mcp/server.py index 5b6b34c..2e43285 100644 --- a/opax-mcp/server.py +++ b/opax-mcp/server.py @@ -7,6 +7,8 @@ import json import uuid import httpx import base64 +from urllib.parse import quote +import difflib import google.auth import google.auth.transport.requests import google.oauth2.id_token @@ -76,8 +78,224 @@ import re from pydantic import BaseModel, Field import hashlib -# --- A2H2A Pydantic Models --- +MAX_PROPOSE_CONTENT_BYTES = 1_048_576 +_REPO_ID_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") +# --- A2H2A Pydantic Models --- +class GiteaChangePlan(BaseModel): + """Immutable admin-approved one-file Gitea change plan.""" + + plan_id: str = Field( + default_factory=lambda: f"gitea-change-{uuid.uuid4().hex}" + ) + status: str = "PENDING" + + created_at: datetime = Field(default_factory=datetime.utcnow) + expires_at: datetime = Field( + default_factory=lambda: datetime.utcnow() + timedelta(minutes=15) + ) + approved_at: Optional[datetime] = None + approved_by: Optional[str] = None + applied_at: Optional[datetime] = None + + repo: str + branch: str + path: str + base_sha: str + new_content: str + content_hash: str + commit_message: str + unified_diff: str + + result_commit_sha: Optional[str] = None + existing_file_sha: Optional[str] = None + apply_idempotency_key: Optional[str] = None + apply_status: Optional[str] = None + apply_error_message: Optional[str] = None + + +async def create_gitea_change_plan(plan: GiteaChangePlan) -> dict: + """Persist a pending one-file Gitea change plan in Firestore.""" + from google.cloud import firestore + + db = firestore.AsyncClient(project=GOOGLE_CLOUD_PROJECT) + plan_data = plan.model_dump(mode="json", exclude={"new_content"}) + plan_data["status"] = "PENDING" + + plan_ref = db.collection("gitea_change_plans").document(plan.plan_id) + await plan_ref.set(plan_data) + + logger.info( + "Created Gitea change plan plan_id=%s repo=%s branch=%s path=%s", + plan.plan_id, + plan.repo, + plan.branch, + plan.path, + ) + return plan_data + + +async def get_gitea_change_plan( + plan_id: str, +) -> Optional[GiteaChangePlan]: + """Return a persisted Gitea change plan, or None when it does not exist.""" + from google.cloud import firestore + + db = firestore.AsyncClient(project=GOOGLE_CLOUD_PROJECT) + plan_ref = db.collection("gitea_change_plans").document(plan_id) + document = await plan_ref.get() + + if not document.exists: + return None + + return GiteaChangePlan(**document.to_dict()) + + +def _validate_admin_gitea_path(path: str) -> None: + """Validate a repository-relative admin file path.""" + if not isinstance(path, str) or not path or path.isspace(): + raise ValueError("Path cannot be empty.") + if path.startswith("/") or "\\" in path or "\x00" in path: + raise ValueError("Path contains unsupported characters.") + segments = path.split("/") + if any(segment in {".", ".."} for segment in segments): + raise ValueError("Path traversal is not allowed.") + + lowered_path = path.lower() + sensitive_markers = ( + ".env", + ".pem", + ".key", + "credentials", + "service_account", + "private_key", + ) + if any(marker in lowered_path for marker in sensitive_markers): + raise ValueError("Path targets a sensitive file.") + + +def _validate_gitea_repo_format(repo: str) -> None: + """Validate the canonical owner/repository identifier format.""" + if not isinstance(repo, str) or not _REPO_ID_RE.fullmatch(repo): + raise ValueError("Repository must use the owner/repository format.") + + +async def _get_gitea_file_details( + repo: str, + path: str, + ref: str, +) -> tuple[Optional[str], Optional[str]]: + """Return decoded file content and file SHA; return (None, None) on 404.""" + api_path = f"/repos/{repo}/contents/{quote(path, safe='/')}?ref={ref}" + + try: + payload = await _gitea_get(api_path) + except httpx.HTTPStatusError as exc: + if exc.response.status_code == 404: + return None, None + logger.warning( + "Gitea file detail lookup failed repo=%s path=%s ref=%s status=%s", + repo, + path, + ref, + exc.response.status_code, + ) + raise ValueError("Gitea API error while reading file details.") from exc + + try: + encoded_content = payload["content"] + content = base64.b64decode(encoded_content).decode("utf-8") + except (KeyError, TypeError, ValueError, UnicodeDecodeError) as exc: + raise ValueError("Gitea returned invalid file content.") from exc + + return content, payload.get("sha") + + +async def propose_gitea_change(p: dict) -> dict: + """Create and persist a read-only, immutable one-file Gitea change plan.""" + repo = p.get("repo") + branch = p.get("branch") + path = p.get("path") + new_content = p.get("new_content") + base_sha = p.get("base_sha") + commit_message = p.get("commit_message") + + if not all((repo, branch, path, base_sha, commit_message)): + raise ValueError( + "Missing required fields: repo, branch, path, base_sha, commit_message." + ) + if new_content is None: + raise ValueError("Missing required field: new_content.") + if not isinstance(new_content, str): + raise ValueError("new_content must be a string.") + if len(new_content.encode("utf-8")) > MAX_PROPOSE_CONTENT_BYTES: + raise ValueError("new_content exceeds the maximum allowed size.") + + _validate_gitea_repo_format(repo) + if repo != GITEA_REPO: + raise ValueError( + "Repository change proposal is not allowed for this repository." + ) + _validate_admin_gitea_path(path) + + live_sha = await resolve_branch_to_commit_sha( + branch_name=branch, + repo_id=repo, + gitea_url=GITEA_URL, + ) + if live_sha != base_sha: + raise ValueError( + "Branch head does not match the supplied base_sha. Refresh and retry." + ) + + old_content, existing_file_sha = await _get_gitea_file_details( + repo=repo, + path=path, + ref=branch, + ) + old_lines = ( + old_content.splitlines(keepends=True) + if old_content is not None + else [] + ) + new_lines = new_content.splitlines(keepends=True) + unified_diff = "".join( + difflib.unified_diff( + old_lines, + new_lines, + fromfile=f"a/{path}", + tofile=f"b/{path}", + ) + ) + if not unified_diff: + raise ValueError("Proposed content produces no file change.") + + content_hash = hashlib.sha256(new_content.encode("utf-8")).hexdigest() + plan = GiteaChangePlan( + repo=repo, + branch=branch, + path=path, + base_sha=base_sha, + new_content=new_content, + content_hash=content_hash, + commit_message=commit_message, + unified_diff=unified_diff, + existing_file_sha=existing_file_sha, + ) + await create_gitea_change_plan(plan) + + return { + "plan_id": plan.plan_id, + "status": "PENDING", + "expires_at": plan.expires_at.isoformat(), + "unified_diff": unified_diff, + "content_hash": content_hash, + "repo": repo, + "branch": branch, + "path": path, + "base_sha": base_sha, + "commit_message": commit_message, + } class TicketSource(BaseModel): @@ -1332,6 +1550,22 @@ TOOLS = { "list_commits": (list_commits, "List siste commits i Gitea-repo", {}), "get_file": (get_file, "Hent fil fra Gitea-repo", {"type":"object","properties":{"path":{"type":"string"}, "ref":{"type":"string"}, "repo":{"type":"string"}},"required":["path", "ref"]}), "list_repo_files": (list_repo_files, "Lists files and directories in a Gitea repository path.", {"type": "object", "properties": {"path": {"type": "string"}}}), + "propose_gitea_change": ( + propose_gitea_change, + "Create a pending, one-file Gitea change plan for review; does not apply or commit changes.", + { + "type": "object", + "properties": { + "repo": {"type": "string"}, + "branch": {"type": "string"}, + "path": {"type": "string"}, + "new_content": {"type": "string"}, + "base_sha": {"type": "string"}, + "commit_message": {"type": "string"} + }, + "required": ["repo", "branch", "path", "new_content", "base_sha", "commit_message"] + }, + ), # Google Workspace "create_email_alias": (create_email_alias, "Opprett et nytt e-postalias", {"type":"object","properties":{"user_key":{"type":"string"},"alias":{"type":"string"}},"required":["user_key","alias"]}), "list_user_aliases": (list_user_aliases, "List en brukers e-postaliaser", {"type":"object","properties":{"user_key":{"type":"string"}},"required":["user_key"]}),