161 lines
6.0 KiB
Python
161 lines
6.0 KiB
Python
"""Pure validation policy for Hermes' least-authority Atlas Forgejo client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import subprocess
|
|
import urllib.parse
|
|
|
|
REPO_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,99}\Z")
|
|
SHA_RE = re.compile(r"[0-9a-fA-F]{40}\Z")
|
|
DRAFT_TITLE_PREFIX = "WIP: "
|
|
GIT_BIN = "/usr/bin/git"
|
|
MAX_QUERY_LENGTH = 128
|
|
MAX_QUERY_ITEMS = 3
|
|
MAX_QUERY_KEY_LENGTH = 16
|
|
MAX_QUERY_VALUE_LENGTH = 16
|
|
MAX_PAGE = 10_000
|
|
MAX_LIMIT = 50
|
|
CANONICAL_QUERY_RE = re.compile(r"[A-Za-z0-9_=&-]*\Z")
|
|
SENSITIVE_TEXT_RE = re.compile(
|
|
r"""(?ix)
|
|
(?:
|
|
(?<![A-Za-z0-9])
|
|
(?:client[_-]?secret|access[_-]?token|refresh[_-]?token|private[_-]?key|
|
|
api[_-]?key|password|passwd|secret|token|authorization|bearer|basic)
|
|
[\"']?\s*[:=]\s*[\"']?
|
|
|
|
|
\b(?:bearer|basic)\s+[A-Za-z0-9+/_=.:-]{8,}
|
|
|
|
|
-----BEGIN\s+[A-Z0-9 ][A-Z0-9 -]{1,62}-----
|
|
|
|
|
\bssh-(?:rsa|ed25519|dss|ecdsa-[A-Za-z0-9-]+)\s+[A-Za-z0-9+/]{16,}={0,3}
|
|
|
|
|
\b(?:gh[pousr]_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{20,}|
|
|
glpat-[A-Za-z0-9_-]{20,}|xox[baprs]-[A-Za-z0-9-]{20,}|
|
|
sk-ant-[A-Za-z0-9_-]{20,}|sk-(?:proj-|live_)?[A-Za-z0-9_-]{20,}|
|
|
[sr]k_live_[A-Za-z0-9]{20,}|
|
|
ya29\.[A-Za-z0-9_-]{20,})\b
|
|
|
|
|
\b(?:AKIA|ASIA)[0-9A-Z]{16}\b
|
|
|
|
|
\bAIza[0-9A-Za-z_-]{35}\b
|
|
|
|
|
\beyJ[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{8,}\b
|
|
)
|
|
"""
|
|
)
|
|
|
|
|
|
class PolicyError(ValueError):
|
|
"""Raised when a requested Forgejo operation exceeds the safe boundary."""
|
|
|
|
|
|
def _validate_repo(repo: str) -> str:
|
|
if not REPO_RE.fullmatch(repo) or repo in {".", ".."}:
|
|
raise PolicyError("repository name is outside the Atlas allowlist")
|
|
return repo
|
|
|
|
|
|
def _validate_ref(value: object, name: str) -> str:
|
|
"""Validate the complete Git ref grammar using Git itself."""
|
|
if not isinstance(value, str) or not value or value.startswith("-"):
|
|
raise PolicyError(f"{name} must be a same-repository branch name")
|
|
if any(ord(character) < 32 or ord(character) == 127 for character in value):
|
|
raise PolicyError(f"{name} must be a safe same-repository branch name")
|
|
result = subprocess.run(
|
|
[GIT_BIN, "check-ref-format", f"refs/heads/{value}"],
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0:
|
|
raise PolicyError(f"{name} must be a safe same-repository branch name")
|
|
return value
|
|
|
|
|
|
def _validate_sha(value: object, name: str = "head SHA") -> str:
|
|
if not isinstance(value, str) or not SHA_RE.fullmatch(value):
|
|
raise PolicyError(f"{name} must be a full Git SHA-1")
|
|
return value.lower()
|
|
|
|
|
|
def _validate_text(value: object, name: str, maximum: int, *, required: bool) -> str:
|
|
if not isinstance(value, str):
|
|
raise PolicyError(f"{name} must be text")
|
|
if required and not value.strip():
|
|
raise PolicyError(f"{name} must not be empty")
|
|
if len(value) > maximum or "\x00" in value:
|
|
raise PolicyError(f"{name} exceeds the safe request limit")
|
|
return value
|
|
|
|
|
|
def _reject_sensitive(value: str, name: str, forbidden: tuple[str, ...]) -> None:
|
|
if any(secret and secret in value for secret in forbidden):
|
|
raise PolicyError(f"pull-request {name} contains runtime credential material")
|
|
if SENSITIVE_TEXT_RE.search(value):
|
|
raise PolicyError(f"pull-request {name} resembles credential material")
|
|
|
|
|
|
def _validate_body(value: object, *, forbidden: tuple[str, ...] = ()) -> str:
|
|
body = _validate_text(value, "body", 16384, required=False)
|
|
_reject_sensitive(body, "body", forbidden)
|
|
return body
|
|
|
|
|
|
def _draft_title(value: object, *, forbidden: tuple[str, ...] = ()) -> str:
|
|
"""Return a bounded, secret-screened Gitea draft title."""
|
|
title = _validate_text(value, "title", 251, required=True).strip()
|
|
_reject_sensitive(title, "title", forbidden)
|
|
for prefix in ("WIP:", "[WIP]"):
|
|
if title.upper().startswith(prefix):
|
|
title = title[len(prefix) :].lstrip()
|
|
break
|
|
if not title:
|
|
raise PolicyError("title must contain text after the draft prefix")
|
|
return DRAFT_TITLE_PREFIX + title
|
|
|
|
|
|
def _validate_query(target: urllib.parse.SplitResult, allowed: set[str]) -> None:
|
|
"""Accept only short canonical ASCII query strings with bounded pagination."""
|
|
raw = target.query
|
|
if (
|
|
len(raw) > MAX_QUERY_LENGTH
|
|
or not raw.isascii()
|
|
or not CANONICAL_QUERY_RE.fullmatch(raw)
|
|
or "%" in raw
|
|
):
|
|
raise PolicyError("API query must use short canonical ASCII form")
|
|
try:
|
|
pairs = urllib.parse.parse_qsl(raw, keep_blank_values=True, strict_parsing=True)
|
|
except ValueError as exc:
|
|
raise PolicyError("invalid API query") from exc
|
|
if len(pairs) > MAX_QUERY_ITEMS:
|
|
raise PolicyError("too many API query parameters")
|
|
if len({key for key, _ in pairs}) != len(pairs):
|
|
raise PolicyError("duplicate API query parameters are not allowed")
|
|
if any(
|
|
not key
|
|
or len(key) > MAX_QUERY_KEY_LENGTH
|
|
or len(value) > MAX_QUERY_VALUE_LENGTH
|
|
for key, value in pairs
|
|
):
|
|
raise PolicyError("API query key or value exceeds its safe limit")
|
|
if any(key not in allowed for key, _ in pairs):
|
|
raise PolicyError("API query parameter is outside the read allowlist")
|
|
values = dict(pairs)
|
|
for name in ("page", "limit"):
|
|
if name not in values:
|
|
continue
|
|
if not re.fullmatch(r"[0-9]+", values[name]):
|
|
raise PolicyError(f"{name} must use ASCII decimal digits")
|
|
number = int(values[name])
|
|
if values[name] != str(number):
|
|
raise PolicyError(f"{name} must use canonical ASCII decimal form")
|
|
ceiling = MAX_PAGE if name == "page" else MAX_LIMIT
|
|
if not 1 <= number <= ceiling:
|
|
raise PolicyError(f"{name} is outside its allowed range")
|
|
if "state" in values and values["state"] not in {"open", "closed", "all"}:
|
|
raise PolicyError("pull-request state is invalid")
|