249 lines
7.5 KiB
Python
249 lines
7.5 KiB
Python
import os
|
|
import httpx
|
|
import logging
|
|
from urllib.parse import quote
|
|
import re
|
|
|
|
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")
|
|
)
|
|
|
|
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
|
|
|
|
|
|
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)
|
|
|
|
async def handle_get_file_content(p: dict, default_repo: str) -> dict:
|
|
"""Gets the raw content of a file from the Gitea repository."""
|
|
gitea_url = os.environ.get("GITEA_URL")
|
|
if not gitea_url:
|
|
raise ValueError("GITEA_URL environment variable is not set.")
|
|
|
|
path = p.get("path")
|
|
if not path:
|
|
raise ValueError("Missing required parameter: 'path' for get_file_content")
|
|
|
|
repo_id = p.get("repo", default_repo)
|
|
ref = p.get("ref", "main")
|
|
|
|
url = f"{gitea_url}/api/v1/repos/{repo_id}/raw/{path}?ref={ref}"
|
|
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
r = await c.get(url, headers=_gitea_headers())
|
|
r.raise_for_status()
|
|
return {"path": path, "content": r.text, "encoding": "text"}
|
|
|
|
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
|
|
}
|