227 lines
8.7 KiB
Python
227 lines
8.7 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
|
|
MAX_PR_NUMBER = 2_147_483_647
|
|
MAX_PR_NUMBER_DIGITS = 10
|
|
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|
|
|
private[_-]?key[_-]?id|client[_-]?email|api[_-]?key|password|passwd|
|
|
secret|token|authorization|bearer|basic|personal[_-]?access[_-]?token|
|
|
pat|github[_-]?token|gitlab[_-]?token|gitea[_-]?token|forgejo[_-]?token|
|
|
aws[_-]?(?:access[_-]?key(?:[_-]?id)?|secret[_-]?(?:access[_-]?)?key|
|
|
session[_-]?token|security[_-]?token)|
|
|
access[_-]?key[_-]?id|secret[_-]?access[_-]?key|session[_-]?token|
|
|
account[_-]?key|shared[_-]?access[_-]?signature|sas[_-]?token|
|
|
azure[_-]?(?:storage[_-]?)?connection[_-]?string|connection[_-]?string|
|
|
docker[_-]?(?:auth[_-]?config|config[_-]?json)|dockerconfigjson|
|
|
registry[_-]?(?:auth|password|token)|identity[_-]?token|
|
|
google[_-]?(?:credentials|oauth[_-]?token)|service[_-]?account[_-]?key|
|
|
webhook[_-]?(?:url|token|secret)|slack[_-]?(?:token|webhook)|
|
|
twilio[_-]?auth[_-]?token|sendgrid[_-]?api[_-]?key)
|
|
[\"']?\s*[:=]\s*[\"']?
|
|
|
|
|
\b(?:bearer|basic)\s+
|
|
(?=[A-Za-z0-9+/_=.:-]{16,}\b)(?=[A-Za-z0-9+/_=.:-]*[0-9+/_=.-])
|
|
[A-Za-z0-9+/_=.:-]{16,}
|
|
|
|
|
[\"'](?:auth|identitytoken|registrytoken)[\"']\s*:\s*
|
|
[\"'][A-Za-z0-9+/_=.-]{8,}[\"']
|
|
|
|
|
[\"']auths[\"']\s*:\s*\{
|
|
|
|
|
[\"']type[\"']\s*:\s*[\"']service_account[\"']
|
|
|
|
|
\bDefaultEndpointsProtocol\s*=\s*https?;[^\r\n]{0,1024}
|
|
AccountKey\s*=
|
|
|
|
|
(?:[?;&]|\b)sv=20[0-9]{2}-[0-9]{2}-[0-9]{2}&[^\s]{0,1024}&sig=
|
|
|
|
|
-----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,}|
|
|
whsec_[A-Za-z0-9]{20,}|npm_[A-Za-z0-9]{20,}|
|
|
pypi-[A-Za-z0-9_-]{30,}|hf_[A-Za-z0-9]{20,}|
|
|
(?:gta|gto|gitea|forgejo)_[A-Za-z0-9_-]{20,}|
|
|
ya29\.[A-Za-z0-9_-]{20,})\b
|
|
|
|
|
\bSG\.[A-Za-z0-9_-]{16,}\.[A-Za-z0-9_-]{20,}\b
|
|
|
|
|
\bSK[0-9a-fA-F]{32}\b
|
|
|
|
|
https://hooks\.slack\.com/services/[A-Za-z0-9/_-]{20,}
|
|
|
|
|
https://(?:discord(?:app)?\.com)/api/webhooks/[0-9]+/[A-Za-z0-9._-]{20,}
|
|
|
|
|
https://[^\s/]*webhook\.office\.com/[^\s]{20,}
|
|
|
|
|
\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_pr_number_segment(value: object) -> int:
|
|
"""Validate a canonical bounded pull-request number from an API path."""
|
|
if (
|
|
not isinstance(value, str)
|
|
or len(value) > MAX_PR_NUMBER_DIGITS
|
|
or not re.fullmatch(r"[1-9][0-9]*", value)
|
|
):
|
|
raise PolicyError("pull-request number must use bounded canonical ASCII digits")
|
|
number = int(value)
|
|
if number > MAX_PR_NUMBER:
|
|
raise PolicyError("pull-request number is outside its allowed range")
|
|
return number
|
|
|
|
|
|
def _validate_pr_number(value: object) -> int:
|
|
"""Validate a bounded pull-request number returned by Forgejo."""
|
|
if (
|
|
not isinstance(value, int)
|
|
or isinstance(value, bool)
|
|
or not 1 <= value <= MAX_PR_NUMBER
|
|
):
|
|
raise PolicyError("Forgejo omitted a valid pull-request number")
|
|
return value
|
|
|
|
|
|
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")
|