450 lines
15 KiB
Python
450 lines
15 KiB
Python
import os
|
|
import httpx
|
|
import logging
|
|
from urllib.parse import quote
|
|
import re
|
|
import base64
|
|
import json
|
|
import binascii
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_SHA_RE = re.compile(r"^[0-9a-f]{40}$")
|
|
_REPO_ID_RE = re.compile(
|
|
r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}/"
|
|
r"[A-Za-z0-9][A-Za-z0-9._-]{0,99}$"
|
|
)
|
|
_BRANCH_RE = re.compile(
|
|
r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$"
|
|
)
|
|
|
|
MAX_SOURCE_ARCHIVE_BYTES = int(
|
|
os.environ.get("OPAX_MAX_SOURCE_ARCHIVE_BYTES", "104857600")
|
|
)
|
|
|
|
ALLOWED_GITEA_NAMESPACE = "chris"
|
|
CATALOG_PAGE_SIZE = 50
|
|
CATALOG_MAX_PAGES = 2
|
|
CATALOG_MAX_RESULTS = 100
|
|
MAX_READ_FILE_BYTES = 1_048_576
|
|
MAX_GET_FILE_RESPONSE_BYTES = 1_500_000
|
|
|
|
def _gitea_headers() -> dict:
|
|
"""Constructs headers for Gitea API requests."""
|
|
return {"Authorization": f"token {os.environ.get('GITEA_TOKEN')}", "Accept": "application/json"}
|
|
|
|
def _validate_repo_id(repo_id: str) -> str:
|
|
if not isinstance(repo_id, str) or not _REPO_ID_RE.fullmatch(repo_id):
|
|
raise ValueError("Invalid configured Gitea repository ID")
|
|
return repo_id
|
|
|
|
def _validate_gitea_url(gitea_url: str) -> str:
|
|
if not isinstance(gitea_url, str):
|
|
raise ValueError("GITEA_URL environment variable is not set")
|
|
|
|
normalized = gitea_url.rstrip("/")
|
|
|
|
if not normalized.startswith("https://"):
|
|
raise ValueError("GITEA_URL must use HTTPS")
|
|
|
|
return normalized
|
|
|
|
def _validate_branch_name(branch_name: str) -> str:
|
|
if not isinstance(branch_name, str):
|
|
raise ValueError("Branch name must be a string")
|
|
|
|
if not _BRANCH_RE.fullmatch(branch_name):
|
|
raise ValueError("Invalid branch name")
|
|
|
|
if (
|
|
branch_name.startswith((".", "/"))
|
|
or branch_name.endswith((".", "/"))
|
|
or ".." in branch_name
|
|
or "//" in branch_name
|
|
or "@{" in branch_name
|
|
or branch_name.endswith(".lock")
|
|
):
|
|
raise ValueError("Invalid branch name")
|
|
|
|
return branch_name
|
|
|
|
def _validate_commit_sha(commit_sha: str) -> str:
|
|
if not isinstance(commit_sha, str) or not _SHA_RE.fullmatch(commit_sha):
|
|
raise ValueError("Invalid commit SHA")
|
|
|
|
return commit_sha
|
|
|
|
def _validate_safe_path(path: str) -> str:
|
|
"""Validates a file path against a strict allowlist and safety rules."""
|
|
if not path or not isinstance(path, str):
|
|
raise ValueError("Repository file request is not allowed.")
|
|
|
|
# 1. Reject malformed path
|
|
if '\\' in path or '\0' in path or '//' in path:
|
|
raise ValueError("Repository file request is not allowed.")
|
|
|
|
# 2. Reject traversal/absolute
|
|
if path.startswith('/') or '..' in path.split('/'):
|
|
raise ValueError("Repository file request is not allowed.")
|
|
|
|
# 3. Apply exact root/prefix allowlist
|
|
is_allowed = (
|
|
path == "README.md" or
|
|
path == "GEMINI.md" or
|
|
path.startswith("docs/") or
|
|
path.startswith(".gemini/") or
|
|
path.startswith("README/")
|
|
)
|
|
if not is_allowed:
|
|
raise ValueError("Repository file request is not allowed.")
|
|
|
|
# 4. Apply deny rules
|
|
denied_patterns = [
|
|
".env", ".pem", ".key", ".p12", ".pfx", "id_rsa", "id_ed25519",
|
|
"credentials", "secret"
|
|
]
|
|
path_segments = path.lower().split('/')
|
|
for segment in path_segments:
|
|
if any(pattern in segment for pattern in denied_patterns):
|
|
raise ValueError("Repository file request is not allowed.")
|
|
if not path.startswith("docs/") and path.endswith(".json"):
|
|
raise ValueError("Repository file request is not allowed.")
|
|
|
|
return path
|
|
|
|
|
|
MAX_PUSH_CONTENT_BYTES = 1_048_576 # 1 MB
|
|
|
|
def validate_repo_for_write(repo_id: str) -> str:
|
|
"""Validates that a repo is in the allowed namespace for write operations."""
|
|
if not isinstance(repo_id, str) or not _REPO_ID_RE.fullmatch(repo_id):
|
|
raise ValueError("Invalid repository format. Must be 'owner/repo'.")
|
|
if not repo_id.startswith(f"{ALLOWED_GITEA_NAMESPACE}/"):
|
|
raise ValueError(f"Write operations are only allowed in the '{ALLOWED_GITEA_NAMESPACE}' namespace.")
|
|
return repo_id
|
|
|
|
def validate_branch_for_write(branch: str) -> str:
|
|
"""Validates that a branch name is safe for write operations."""
|
|
if not branch or not isinstance(branch, str):
|
|
raise ValueError("Branch name cannot be empty.")
|
|
if branch.lower() in ["main", "master"]:
|
|
raise ValueError(f"Direct writes to protected branch '{branch}' are not allowed.")
|
|
return _validate_branch_name(branch)
|
|
|
|
def validate_path_for_write(path: str) -> str:
|
|
"""Validates a file path for write operations with segment-based checks."""
|
|
if not path or not isinstance(path, str):
|
|
raise ValueError("Path cannot be empty.")
|
|
if "\\" in path or "\x00" in path or "//" in path:
|
|
raise ValueError("Path contains invalid characters.")
|
|
if path.startswith('/') or '..' in path.split('/'):
|
|
raise ValueError("Path must be relative and cannot contain traversal elements.")
|
|
|
|
path_segments = path.lower().split('/')
|
|
filename = path_segments[-1]
|
|
|
|
if '.git' in path_segments:
|
|
raise ValueError("Changes within a '.git' directory are not allowed.")
|
|
if filename == '.env' or filename.startswith('.env.'):
|
|
raise ValueError("Path targets a '.env' file, which is not allowed.")
|
|
sensitive_basenames = ["credentials", "service_account", "private_key", "id_rsa", "id_ed25519", "secret", "token"]
|
|
if filename in sensitive_basenames or any(filename.endswith(ext) for ext in ['.pem', '.key']):
|
|
raise ValueError(f"Path targets a sensitive basename or extension.")
|
|
return path
|
|
|
|
|
|
async def resolve_branch_to_commit_sha(
|
|
branch_name: str,
|
|
repo_id: str,
|
|
gitea_url: str,
|
|
) -> str:
|
|
"""Resolve an allowed branch name to one immutable 40-character SHA."""
|
|
validated_branch = _validate_branch_name(branch_name)
|
|
validated_repo = _validate_repo_id(repo_id)
|
|
validated_url = _validate_gitea_url(gitea_url)
|
|
|
|
encoded_branch = quote(validated_branch, safe="")
|
|
url = (
|
|
f"{validated_url}/api/v1/repos/"
|
|
f"{validated_repo}/branches/{encoded_branch}"
|
|
)
|
|
|
|
async with httpx.AsyncClient(
|
|
timeout=httpx.Timeout(15.0),
|
|
follow_redirects=False,
|
|
) as client:
|
|
response = await client.get(
|
|
url,
|
|
headers=_gitea_headers(),
|
|
)
|
|
|
|
response.raise_for_status()
|
|
|
|
commit_sha = response.json().get("commit", {}).get("id")
|
|
return _validate_commit_sha(commit_sha)
|
|
|
|
|
|
async def download_repo_archive(
|
|
commit_sha: str,
|
|
repo_id: str,
|
|
gitea_url: str,
|
|
) -> bytes:
|
|
"""Download a bounded source archive for an already resolved SHA."""
|
|
validated_sha = _validate_commit_sha(commit_sha)
|
|
validated_repo = _validate_repo_id(repo_id)
|
|
validated_url = _validate_gitea_url(gitea_url)
|
|
|
|
url = (
|
|
f"{validated_url}/api/v1/repos/"
|
|
f"{validated_repo}/archive/{validated_sha}.tar.gz"
|
|
)
|
|
|
|
async with httpx.AsyncClient(
|
|
timeout=httpx.Timeout(connect=15.0, read=60.0, write=15.0, pool=15.0),
|
|
follow_redirects=False,
|
|
) as client:
|
|
async with client.stream(
|
|
"GET",
|
|
url,
|
|
headers=_gitea_headers(),
|
|
) as response:
|
|
response.raise_for_status()
|
|
|
|
content_length = response.headers.get("content-length")
|
|
if (
|
|
content_length is not None
|
|
and int(content_length) > MAX_SOURCE_ARCHIVE_BYTES
|
|
):
|
|
raise ValueError("Source archive exceeds allowed size")
|
|
|
|
chunks = []
|
|
total_bytes = 0
|
|
|
|
async for chunk in response.aiter_bytes():
|
|
total_bytes += len(chunk)
|
|
|
|
if total_bytes > MAX_SOURCE_ARCHIVE_BYTES:
|
|
raise ValueError("Source archive exceeds allowed size")
|
|
|
|
chunks.append(chunk)
|
|
|
|
return b"".join(chunks)
|
|
|
|
|
|
def _normalize_repository_item(item: dict) -> dict | None:
|
|
"""Safely extracts and transforms a single repository item from the Gitea API response."""
|
|
if not isinstance(item, dict):
|
|
return None
|
|
|
|
full_name = item.get("full_name")
|
|
name = item.get("name")
|
|
|
|
if not all(isinstance(val, str) and val for val in [full_name, name]):
|
|
return None
|
|
|
|
if not full_name.startswith(f"{ALLOWED_GITEA_NAMESPACE}/"):
|
|
return None
|
|
|
|
return {
|
|
"name": name,
|
|
"full_name": full_name,
|
|
"default_branch": item.get("default_branch"),
|
|
"updated_at": item.get("updated_at"),
|
|
"archived": item.get("archived", False),
|
|
}
|
|
|
|
|
|
async def list_allowed_namespace_repositories() -> dict:
|
|
"""
|
|
Fetches a read-only, metadata-only catalog of Gitea repositories
|
|
from a fixed, approved namespace.
|
|
"""
|
|
gitea_url = _validate_gitea_url(os.environ.get("GITEA_URL"))
|
|
|
|
all_repos = []
|
|
seen_repos = set()
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15.0, follow_redirects=False) as client:
|
|
for page in range(1, CATALOG_MAX_PAGES + 1):
|
|
if len(all_repos) >= CATALOG_MAX_RESULTS:
|
|
break
|
|
|
|
url = (
|
|
f"{gitea_url}/api/v1/users/{ALLOWED_GITEA_NAMESPACE}/repos"
|
|
f"?limit={CATALOG_PAGE_SIZE}&page={page}"
|
|
)
|
|
|
|
response = await client.get(url, headers=_gitea_headers())
|
|
response.raise_for_status()
|
|
|
|
page_items = response.json()
|
|
if not isinstance(page_items, list) or not page_items:
|
|
break
|
|
|
|
for item in page_items:
|
|
normalized = _normalize_repository_item(item)
|
|
if normalized and normalized["full_name"] not in seen_repos:
|
|
seen_repos.add(normalized["full_name"])
|
|
all_repos.append(normalized)
|
|
|
|
except (httpx.HTTPError, json.JSONDecodeError) as e:
|
|
logger.error(f"Gitea repository catalog failed: {e}")
|
|
raise ValueError("Repository catalog unavailable.")
|
|
|
|
return {"repositories": all_repos[:CATALOG_MAX_RESULTS]}
|
|
|
|
|
|
async def handle_get_file_content(p: dict, server_repo_id: str) -> dict:
|
|
"""
|
|
Gets the raw content of a file from the Gitea repository after strict validation.
|
|
Uses the JSON/base64 Contents API with bounded reads.
|
|
"""
|
|
gitea_url = _validate_gitea_url(os.environ.get("GITEA_URL"))
|
|
validated_server_repo = _validate_repo_id(server_repo_id)
|
|
|
|
caller_repo = p.get("repo")
|
|
if caller_repo is not None and caller_repo != validated_server_repo:
|
|
raise ValueError("Repository file request is not allowed.")
|
|
|
|
ref = _validate_commit_sha(p.get("ref"))
|
|
path = _validate_safe_path(p.get("path"))
|
|
|
|
url = f"{gitea_url}/api/v1/repos/{validated_server_repo}/contents/{quote(path, safe='')}?ref={ref}"
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
async with c.stream("GET", url, headers=_gitea_headers()) as response:
|
|
response.raise_for_status()
|
|
|
|
content_length = -1
|
|
content_length_str = response.headers.get("content-length")
|
|
if content_length_str:
|
|
try:
|
|
content_length = int(content_length_str)
|
|
except (ValueError, TypeError):
|
|
content_length = -1
|
|
|
|
if content_length >= 0 and content_length > MAX_GET_FILE_RESPONSE_BYTES:
|
|
raise ValueError("Repository file exceeds the allowed size.")
|
|
|
|
body_bytes = bytearray()
|
|
async for chunk in response.aiter_bytes():
|
|
if len(body_bytes) + len(chunk) > MAX_GET_FILE_RESPONSE_BYTES:
|
|
raise ValueError("Repository file exceeds the allowed size.")
|
|
body_bytes.extend(chunk)
|
|
|
|
data = json.loads(body_bytes)
|
|
|
|
except (httpx.HTTPError, json.JSONDecodeError):
|
|
raise ValueError("Repository file is unavailable.")
|
|
|
|
if not isinstance(data, dict) or "content" not in data:
|
|
raise ValueError("Repository file is unavailable.")
|
|
|
|
size = data.get("size")
|
|
if isinstance(size, int) and not isinstance(size, bool) and size >= 0:
|
|
if size > MAX_READ_FILE_BYTES:
|
|
raise ValueError("Repository file exceeds the allowed size.")
|
|
|
|
try:
|
|
decoded_content = base64.b64decode(data["content"], validate=True)
|
|
except (TypeError, ValueError, binascii.Error):
|
|
raise ValueError("Repository file content is not readable text.")
|
|
|
|
if len(decoded_content) > MAX_READ_FILE_BYTES:
|
|
raise ValueError("Repository file exceeds the allowed size.")
|
|
|
|
try:
|
|
text_content = decoded_content.decode('utf-8')
|
|
except UnicodeDecodeError:
|
|
raise ValueError("Repository file content is not readable text.")
|
|
|
|
if '\0' in text_content:
|
|
raise ValueError("Repository file content is not readable text.")
|
|
|
|
return {"path": path, "content": text_content, "encoding": "utf-8"}
|
|
|
|
|
|
async def handle_list_repo_files(p: dict, default_repo: str) -> dict:
|
|
"""Lists files and directories in a given path in the Gitea repository."""
|
|
gitea_url = os.environ.get("GITEA_URL")
|
|
if not gitea_url:
|
|
raise ValueError("GITEA_URL environment variable is not set.")
|
|
|
|
if not isinstance(p, dict):
|
|
raise ValueError("Invalid list_repo_files input.")
|
|
|
|
if set(p.keys()) - {"path"}:
|
|
raise ValueError("Unsupported list_repo_files input field.")
|
|
|
|
path = p.get("path")
|
|
|
|
if path is None or path == "" or path == ".":
|
|
normalized_path = ""
|
|
else:
|
|
if not isinstance(path, str):
|
|
raise ValueError("Invalid repository path.")
|
|
normalized_path = path
|
|
|
|
if normalized_path.startswith('/') or \
|
|
'\\' in normalized_path or \
|
|
any(c in normalized_path for c in (':', '?', '#')):
|
|
raise ValueError("Invalid repository path.")
|
|
|
|
parts = normalized_path.split('/')
|
|
if any(part in ('.', '..') for part in parts):
|
|
raise ValueError("Invalid repository path.")
|
|
if '' in parts and normalized_path != '': # check for empty segments
|
|
raise ValueError("Invalid repository path.")
|
|
|
|
repo_id = default_repo
|
|
ref = "main"
|
|
|
|
safe_path = quote(normalized_path, safe='/')
|
|
|
|
url = f"{gitea_url.rstrip('/')}/api/v1/repos/{repo_id}/contents/{safe_path}"
|
|
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
r = await c.get(url, headers=_gitea_headers(), params={"ref": ref})
|
|
r.raise_for_status()
|
|
|
|
response_data = r.json()
|
|
|
|
if isinstance(response_data, dict):
|
|
raise ValueError("Path is a file, not a directory. Use get_file_content instead.")
|
|
|
|
if not isinstance(response_data, list):
|
|
raise ValueError("Invalid upstream response.")
|
|
|
|
result_files = []
|
|
for entry in response_data:
|
|
if not isinstance(entry, dict):
|
|
raise ValueError("Invalid upstream response.")
|
|
|
|
name = entry.get("name")
|
|
entry_path = entry.get("path")
|
|
entry_type = entry.get("type")
|
|
|
|
if not all([name, entry_path, entry_type]):
|
|
raise ValueError("Invalid upstream response.")
|
|
|
|
result_entry = {
|
|
"name": name,
|
|
"path": entry_path,
|
|
"type": entry_type
|
|
}
|
|
|
|
if "sha" in entry and entry["sha"] is not None:
|
|
result_entry["sha"] = entry["sha"]
|
|
|
|
if "size" in entry and entry["size"] is not None:
|
|
result_entry["size"] = entry["size"]
|
|
|
|
result_files.append(result_entry)
|
|
|
|
return {
|
|
"path": normalized_path,
|
|
"files": result_files
|
|
}
|