Compare commits
1 Commits
f416b8a47c
...
c8ccc8f249
| Author | SHA1 | Date | |
|---|---|---|---|
|
c8ccc8f249 |
@ -34,7 +34,6 @@ MAX_ERROR_BYTES = 8192
|
||||
MAX_RESPONSE_BYTES = 2 * 1024 * 1024
|
||||
MAX_API_TARGET_LENGTH = 768
|
||||
MAX_API_PATH_LENGTH = 512
|
||||
RAW_API_TARGET_RE = re.compile(r"[A-Za-z0-9/_.?&=-]+\Z")
|
||||
|
||||
|
||||
class RejectRedirectHandler(urllib.request.HTTPRedirectHandler):
|
||||
@ -62,7 +61,7 @@ def read_token(path: Path = DEFAULT_TOKEN_FILE) -> str:
|
||||
|
||||
def configured_base_url() -> str:
|
||||
"""Reject attempts to redirect credentials away from the canonical origin."""
|
||||
configured = os.environ.get("GITEA_BASE_URL", CANONICAL_BASE_URL)
|
||||
configured = os.environ.get("GITEA_BASE_URL", CANONICAL_BASE_URL).rstrip("/")
|
||||
if configured != CANONICAL_BASE_URL:
|
||||
raise PolicyError("Forgejo origin is fixed to the private Atlas SCM service")
|
||||
return configured
|
||||
@ -71,25 +70,14 @@ def configured_base_url() -> str:
|
||||
def _split_api_path(path: str) -> urllib.parse.SplitResult:
|
||||
if not isinstance(path, str) or len(path) > MAX_API_TARGET_LENGTH:
|
||||
raise PolicyError("API target exceeds the safe size limit")
|
||||
if (
|
||||
not path.isascii()
|
||||
or any(ord(character) < 32 or ord(character) == 127 for character in path)
|
||||
or "\\" in path
|
||||
or not RAW_API_TARGET_RE.fullmatch(path)
|
||||
):
|
||||
raise PolicyError("API target contains non-canonical raw characters")
|
||||
target = urllib.parse.urlsplit(path)
|
||||
canonical = urllib.parse.urlunsplit(("", "", target.path, target.query, ""))
|
||||
if canonical != path:
|
||||
raise PolicyError("API target is not in exact canonical form")
|
||||
if target.scheme or target.netloc or target.fragment:
|
||||
raise PolicyError("API path must be relative to the Atlas SCM origin")
|
||||
if not target.path.startswith("/api/v1/"):
|
||||
raise PolicyError("API path must start with /api/v1/")
|
||||
if len(target.path) > MAX_API_PATH_LENGTH:
|
||||
raise PolicyError("API path exceeds the safe size limit")
|
||||
segments = target.path.split("/")
|
||||
if "//" in target.path or any(segment in {".", ".."} for segment in segments):
|
||||
if "%" in target.path or "//" in target.path or "/../" in f"{target.path}/":
|
||||
raise PolicyError("encoded or non-canonical API paths are not allowed")
|
||||
return target
|
||||
|
||||
@ -148,7 +136,7 @@ def _authorize_read(target: urllib.parse.SplitResult) -> str:
|
||||
|
||||
def api_url(base_url: str, path: str) -> str:
|
||||
"""Return a canonical same-origin URL without forwarding credentials."""
|
||||
if base_url != CANONICAL_BASE_URL:
|
||||
if base_url.rstrip("/") != CANONICAL_BASE_URL:
|
||||
raise PolicyError("Forgejo origin is fixed to the private Atlas SCM service")
|
||||
target = _split_api_path(path)
|
||||
return urllib.parse.urlunsplit(
|
||||
|
||||
@ -1,18 +1,10 @@
|
||||
"""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.
|
||||
"""
|
||||
"""Pure validation policy for Hermes' least-authority Atlas Forgejo client."""
|
||||
|
||||
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")
|
||||
@ -27,157 +19,73 @@ 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,;}&]*
|
||||
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
|
||||
)
|
||||
"""
|
||||
)
|
||||
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):
|
||||
@ -249,109 +157,10 @@ def _validate_text(value: object, name: str, maximum: int, *, required: bool) ->
|
||||
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):
|
||||
if SENSITIVE_TEXT_RE.search(value):
|
||||
raise PolicyError(f"pull-request {name} resembles credential material")
|
||||
|
||||
|
||||
|
||||
@ -29,11 +29,7 @@ evidence, not authorization to change it.
|
||||
Before opening a PR, verify the remote, branch, clean worktree, diff, tests, and
|
||||
exact pushed commit. Never force-push. Pass that full 40-character pushed head
|
||||
SHA explicitly. Keep the short body free of credentials and credential-like
|
||||
text. The client screens structured assignments, known token formats, and long
|
||||
high-entropy values as an accidental-disclosure barrier. It cannot prove text
|
||||
is secret-free: short or multiword secrets under innocuous names may resemble
|
||||
ordinary prose, so never place credential material in either field. Validate
|
||||
without reading a credential or using the network:
|
||||
text. Validate without reading a credential or using the network:
|
||||
|
||||
```sh
|
||||
/opt/coordinator/gitea_api.py --dry-run create-draft REPO \
|
||||
|
||||
@ -103,10 +103,6 @@ def test_redirect_handler_rejects_cross_origin_with_sentinel_authorization():
|
||||
("https://evil.example", "/api/v1/repos/atlas/cassandra"),
|
||||
("http://scm.bstein.dev", "/api/v1/repos/atlas/cassandra"),
|
||||
("https://scm.bstein.dev:443", "/api/v1/repos/atlas/cassandra"),
|
||||
("https://scm.bstein.dev/", "/api/v1/repos/atlas/cassandra"),
|
||||
("HTTPS://scm.bstein.dev", "/api/v1/repos/atlas/cassandra"),
|
||||
("https://SCM.bstein.dev", "/api/v1/repos/atlas/cassandra"),
|
||||
("https://user@scm.bstein.dev", "/api/v1/repos/atlas/cassandra"),
|
||||
("https://scm.bstein.dev", "https://evil.example/api/v1/repos/atlas/cassandra"),
|
||||
("https://scm.bstein.dev", "/api/v1/repos/evil/cassandra"),
|
||||
("https://scm.bstein.dev", "/api/v1/repos/%61tlas/cassandra"),
|
||||
@ -120,72 +116,6 @@ def test_host_owner_and_path_escape_attempts_are_rejected(base_url: str, path: s
|
||||
client.build_request("GET", path, base_url=base_url, token="secret")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path",
|
||||
[
|
||||
"/api/v1/repos/atlas/cassandra/pulls/1\nHost: evil.example",
|
||||
"/api/v1/repos/atlas/cassandra/pulls/1\r\nX-Test: value",
|
||||
"/api/v1/repos/atlas/cassandra/pulls/1\tignored",
|
||||
"/api/v1/repos/atlas/cassandra/pulls/1\x00ignored",
|
||||
"/api/v1/repos/atlas/cassandra/pulls/1\x1fignored",
|
||||
"/api/v1/repos/atlas/cassandra/pulls/1\x7fignored",
|
||||
"/api/v1/repos/atlas/cassandra\\pulls\\1",
|
||||
"/api/v1/repos/atlas/cassandra/pulls/%31",
|
||||
"/api/v1/repos/atlas/cassandra/pulls/１",
|
||||
" https://scm.bstein.dev/api/v1/repos/atlas/cassandra",
|
||||
"https://scm.bstein.dev/api/v1/repos/atlas/cassandra",
|
||||
],
|
||||
)
|
||||
def test_raw_noncanonical_target_is_rejected_before_urlsplit_and_opener(
|
||||
path: str, monkeypatch
|
||||
):
|
||||
client = _load()
|
||||
split_called = False
|
||||
opener_called = False
|
||||
original_urlsplit = client.urllib.parse.urlsplit
|
||||
|
||||
def urlsplit(*args, **kwargs):
|
||||
nonlocal split_called
|
||||
split_called = True
|
||||
return original_urlsplit(*args, **kwargs)
|
||||
|
||||
def opener(*_args, **_kwargs):
|
||||
nonlocal opener_called
|
||||
opener_called = True
|
||||
return Response(b"{}")
|
||||
|
||||
monkeypatch.setattr(client.urllib.parse, "urlsplit", urlsplit)
|
||||
with pytest.raises(client.PolicyError):
|
||||
client.read(path, token="runtime", opener=opener)
|
||||
assert split_called is False
|
||||
assert opener_called is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path",
|
||||
[
|
||||
"/api/v1/repos/atlas/cassandra/pulls/1?",
|
||||
"//scm.bstein.dev/api/v1/repos/atlas/cassandra",
|
||||
"/api/v1/repos/atlas/cassandra/./pulls/1",
|
||||
"/api/v1/repos/atlas/cassandra/../admin",
|
||||
"/api/v1/repos/atlas/cassandra//pulls/1",
|
||||
"/api/v1/repos/atlas/cassandra/pulls/1?limit=01",
|
||||
],
|
||||
)
|
||||
def test_noncanonical_round_trip_or_segments_never_reach_opener(path: str):
|
||||
client = _load()
|
||||
opener_called = False
|
||||
|
||||
def opener(*_args, **_kwargs):
|
||||
nonlocal opener_called
|
||||
opener_called = True
|
||||
return Response(b"{}")
|
||||
|
||||
with pytest.raises(client.PolicyError):
|
||||
client.read(path, token="runtime", opener=opener)
|
||||
assert opener_called is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path",
|
||||
[
|
||||
@ -501,21 +431,6 @@ def test_runtime_token_is_only_an_authorization_header():
|
||||
"https://discord.com/api/webhooks/123456789/" + "notarealcredentialvalue123456",
|
||||
"https://fake.webhook.office.com/" + "notarealcredentialvalue123456",
|
||||
"webhook_" + "url=https://example.test/not-real",
|
||||
"FutureCloudSigning" + "Credential=not-a-real-value",
|
||||
"future-client-signing-" + "key: not-a-real-value",
|
||||
"future_client_signing_" + "key='not-a-real-value'",
|
||||
'{"serviceAccountPrivate' + 'Key":"not-a-real-value"}',
|
||||
"CONTAINER_REGISTRY_" + "CREDENTIAL=not-a-real-value",
|
||||
"someWebhookSigning" + "Secret: not-a-real-value",
|
||||
"client" + "Key=not-a-real-value",
|
||||
"session" + "Key: not-a-real-value",
|
||||
"access" + "Id=not-a-real-value",
|
||||
"credentials" + ": {user: fake}",
|
||||
"private" + "Key: |",
|
||||
"nuget_api_" + "key=not-a-real-value",
|
||||
"oy2" + "a" * 44,
|
||||
"sk_test_" + "notarealcredentialvalue123456",
|
||||
"A1b2C3d4E5f6G7h8I9j0K_l-M+n/O=pQ2rS3tU4vW5xY6zZ7aB8cC9d",
|
||||
"AKIA" + "A" * 16,
|
||||
"AIza" + "a" * 35,
|
||||
"-----BEGIN OPENSSH " + "PRIVATE KEY-----",
|
||||
@ -562,54 +477,16 @@ def test_create_rejects_expanded_credential_shapes_before_network(
|
||||
assert called is False
|
||||
|
||||
|
||||
def test_very_long_compact_body_is_rejected_before_request_or_network():
|
||||
client = _load()
|
||||
request_built = False
|
||||
opener_called = False
|
||||
original_build_request = client.build_request
|
||||
|
||||
def build_request(*args, **kwargs):
|
||||
nonlocal request_built
|
||||
request_built = True
|
||||
return original_build_request(*args, **kwargs)
|
||||
|
||||
def opener(*_args, **_kwargs):
|
||||
nonlocal opener_called
|
||||
opener_called = True
|
||||
return Response(_draft_response())
|
||||
|
||||
client.build_request = build_request
|
||||
with pytest.raises(client.PolicyError, match="credential material"):
|
||||
client.create_draft(
|
||||
"cassandra",
|
||||
base="main",
|
||||
head="hermes/fix",
|
||||
head_sha=HEAD_SHA,
|
||||
title="Focused fix",
|
||||
body="a" * 300,
|
||||
token="runtime-sentinel",
|
||||
opener=opener,
|
||||
)
|
||||
assert request_built is False
|
||||
assert opener_called is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"safe_text",
|
||||
[
|
||||
"AWS_SECRET_ACCESS_KEY is injected at runtime",
|
||||
"Token: reject empty values",
|
||||
"Authorization = preserve header behavior",
|
||||
"Password: add regression",
|
||||
"Review the Authorization header behavior",
|
||||
"Bearer authentication is required for this route",
|
||||
"Document Docker auths payload rejection",
|
||||
"The client_email field belongs to service accounts",
|
||||
"AccountKey assignments must be blocked",
|
||||
"This patch changes token validation without including a value",
|
||||
"FutureCloudSigningCredential handling needs a regression test",
|
||||
"The clientKey name is documented without an assigned value",
|
||||
"A SHA-256 digest 0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef is evidence",
|
||||
],
|
||||
)
|
||||
def test_credential_policy_keeps_normal_engineering_prose_usable(safe_text: str):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user