418 lines
14 KiB
Python
418 lines
14 KiB
Python
"""Validation policy for Hermes' least-authority Atlas Forgejo client.
|
|
|
|
The screening catches structured credential assignments, known token formats,
|
|
and long high-entropy values. It is a fail-closed accident barrier, not proof
|
|
that text is secret-free: a short or multiword secret under an innocuous key is
|
|
not reliably distinguishable from prose and must never be supplied by callers.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import subprocess
|
|
import urllib.parse
|
|
from collections import Counter
|
|
from math import log2
|
|
|
|
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")
|
|
ASSIGNMENT_RE = re.compile(
|
|
r"""(?mx)
|
|
(?<![A-Za-z0-9_.-])
|
|
(?P<key_quote>[\"']?)
|
|
(?P<key>\.?[A-Za-z][A-Za-z0-9_.-]{0,127})
|
|
(?P=key_quote)
|
|
[ \t]*(?P<separator>[:=])[ \t]*
|
|
(?P<value>
|
|
\"(?:\\.|[^\"\\\r\n])*\" |
|
|
'(?:\\.|[^'\\\r\n])*' |
|
|
[^\r\n,;}&]*
|
|
)
|
|
"""
|
|
)
|
|
CAMEL_BOUNDARY_RE = re.compile(r"(?<=[a-z0-9])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])")
|
|
CREDENTIAL_SCHEME_RE = re.compile(
|
|
r"(?i)\b(?:bearer|basic|token)\s+"
|
|
r"(?=[A-Za-z0-9+/_=.:-]{8,}\b)(?=[A-Za-z0-9+/_=.:-]*[0-9+/_=.-])"
|
|
r"[A-Za-z0-9+/_=.:-]{8,}"
|
|
)
|
|
STANDALONE_SECRET_PATTERNS = tuple(
|
|
re.compile(pattern, re.IGNORECASE)
|
|
for pattern in (
|
|
r"\bgh[pousr]_[A-Za-z0-9]{20,}\b",
|
|
r"\bgithub_pat_[A-Za-z0-9_]{20,}\b",
|
|
r"\bglpat-[A-Za-z0-9_-]{20,}\b",
|
|
r"\bxox[baprs]-[A-Za-z0-9-]{20,}\b",
|
|
r"\bsk-(?:ant-|proj-)?[A-Za-z0-9_-]{20,}\b",
|
|
r"\b[rs]k_(?:test|live)_[A-Za-z0-9]{20,}\b",
|
|
r"\bwhsec_[A-Za-z0-9]{20,}\b",
|
|
r"\bnpm_[A-Za-z0-9]{20,}\b",
|
|
r"\bpypi-[A-Za-z0-9_-]{30,}\b",
|
|
r"\bhf_[A-Za-z0-9]{20,}\b",
|
|
r"\b(?:gta|gto|gitea|forgejo)_[A-Za-z0-9_-]{20,}\b",
|
|
r"\bya29\.[A-Za-z0-9_-]{20,}\b",
|
|
r"\boy2[A-Za-z0-9]{40,}\b",
|
|
r"\bSG\.[A-Za-z0-9_-]{16,}\.[A-Za-z0-9_-]{20,}\b",
|
|
r"\bSK[0-9a-f]{32}\b",
|
|
r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b",
|
|
r"\bAIza[0-9A-Za-z_-]{35}\b",
|
|
r"\beyJ[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{8,}\b",
|
|
r"-----BEGIN\s+[A-Z0-9 ][A-Z0-9 -]{1,62}-----",
|
|
r"\bssh-(?:rsa|ed25519|dss|ecdsa-[A-Za-z0-9-]+)\s+[A-Za-z0-9+/]{16,}={0,3}",
|
|
r"https://hooks\.slack\.com/services/[A-Za-z0-9/_-]{20,}",
|
|
r"https://(?:discord(?:app)?\.com)/api/webhooks/[0-9]+/[A-Za-z0-9._-]{20,}",
|
|
r"https://[^\s/]*webhook\.office\.com/[^\s]{20,}",
|
|
)
|
|
)
|
|
HIGH_ENTROPY_TOKEN_RE = re.compile(
|
|
r"(?<![A-Za-z0-9+/_=-])[A-Za-z0-9+/_=-]{48,}(?![A-Za-z0-9+/_=-])"
|
|
)
|
|
PROSE_LEAD_WORDS = {
|
|
"add",
|
|
"allow",
|
|
"change",
|
|
"check",
|
|
"describe",
|
|
"document",
|
|
"ensure",
|
|
"explain",
|
|
"fix",
|
|
"handle",
|
|
"keep",
|
|
"preserve",
|
|
"prevent",
|
|
"reject",
|
|
"remove",
|
|
"review",
|
|
"support",
|
|
"test",
|
|
"validate",
|
|
}
|
|
SENSITIVE_KEY_WORDS = {
|
|
"auth",
|
|
"authentication",
|
|
"authorization",
|
|
"auths",
|
|
"basic",
|
|
"bearer",
|
|
"credential",
|
|
"credentials",
|
|
"passwd",
|
|
"password",
|
|
"pat",
|
|
"private",
|
|
"sas",
|
|
"secret",
|
|
"secrets",
|
|
"sig",
|
|
"signature",
|
|
"signing",
|
|
"token",
|
|
"tokens",
|
|
"webhook",
|
|
}
|
|
KEY_MODIFIER_WORDS = {
|
|
"access",
|
|
"account",
|
|
"api",
|
|
"auth",
|
|
"client",
|
|
"encryption",
|
|
"identity",
|
|
"private",
|
|
"registry",
|
|
"secret",
|
|
"service",
|
|
"session",
|
|
"signing",
|
|
"ssh",
|
|
}
|
|
SENSITIVE_COMPACT_KEYS = {
|
|
"accessid",
|
|
"accesskeyid",
|
|
"accountkey",
|
|
"clientemail",
|
|
"clientid",
|
|
"connectionstring",
|
|
"dockerconfigjson",
|
|
"privatekeyid",
|
|
"serviceaccountkey",
|
|
"sharedaccesssignature",
|
|
}
|
|
SENSITIVE_COMPACT_SUFFIXES = (
|
|
"accesskey",
|
|
"accesskeyid",
|
|
"accountkey",
|
|
"apikey",
|
|
"authkey",
|
|
"clientemail",
|
|
"clientid",
|
|
"clientkey",
|
|
"connectionstring",
|
|
"credential",
|
|
"credentials",
|
|
"encryptionkey",
|
|
"identitykey",
|
|
"password",
|
|
"passwd",
|
|
"privatekeyid",
|
|
"privatekey",
|
|
"secret",
|
|
"secretkey",
|
|
"servicekey",
|
|
"sessionkey",
|
|
"signature",
|
|
"signingkey",
|
|
"sshkey",
|
|
"token",
|
|
"webhook",
|
|
)
|
|
|
|
|
|
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 _normalized_key(value: str) -> tuple[tuple[str, ...], str]:
|
|
"""Split camelCase and separator-based assignment keys into semantics."""
|
|
separated = CAMEL_BOUNDARY_RE.sub(" ", value.strip(".\"'"))
|
|
words = tuple(re.findall(r"[A-Za-z0-9]+", separated.lower()))
|
|
return words, "".join(words)
|
|
|
|
|
|
def _is_sensitive_key(value: str) -> bool:
|
|
words, compact = _normalized_key(value)
|
|
word_set = set(words)
|
|
if word_set & SENSITIVE_KEY_WORDS:
|
|
return True
|
|
if compact in SENSITIVE_COMPACT_KEYS:
|
|
return True
|
|
if compact.endswith(SENSITIVE_COMPACT_SUFFIXES):
|
|
return True
|
|
if "key" in word_set and word_set & KEY_MODIFIER_WORDS:
|
|
return True
|
|
if {"client", "email"} <= word_set or {"client", "id"} <= word_set:
|
|
return True
|
|
if {"access", "id"} <= word_set or {"connection", "string"} <= word_set:
|
|
return True
|
|
return compact.endswith("configjson")
|
|
|
|
|
|
def _strip_assignment_value(value: str) -> tuple[str, bool]:
|
|
stripped = value.strip()
|
|
quoted = (
|
|
len(stripped) >= 2 and stripped[0] in {'"', "'"} and stripped[-1] == stripped[0]
|
|
)
|
|
if quoted:
|
|
stripped = stripped[1:-1].strip()
|
|
return stripped, quoted
|
|
|
|
|
|
def _looks_like_prose(value: str) -> bool:
|
|
words = re.findall(r"[A-Za-z]+(?:[-'][A-Za-z]+)?", value)
|
|
if len(words) < 2:
|
|
return False
|
|
if re.search(r"[$`{}\[\]\\@/:=+_]", value):
|
|
return False
|
|
return words[0].lower() in PROSE_LEAD_WORDS or len(words) >= 4
|
|
|
|
|
|
def _looks_sensitive_assignment_value(value: str) -> bool:
|
|
stripped, quoted = _strip_assignment_value(value)
|
|
if not stripped:
|
|
return False
|
|
if quoted or stripped.startswith(("{", "[", "|", ">", "!!", "&", "*")):
|
|
return True
|
|
if CREDENTIAL_SCHEME_RE.search(stripped):
|
|
return True
|
|
if any(pattern.search(stripped) for pattern in STANDALONE_SECRET_PATTERNS):
|
|
return True
|
|
if re.search(r"(?:https?://|[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+)", stripped):
|
|
return True
|
|
if re.search(r"(?:\$\{|\$\(|\\[nrt]|[A-Za-z0-9+/]{16,}={0,3}\Z)", stripped):
|
|
return True
|
|
return not _looks_like_prose(stripped)
|
|
|
|
|
|
def _is_structural_credential_assignment(key: str, value: str) -> bool:
|
|
words, compact_key = _normalized_key(key)
|
|
stripped, _quoted = _strip_assignment_value(value)
|
|
_value_words, compact_value = _normalized_key(stripped)
|
|
return (
|
|
compact_key == "type" and compact_value in {"serviceaccount", "credential"}
|
|
) or ("service" in words and "account" in words and bool(stripped))
|
|
|
|
|
|
def _has_high_entropy_token(value: str) -> bool:
|
|
for match in HIGH_ENTROPY_TOKEN_RE.finditer(value):
|
|
token = match.group(0)
|
|
if len(token) > 256:
|
|
return True
|
|
groups = sum(
|
|
bool(re.search(pattern, token))
|
|
for pattern in (r"[a-z]", r"[A-Z]", r"[0-9]", r"[+/_=-]")
|
|
)
|
|
counts = Counter(token)
|
|
entropy = -sum(
|
|
(count / len(token)) * log2(count / len(token)) for count in counts.values()
|
|
)
|
|
if groups >= 3 and entropy >= 4.0:
|
|
return True
|
|
return False
|
|
|
|
|
|
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 CREDENTIAL_SCHEME_RE.search(value):
|
|
raise PolicyError(f"pull-request {name} resembles credential material")
|
|
if any(pattern.search(value) for pattern in STANDALONE_SECRET_PATTERNS):
|
|
raise PolicyError(f"pull-request {name} resembles credential material")
|
|
for candidate in ASSIGNMENT_RE.finditer(value):
|
|
key = candidate.group("key")
|
|
assigned = candidate.group("value")
|
|
if _is_structural_credential_assignment(key, assigned) or (
|
|
_is_sensitive_key(key) and _looks_sensitive_assignment_value(assigned)
|
|
):
|
|
raise PolicyError(f"pull-request {name} resembles credential material")
|
|
if _has_high_entropy_token(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")
|