feat: harden gateway with policy engine, secure tools, and governance docs

src/aegis_gitea_mcp/tools/arguments.py

@@ -0,0 +1,208 @@
+"""Pydantic argument models for MCP tools."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+_REPO_PART_PATTERN = r"^[A-Za-z0-9._-]{1,100}$"
+
+
+class StrictBaseModel(BaseModel):
+    """Strict model base that rejects unexpected fields."""
+
+    model_config = ConfigDict(extra="forbid")
+
+
+class ListRepositoriesArgs(StrictBaseModel):
+    """Arguments for list_repositories tool."""
+
+
+class RepositoryArgs(StrictBaseModel):
+    """Common repository locator arguments."""
+
+    owner: str = Field(..., pattern=_REPO_PART_PATTERN)
+    repo: str = Field(..., pattern=_REPO_PART_PATTERN)
+
+
+class FileTreeArgs(RepositoryArgs):
+    """Arguments for get_file_tree."""
+
+    ref: str = Field(default="main", min_length=1, max_length=200)
+    recursive: bool = Field(default=False)
+
+
+class FileContentsArgs(RepositoryArgs):
+    """Arguments for get_file_contents."""
+
+    filepath: str = Field(..., min_length=1, max_length=1024)
+    ref: str = Field(default="main", min_length=1, max_length=200)
+
+    @model_validator(mode="after")
+    def validate_filepath(self) -> FileContentsArgs:
+        """Validate path safety constraints."""
+        normalized = self.filepath.replace("\\", "/")
+        # Security decision: block traversal and absolute paths.
+        if normalized.startswith("/") or ".." in normalized.split("/"):
+            raise ValueError("filepath must be a relative path without traversal")
+        if "\x00" in normalized:
+            raise ValueError("filepath cannot contain null bytes")
+        return self
+
+
+class SearchCodeArgs(RepositoryArgs):
+    """Arguments for search_code."""
+
+    query: str = Field(..., min_length=1, max_length=256)
+    ref: str = Field(default="main", min_length=1, max_length=200)
+    page: int = Field(default=1, ge=1, le=10_000)
+    limit: int = Field(default=25, ge=1, le=100)
+
+
+class ListCommitsArgs(RepositoryArgs):
+    """Arguments for list_commits."""
+
+    ref: str = Field(default="main", min_length=1, max_length=200)
+    page: int = Field(default=1, ge=1, le=10_000)
+    limit: int = Field(default=25, ge=1, le=100)
+
+
+class CommitDiffArgs(RepositoryArgs):
+    """Arguments for get_commit_diff."""
+
+    sha: str = Field(..., min_length=7, max_length=64)
+
+
+class CompareRefsArgs(RepositoryArgs):
+    """Arguments for compare_refs."""
+
+    base: str = Field(..., min_length=1, max_length=200)
+    head: str = Field(..., min_length=1, max_length=200)
+
+
+class ListIssuesArgs(RepositoryArgs):
+    """Arguments for list_issues."""
+
+    state: Literal["open", "closed", "all"] = Field(default="open")
+    page: int = Field(default=1, ge=1, le=10_000)
+    limit: int = Field(default=25, ge=1, le=100)
+    labels: list[str] = Field(default_factory=list, max_length=20)
+
+
+class IssueArgs(RepositoryArgs):
+    """Arguments for get_issue."""
+
+    issue_number: int = Field(..., ge=1)
+
+
+class ListPullRequestsArgs(RepositoryArgs):
+    """Arguments for list_pull_requests."""
+
+    state: Literal["open", "closed", "all"] = Field(default="open")
+    page: int = Field(default=1, ge=1, le=10_000)
+    limit: int = Field(default=25, ge=1, le=100)
+
+
+class PullRequestArgs(RepositoryArgs):
+    """Arguments for get_pull_request."""
+
+    pull_number: int = Field(..., ge=1)
+
+
+class ListLabelsArgs(RepositoryArgs):
+    """Arguments for list_labels."""
+
+    page: int = Field(default=1, ge=1, le=10_000)
+    limit: int = Field(default=50, ge=1, le=100)
+
+
+class ListTagsArgs(RepositoryArgs):
+    """Arguments for list_tags."""
+
+    page: int = Field(default=1, ge=1, le=10_000)
+    limit: int = Field(default=50, ge=1, le=100)
+
+
+class ListReleasesArgs(RepositoryArgs):
+    """Arguments for list_releases."""
+
+    page: int = Field(default=1, ge=1, le=10_000)
+    limit: int = Field(default=25, ge=1, le=100)
+
+
+class CreateIssueArgs(RepositoryArgs):
+    """Arguments for create_issue."""
+
+    title: str = Field(..., min_length=1, max_length=256)
+    body: str = Field(default="", max_length=20_000)
+    labels: list[str] = Field(default_factory=list, max_length=20)
+    assignees: list[str] = Field(default_factory=list, max_length=20)
+
+
+class UpdateIssueArgs(RepositoryArgs):
+    """Arguments for update_issue."""
+
+    issue_number: int = Field(..., ge=1)
+    title: str | None = Field(default=None, min_length=1, max_length=256)
+    body: str | None = Field(default=None, max_length=20_000)
+    state: Literal["open", "closed"] | None = Field(default=None)
+
+    @model_validator(mode="after")
+    def require_change(self) -> UpdateIssueArgs:
+        """Require at least one mutable field in update payload."""
+        if self.title is None and self.body is None and self.state is None:
+            raise ValueError("At least one of title, body, or state must be provided")
+        return self
+
+
+class CreateIssueCommentArgs(RepositoryArgs):
+    """Arguments for create_issue_comment."""
+
+    issue_number: int = Field(..., ge=1)
+    body: str = Field(..., min_length=1, max_length=10_000)
+
+
+class CreatePrCommentArgs(RepositoryArgs):
+    """Arguments for create_pr_comment."""
+
+    pull_number: int = Field(..., ge=1)
+    body: str = Field(..., min_length=1, max_length=10_000)
+
+
+class AddLabelsArgs(RepositoryArgs):
+    """Arguments for add_labels."""
+
+    issue_number: int = Field(..., ge=1)
+    labels: list[str] = Field(..., min_length=1, max_length=20)
+
+
+class AssignIssueArgs(RepositoryArgs):
+    """Arguments for assign_issue."""
+
+    issue_number: int = Field(..., ge=1)
+    assignees: list[str] = Field(..., min_length=1, max_length=20)
+
+
+def extract_repository(arguments: dict[str, object]) -> str | None:
+    """Extract `owner/repo` from raw argument mapping.
+
+    Args:
+        arguments: Raw tool arguments.
+
+    Returns:
+        `owner/repo` or None when arguments are incomplete.
+    """
+    owner = arguments.get("owner")
+    repo = arguments.get("repo")
+    if isinstance(owner, str) and isinstance(repo, str) and owner and repo:
+        return f"{owner}/{repo}"
+    return None
+
+
+def extract_target_path(arguments: dict[str, object]) -> str | None:
+    """Extract optional target path argument for policy path checks."""
+    filepath = arguments.get("filepath")
+    if isinstance(filepath, str) and filepath:
+        return filepath
+    return None