feat: add Gitea request executor with result envelope, pagination and redaction
This commit is contained in:
Latte committed
1 parent
041b3ea7ff
commit
9e50859133
8 files changed
+3021
No files matched your search
@@ -0,0 +1,355 @@
|
||||
"""Request building for the Gitea executor: paths, query strings and bodies.
|
||||
|
||||
Everything here is pure (no I/O) and raises :class:`ArgumentError` with a
|
||||
message that never contains the offending value, because a rejected value may
|
||||
itself be a credential a model pasted into the wrong argument.
|
||||
|
||||
Path safety rules
|
||||
-----------------
|
||||
Every value substituted into a path template (and every raw ``request()``
|
||||
path) is rejected when it is empty, contains a ``..`` or ``.`` path segment, an
|
||||
empty segment, a backslash, ``?``, ``#``, ``%`` or a control character. After
|
||||
that check the value is percent-encoded. Most parameters are quoted with
|
||||
``safe=""`` so a ``/`` becomes ``%2F`` and can never open a new path segment.
|
||||
The parameters in :data:`SLASH_PATH_PARAMS` legitimately hold slashes (file
|
||||
paths, branch, tag and ref names) and are routed by Gitea through wildcard
|
||||
routes, so they are quoted with ``safe="/"``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import binascii
|
||||
import re
|
||||
import unicodedata
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
from urllib.parse import quote
|
||||
|
||||
from aegis_gitea_mcp.catalog.model import Operation, Parameter
|
||||
|
||||
#: Path parameters that may contain ``/``. Gitea serves these routes with
|
||||
#: wildcard patterns (``contents/*``, ``branches/*``, ``compare/*``,
|
||||
#: ``git/refs/*``, ``tags/*``, ``pulls/{base}/*``), so refs such as
|
||||
#: ``feature/login`` and file paths such as ``src/app/main.py`` are sent with
|
||||
#: real slashes. Every other parameter is quoted with ``safe=""``.
|
||||
SLASH_PATH_PARAMS = frozenset({"filepath", "branch", "ref", "basehead", "head", "tag"})
|
||||
|
||||
#: Query parameter names that carry credentials or switch the acting user in
|
||||
#: Gitea. They are never forwarded: the identity and the ``sudo`` argument are
|
||||
#: the only sanctioned ways to set them.
|
||||
FORBIDDEN_QUERY_NAMES = frozenset({"token", "access_token", "sudo", "_sudo"})
|
||||
|
||||
#: Raw-request methods.
|
||||
RAW_METHODS = frozenset({"GET", "POST", "PUT", "PATCH", "DELETE"})
|
||||
|
||||
_PLACEHOLDER = re.compile(r"\{([^{}]+)\}")
|
||||
_BAD_CHARS = frozenset("\\?#%")
|
||||
_BAD_CATEGORIES = frozenset({"Cc", "Cs", "Zl", "Zp"})
|
||||
_SUDO_PATTERN = re.compile(r"^[A-Za-z0-9_.-]{1,255}$")
|
||||
_MAX_FILENAME_CHARS = 255
|
||||
_MAX_CONTENT_TYPE_CHARS = 255
|
||||
_BODY_DEPTH_LIMIT = 8
|
||||
|
||||
#: Body keys whose string values are treated as secrets for redaction.
|
||||
_SENSITIVE_BODY_KEYS = frozenset(
|
||||
{
|
||||
"password",
|
||||
"secret",
|
||||
"client_secret",
|
||||
"token",
|
||||
"access_token",
|
||||
"auth_password",
|
||||
"auth_token",
|
||||
"private_key",
|
||||
}
|
||||
)
|
||||
_MIN_SENSITIVE_VALUE_CHARS = 4
|
||||
|
||||
|
||||
class ArgumentError(ValueError):
|
||||
"""An argument cannot be turned into a safe request (message never echoes the value)."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RequestBody:
|
||||
"""Body of an outgoing request; at most one of the fields is set.
|
||||
|
||||
Attributes:
|
||||
json: Value to send as ``application/json`` (``has_json`` tells ``None`` apart).
|
||||
has_json: True when ``json`` is the body.
|
||||
content: Raw bytes for ``text/plain`` bodies.
|
||||
content_type: Content type for ``content`` bodies.
|
||||
files: Multipart parts as ``(field, (filename|None, bytes, content_type|None))``.
|
||||
"""
|
||||
|
||||
json: Any = None
|
||||
has_json: bool = False
|
||||
content: bytes | None = None
|
||||
content_type: str | None = None
|
||||
files: list[tuple[str, tuple[str | None, bytes, str | None]]] = field(default_factory=list)
|
||||
|
||||
|
||||
def _has_bad_character(value: str) -> bool:
|
||||
return any(
|
||||
char in _BAD_CHARS or unicodedata.category(char) in _BAD_CATEGORIES for char in value
|
||||
)
|
||||
|
||||
|
||||
def check_segments(value: str, label: str) -> None:
|
||||
"""Reject values that could escape or reshape a URL path.
|
||||
|
||||
Args:
|
||||
value: Candidate path value (a parameter value or a whole raw path).
|
||||
label: Name used in the error message.
|
||||
|
||||
Raises:
|
||||
ArgumentError: The value is empty or unsafe.
|
||||
"""
|
||||
if not value:
|
||||
raise ArgumentError(f"{label} must not be empty")
|
||||
if _has_bad_character(value):
|
||||
raise ArgumentError(
|
||||
f"{label} contains a forbidden character (backslash, ?, #, % or control character)"
|
||||
)
|
||||
for segment in value.split("/"):
|
||||
if segment in {"", ".", ".."}:
|
||||
raise ArgumentError(f"{label} contains an empty, '.' or '..' path segment")
|
||||
|
||||
|
||||
def _path_value(parameter: Parameter, arguments: Mapping[str, Any]) -> str:
|
||||
value = arguments.get(parameter.arg_name)
|
||||
if value is None:
|
||||
raise ArgumentError(f"missing path parameter '{parameter.arg_name}'")
|
||||
if isinstance(value, bool) or not isinstance(value, (str, int)):
|
||||
raise ArgumentError(f"path parameter '{parameter.arg_name}' must be a string or integer")
|
||||
return str(value)
|
||||
|
||||
|
||||
def render_path(operation: Operation, arguments: Mapping[str, Any]) -> tuple[str, str]:
|
||||
"""Fill the operation's path template.
|
||||
|
||||
Args:
|
||||
operation: Catalog operation.
|
||||
arguments: Tool arguments keyed by ``arg_name``.
|
||||
|
||||
Returns:
|
||||
``(encoded_path, display_path)``: the percent-encoded path to send and
|
||||
the readable path (raw values) for the result envelope.
|
||||
|
||||
Raises:
|
||||
ArgumentError: A placeholder has no parameter, is missing, or is unsafe.
|
||||
"""
|
||||
by_name = {p.name: p for p in operation.parameters if p.location == "path"}
|
||||
display: dict[str, str] = {}
|
||||
|
||||
def fill(match: re.Match[str]) -> str:
|
||||
name = match.group(1)
|
||||
parameter = by_name.get(name)
|
||||
if parameter is None:
|
||||
raise ArgumentError(f"path placeholder '{name}' has no matching parameter")
|
||||
value = _path_value(parameter, arguments)
|
||||
check_segments(value, f"path parameter '{parameter.arg_name}'")
|
||||
safe = "/" if name in SLASH_PATH_PARAMS else ""
|
||||
display[name] = value
|
||||
return quote(value, safe=safe)
|
||||
|
||||
encoded_path = _PLACEHOLDER.sub(fill, operation.path)
|
||||
display_path = _PLACEHOLDER.sub(lambda m: display[m.group(1)], operation.path)
|
||||
return encoded_path, display_path
|
||||
|
||||
|
||||
def render_raw_path(path: str) -> tuple[str, str]:
|
||||
"""Validate and encode a raw ``/api/v1``-relative path.
|
||||
|
||||
Returns:
|
||||
``(encoded_path, display_path)``.
|
||||
|
||||
Raises:
|
||||
ArgumentError: The path does not start with ``/`` or fails the segment checks.
|
||||
"""
|
||||
if not isinstance(path, str) or not path.startswith("/"):
|
||||
raise ArgumentError("path must start with '/' and be relative to /api/v1")
|
||||
check_segments(path[1:], "path")
|
||||
return quote(path, safe="/"), path
|
||||
|
||||
|
||||
def _scalar(value: Any, name: str) -> str:
|
||||
if isinstance(value, bool):
|
||||
return "true" if value else "false"
|
||||
if isinstance(value, (int, float)):
|
||||
return str(value)
|
||||
if isinstance(value, str):
|
||||
return value
|
||||
raise ArgumentError(f"query parameter '{name}' must be a string, number, boolean or list")
|
||||
|
||||
|
||||
def _query_pairs(name: str, value: Any, csv: bool) -> list[tuple[str, str]]:
|
||||
if name.lower() in FORBIDDEN_QUERY_NAMES:
|
||||
raise ArgumentError(f"query parameter '{name}' is reserved and cannot be set")
|
||||
if isinstance(value, (list, tuple)):
|
||||
items = [_scalar(item, name) for item in value]
|
||||
if not items:
|
||||
return []
|
||||
if csv:
|
||||
return [(name, ",".join(items))]
|
||||
return [(name, item) for item in items]
|
||||
return [(name, _scalar(value, name))]
|
||||
|
||||
|
||||
def build_query(operation: Operation, arguments: Mapping[str, Any]) -> list[tuple[str, str]]:
|
||||
"""Build the query string pairs from the operation's query parameters.
|
||||
|
||||
Only provided (non-null) arguments are sent. Booleans become ``true`` or
|
||||
``false``; arrays become repeated parameters unless the parameter schema
|
||||
declares ``collectionFormat``/``x-collection-format`` ``csv``.
|
||||
"""
|
||||
pairs: list[tuple[str, str]] = []
|
||||
for parameter in operation.parameters:
|
||||
if parameter.location != "query":
|
||||
continue
|
||||
value = arguments.get(parameter.arg_name)
|
||||
if value is None:
|
||||
continue
|
||||
schema = parameter.schema
|
||||
csv = "csv" in (schema.get("collectionFormat"), schema.get("x-collection-format"))
|
||||
pairs.extend(_query_pairs(parameter.name, value, csv))
|
||||
return pairs
|
||||
|
||||
|
||||
def build_raw_query(query: Mapping[str, Any] | None) -> list[tuple[str, str]]:
|
||||
"""Build query pairs for a raw request (repeated parameters for lists)."""
|
||||
pairs: list[tuple[str, str]] = []
|
||||
for name, value in (query or {}).items():
|
||||
if value is None:
|
||||
continue
|
||||
pairs.extend(_query_pairs(str(name), value, csv=False))
|
||||
return pairs
|
||||
|
||||
|
||||
def validate_sudo(sudo: Any) -> str | None:
|
||||
"""Validate the optional ``sudo`` login (None when not given)."""
|
||||
if sudo is None:
|
||||
return None
|
||||
if not isinstance(sudo, str) or not _SUDO_PATTERN.match(sudo):
|
||||
raise ArgumentError("sudo must be a Gitea username (letters, digits, '-', '_', '.')")
|
||||
return sudo
|
||||
|
||||
|
||||
def _single_line(value: str, label: str, limit: int) -> str:
|
||||
if not value or len(value) > limit or _has_control(value):
|
||||
raise ArgumentError(f"{label} must be a non-empty single-line string")
|
||||
return value
|
||||
|
||||
|
||||
def _has_control(value: str) -> bool:
|
||||
return any(unicodedata.category(char) in _BAD_CATEGORIES for char in value)
|
||||
|
||||
|
||||
def _file_part(
|
||||
parameter: Parameter, value: Any
|
||||
) -> tuple[str, tuple[str | None, bytes, str | None]]:
|
||||
label = f"file argument '{parameter.arg_name}'"
|
||||
if not isinstance(value, Mapping):
|
||||
raise ArgumentError(f"{label} must be an object with filename and content_base64")
|
||||
unknown = set(value) - {"filename", "content_base64", "content_type"}
|
||||
if unknown:
|
||||
raise ArgumentError(f"{label} has unexpected keys")
|
||||
filename = value.get("filename")
|
||||
encoded = value.get("content_base64")
|
||||
if not isinstance(filename, str) or not isinstance(encoded, str):
|
||||
raise ArgumentError(f"{label} needs string filename and content_base64")
|
||||
filename = _single_line(filename, f"{label} filename", _MAX_FILENAME_CHARS)
|
||||
if '"' in filename:
|
||||
raise ArgumentError(f"{label} filename must not contain a double quote")
|
||||
try:
|
||||
content = base64.b64decode(encoded, validate=True)
|
||||
except (binascii.Error, ValueError) as exc:
|
||||
raise ArgumentError(f"{label} content_base64 is not valid base64") from exc
|
||||
content_type = value.get("content_type")
|
||||
if content_type is not None:
|
||||
if not isinstance(content_type, str):
|
||||
raise ArgumentError(f"{label} content_type must be a string")
|
||||
content_type = _single_line(content_type, f"{label} content_type", _MAX_CONTENT_TYPE_CHARS)
|
||||
return parameter.name, (filename, content, content_type or "application/octet-stream")
|
||||
|
||||
|
||||
def build_body(operation: Operation, arguments: Mapping[str, Any]) -> RequestBody:
|
||||
"""Build the request body from the operation's body and formData parameters.
|
||||
|
||||
``formData`` parameters produce a multipart body (files are decoded from
|
||||
base64 strictly). A ``body`` parameter is sent as JSON, or as a plain string
|
||||
when the operation only consumes ``text/plain``.
|
||||
|
||||
Raises:
|
||||
ArgumentError: A file argument or text body is malformed.
|
||||
"""
|
||||
form_parts: list[tuple[str, tuple[str | None, bytes, str | None]]] = []
|
||||
for parameter in operation.parameters:
|
||||
if parameter.location != "formData":
|
||||
continue
|
||||
value = arguments.get(parameter.arg_name)
|
||||
if value is None:
|
||||
continue
|
||||
if parameter.is_file:
|
||||
form_parts.append(_file_part(parameter, value))
|
||||
else:
|
||||
form_parts.append(
|
||||
(parameter.name, (None, _scalar(value, parameter.arg_name).encode(), None))
|
||||
)
|
||||
if form_parts:
|
||||
return RequestBody(files=form_parts)
|
||||
|
||||
for parameter in operation.parameters:
|
||||
if parameter.location != "body":
|
||||
continue
|
||||
value = arguments.get(parameter.arg_name)
|
||||
if value is None:
|
||||
continue
|
||||
if "text/plain" in operation.consumes and "application/json" not in operation.consumes:
|
||||
if not isinstance(value, str):
|
||||
raise ArgumentError(f"body of '{parameter.arg_name}' must be a string")
|
||||
return RequestBody(content=value.encode("utf-8"), content_type="text/plain")
|
||||
return RequestBody(json=value, has_json=True)
|
||||
return RequestBody()
|
||||
|
||||
|
||||
def _walk_sensitive(value: Any, found: list[str], depth: int) -> None:
|
||||
if depth > _BODY_DEPTH_LIMIT:
|
||||
return
|
||||
if isinstance(value, Mapping):
|
||||
for key, item in value.items():
|
||||
if (
|
||||
isinstance(key, str)
|
||||
and key.lower() in _SENSITIVE_BODY_KEYS
|
||||
and isinstance(item, str)
|
||||
and len(item) >= _MIN_SENSITIVE_VALUE_CHARS
|
||||
):
|
||||
found.append(item)
|
||||
else:
|
||||
_walk_sensitive(item, found, depth + 1)
|
||||
elif isinstance(value, (list, tuple)):
|
||||
for item in value:
|
||||
_walk_sensitive(item, found, depth + 1)
|
||||
|
||||
|
||||
def sensitive_body_values(body: Any, *, secret_data: bool) -> list[str]:
|
||||
"""Collect submitted values that must never be echoed back in a result.
|
||||
|
||||
Args:
|
||||
body: JSON request body.
|
||||
secret_data: True for Action-secret writes, where ``body["data"]`` is the
|
||||
secret itself (redacted at any length).
|
||||
|
||||
Returns:
|
||||
Values to redact exactly: ``password``, ``secret``, ``token`` style
|
||||
fields of at least 4 characters, plus the secret ``data`` value.
|
||||
"""
|
||||
found: list[str] = []
|
||||
if secret_data and isinstance(body, Mapping):
|
||||
data = body.get("data")
|
||||
if isinstance(data, str) and data:
|
||||
found.append(data)
|
||||
_walk_sensitive(body, found, 0)
|
||||
return found
|
||||
Reference in new issue
Block a user