From c2f5b4e139bbc31c660184d741629c468827c13a Mon Sep 17 00:00:00 2001 From: Arseniy Dugin Date: Tue, 14 Jul 2026 12:18:46 +0000 Subject: [PATCH] fix(api): clean dynamic API HTTP and connection errors Dynamic `tangle api ` calls now fail with a concise, one-line, credential-safe message and a non-zero exit instead of a raw traceback or a truncated HTTP status exit code. - HTTP status errors report `HTTP for ` plus a length-bounded slice of the backend body. - Connection, timeout, proxy, and TLS errors report an actionable reason. - Failures exit 1. The HTTP status used to become the exit code and was truncated to 8 bits, so 404 surfaced as 148 and 512 as a false success. - Schema refresh/fetch failures reuse the same formatting and omit the response body entirely rather than relying on redaction. Backends and proxies routinely reflect submitted fields back into validation and authentication errors, so the body slice is redacted before it is whitespace-normalized and bounded; a reflected secret cannot survive by sitting past the truncation point. JSON is redacted key by key and each string leaf is scrubbed as text; form, plain-text, HTML, and truncated JSON go through the same scrubber. Sensitivity is decided from the trailing token of a field name rather than from substring containment. Names split on non-alphanumerics and on camelCase word boundaries, and a trailing identifier word is stripped and the remainder judged again, so `access_token`, `myApiKey`, and `accessKeyId` are redacted while `max_tokens`, `session_id`, and `requestId` keep their values. An access-key ID is one half of a credential pair and is revoked with it, so naming it an ID does not make it publishable. Spellings that arrive as one unbroken lowercase run offer no boundary to tokenize on, so they are listed in full and matched exactly rather than as substrings. Displayed URLs keep scheme, host, port, and path and lose userinfo, credential parameters, and presigned/SAS signatures; the rest of a SigV4 query stays readable so an expired link is still diagnosable. Fragments are sanitized on the same terms, because the OAuth implicit flow delivers its access token there. A credential need not be a parameter of its own to be displayed, so decoded values are examined too: as userinfo, as a nested return URL, or as a bare `access_token=...` assignment round-tripped through something like `state`. A nested URL is descended exactly one level, keeping its host and path so the destination stays diagnosable. Encoding a credential again hides it from a scanner that decodes once, so a displayed leaf peels up to four percent-encoding layers and scans each one for every shape above, not just for assignments: `alice%253Apw%2540host` is userinfo one decode further down, and a scan for `=` alone walks straight past it. A layer counts as hiding a credential when either the assignment scrub or the userinfo scrub would rewrite it. Both halves of a query pair go through that scrub, because a name is rendered just as a value is and `access_token%3D...=1` hides a credential in the name; sensitivity is still judged on the name as parsed, so rewriting a name cannot change the verdict on its value. The bounds fail closed. The one-level descent, the JSON depth limit, and the decode cap redact wholesale rather than emit a value that was never examined, and a leaf still changing under decode when the cap runs out is dropped rather than published, so burying a credential deeper than the sanitizer looks costs the whole value. Bounding rather than running to a fixpoint is deliberate: the value is attacker-supplied, so the number of layers must not be theirs to choose, just as re-entering the URL sanitizer would let an untrusted body pick the recursion depth. The tradeoff is that a benign value shaped like a credential, or wrapped in five layers of encoding, is redacted -- cheaper than leaking a token. This runs on every failed request over a body an untrusted backend controls, so the scanners are linear in body length and the JSON sanitizer is iterative and depth-bounded; a few kilobytes of nested arrays would otherwise replace the error message with a RecursionError. The original exception is preserved as the cause for Python callers, and programmatic clients still receive raw httpx exceptions. --- packages/tangle-cli/src/tangle_cli/api_cli.py | 43 +- .../src/tangle_cli/api_transport.py | 791 ++++++++- tests/test_api_cli.py | 171 +- tests/test_api_transport.py | 1469 +++++++++++++++++ 4 files changed, 2442 insertions(+), 32 deletions(-) diff --git a/packages/tangle-cli/src/tangle_cli/api_cli.py b/packages/tangle-cli/src/tangle_cli/api_cli.py index 4a80863..dafad2e 100644 --- a/packages/tangle-cli/src/tangle_cli/api_cli.py +++ b/packages/tangle-cli/src/tangle_cli/api_cli.py @@ -70,7 +70,12 @@ default_auth_header, default_base_url, default_token, + describe_request_error, + format_http_status_error, + format_request_error, + http_status_line, request_operation, + sanitize_url, ) from .cli_helpers import api_arg_specs, load_args_or_exit from .cli_options import ( @@ -185,14 +190,11 @@ def refresh( include_env_credentials=not base_url_from_config, ) except httpx.HTTPStatusError as exc: - message = f"HTTP {exc.response.status_code} {exc.response.reason_phrase}" - raise SystemExit( - f"Failed to fetch {_openapi_url(normalized_base_url)}: {message}" - ) from exc + # Never echo the /openapi.json response body: an auth failure can + # reflect the credentials we just sent. Status line only. + raise SystemExit(_schema_fetch_error_message(normalized_base_url, exc)) from exc except httpx.RequestError as exc: - raise SystemExit( - f"Failed to fetch {_openapi_url(normalized_base_url)}: {exc}" - ) from exc + raise SystemExit(_schema_fetch_error_message(normalized_base_url, exc)) from exc path_count = len(schema.get("paths", {})) print(f"Cached OpenAPI schema for {normalized_base_url}") print(f"Path: {path}") @@ -452,11 +454,13 @@ def _invoke_operation_once( include_env_credentials=include_env_credentials, ) except httpx.HTTPStatusError as exc: - message = exc.response.text or exc.response.reason_phrase - print(message, file=sys.stderr) - raise SystemExit(exc.response.status_code) from exc + # One-line, credential-safe failure with a non-zero exit. The prior code + # raised the HTTP status as the exit code, but exit codes are 8-bit, so a + # status was truncated (404 -> 148, 500 -> 244) and multiples of 256 + # reported success. A string exit prints to stderr and exits 1. + raise SystemExit(format_http_status_error(exc)) from exc except httpx.RequestError as exc: - raise SystemExit(f"Failed to call {exc.request.url}: {exc}") from exc + raise SystemExit(format_request_error(exc)) from exc except TypeError as exc: raise SystemExit(str(exc)) from exc @@ -697,6 +701,23 @@ def _validate_schema_source(value: str) -> str: return normalized +def _schema_fetch_error_message(base_url: str, exc: Exception) -> str: + """One-line, credential-safe failure for an /openapi.json fetch. + + The response body is deliberately omitted for HTTP status errors so an auth + failure cannot reflect the credentials that were just sent. + """ + + target = sanitize_url(_openapi_url(base_url)) + if isinstance(exc, httpx.HTTPStatusError): + reason = http_status_line(exc) + elif isinstance(exc, httpx.RequestError): + reason = describe_request_error(exc) + else: + reason = exc.__class__.__name__ + return f"Failed to fetch {target}: {reason}" + + def _schema_fetch_failure_message(base_url: str, exc: Exception) -> str: if isinstance(exc, httpx.HTTPStatusError): reason = f"HTTP {exc.response.status_code} {exc.response.reason_phrase}" diff --git a/packages/tangle-cli/src/tangle_cli/api_transport.py b/packages/tangle-cli/src/tangle_cli/api_transport.py index b128070..12c0a03 100644 --- a/packages/tangle-cli/src/tangle_cli/api_transport.py +++ b/packages/tangle-cli/src/tangle_cli/api_transport.py @@ -5,6 +5,7 @@ import json import os import re +import string import sys import urllib.parse from pathlib import Path @@ -17,12 +18,171 @@ _HEADER_NAME_RE = re.compile(r"^[!#$%&'*+.^_`|~0-9A-Za-z-]+$") _MISSING = object() _SENSITIVE_HEADER_NAMES = {"authorization", "cloud-auth", "cookie", "x-api-key"} -_SENSITIVE_KEY_RE = re.compile( - r"(authorization|authentication|(^|[-_])auth($|[-_])|cloud[-_]?auth|cookie|x[-_]?api[-_]?key|token|secret|password|credential|pre[-_]?signed[-_]?url|signed[-_]?url)", +# A field name is judged by its *tokens*, not by substring containment, so +# ``max_tokens``, ``token_count``, ``function_signature``, and ``secretary_email`` +# keep their values while ``access_token`` and ``my_api_key`` lose theirs. +_FIELD_NAME_TOKEN_SPLIT_RE = re.compile(r"[^A-Za-z0-9]+") +# camelCase/PascalCase word boundaries, so ``accessToken`` tokenizes the same way +# ``access_token`` does. The second alternative splits an acronym from the word +# that follows it (``APIKey`` -> ``API`` ``Key``) without splitting the acronym +# itself, which is what keeps ``XApiKey`` and ``oauth2Token`` recognizable. +_FIELD_NAME_CAMEL_BOUNDARY_RE = re.compile(r"(?<=[a-z0-9])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])") +# Tokens that name a credential on their own, as the whole field name or as its +# final token. Deliberately singular: ``tokens`` counts usage, ``token`` is one. +_CREDENTIAL_WORDS = frozenset( + { + "auth", + "authentication", + "authorization", + "cookie", + "credential", + "credentials", + "passphrase", + "passwd", + "password", + "pwd", + "secret", + "token", + # Credential names that arrive as one unbroken lowercase run and so offer + # no boundary to tokenize on. Listed in full rather than matched as + # substrings, which is what keeps ``accesskeyidformat``, ``privatekeypath``, + # ``maxtokens``, ``sessionid``, ``oauthlib``, and ``tokenizer`` readable. + "accessid", + "accesskey", + "accesskeyid", + "accesstoken", + "apikey", + "apisecret", + "apitoken", + "authtoken", + "awsaccesskeyid", + "bearertoken", + "clientsecret", + "googleaccessid", + "idtoken", + "oauth", + "privatekey", + "refreshtoken", + "secretkey", + "sessionkey", + "sessiontoken", + } +) +# Trailing words that name an identifier. They do not make a field safe: an +# access-key ID is one half of a credential pair and is issued and revoked with +# it, so the suffix is stripped and what it was attached to is judged again. +# ``access_key_id_format`` is unaffected -- its trailing word is ``format``. +_IDENTIFIER_SUFFIX_WORDS = frozenset({"id", "ident", "identifier"}) +# Words that name key material only once an identifier suffix is stripped. +# ``access`` alone is ordinary; ``access_id`` and ``GoogleAccessId`` are not. +_CREDENTIAL_ID_WORDS = frozenset({"access"}) +# Credential names that only read as one across two tokens. ``key`` and ``url`` +# are far too common alone, so they are sensitive only in these pairings. +_CREDENTIAL_PHRASES = frozenset( + { + ("access", "key"), + ("api", "key"), + ("presigned", "url"), + ("private", "key"), + ("secret", "key"), + ("session", "key"), + ("signed", "url"), + } +) +# Credential fields whose value is a URL. Blanket redaction throws away the host +# and path -- the part that separates a wrong bucket from an expired grant -- so +# inside a parsed body these go through the URL scrubber instead, which strips +# the signature and keeps the rest. A value with no URL in it has no such +# structure to lean on and is still redacted whole, as are header and query +# occurrences, whose values are never scrubbed. +_URL_VALUED_CREDENTIAL_PHRASES = frozenset({("presigned", "url"), ("signed", "url")}) +# Query parameters that carry the credential portion of a presigned/SAS URL +# (AWS SigV4, GCS, Azure). Redacting the signature neutralizes the grant, so the +# rest of the SigV4 parameter set (``X-Amz-Algorithm``, ``X-Amz-Date``, +# ``X-Amz-Expires``, ``X-Amz-SignedHeaders``) stays readable and remains useful +# for diagnosing an expired or malformed link. These names are matched in full, +# so ``function_signature`` is untouched. +_SIGNED_URL_QUERY_RE = re.compile( + r"^(x-(amz|goog|ms)-(signature|credential|security-token)" + r"|sig|signature|awsaccesskeyid|googleaccessid)$", + re.IGNORECASE, +) +# HTTP authentication schemes that prefix the credential rather than being one. +# The scheme name is diagnostic and kept; only the credential after it is cut. +_AUTH_SCHEME = ( + r"Bearer|Basic|Digest|Token|JWT|OAuth|ApiKey|SSWS|Negotiate|NTLM" + r"|GoogleLogin|AWS4-HMAC-SHA256" +) +# The ``=``/``:`` separator of an assignment, on its own. Scanning for +# separators rather than for a fixed list of key names is what makes the field +# name open-ended: the name is recovered by a bounded lookbehind and judged by +# :func:`_is_sensitive_query_key`, so affixed names (``access_token``, +# ``my_api_key``) are covered without an alternation having to enumerate them. +# Free text is judged by the query predicate rather than the body one because a +# reflected ``X-Amz-Signature=...`` reads as a credential wherever it appears. +# The pattern deliberately stops at the separator so a declined match consumes +# nothing of the value -- a nested assignment under a harmless outer key +# (``{"detail": "token=..."``) is still reached. Optional quotes are absorbed so +# truncated JSON is scrubbed too, and ``(?!//)`` keeps ``https://host`` from +# reading as an assignment. Everything after the separator character is only +# scanned once a separator matched, so scanning is linear in the text length. +_ASSIGNMENT_SEPARATOR_RE = re.compile(r"[:=][ \t]*[\"']?(?!//)") +# Characters that may sit between a field name and its separator. +_FIELD_NAME_PADDING = " \t\"'" +# Characters a field name (or its padding) can end with, for an O(1) pre-check +# that skips the lookbehind for separators no name could precede. +_FIELD_NAME_TAIL_CHARS = _FIELD_NAME_PADDING + "_.-" +# The value of an assignment, anchored immediately after its separator. The +# value stops at whitespace, a form/list delimiter, a quote, or ``?`` so +# form-encoded, HTML-embedded, and query-string values stay bounded. +_ASSIGNMENT_VALUE_RE = re.compile( + rf"(?:(?P{_AUTH_SCHEME})[ \t]+)?(?P[^\s&;,<>\"'?]+)", re.IGNORECASE, ) +# Trailing field name immediately preceding an assignment separator. +_FIELD_NAME_RE = re.compile(r"[A-Za-z][A-Za-z0-9_.\-]*$") +# Longest field name considered; bounds the per-separator lookbehind. +_MAX_FIELD_NAME_CHARS = 64 +# An auth scheme carrying its credential without a preceding field name (e.g. +# ``rejected header: Bearer sk-1``, where the name is not itself sensitive). +# Matched case-sensitively against canonical scheme spellings so ordinary prose +# ("token count: 42") is not read as a credential. +_BARE_AUTH_SCHEME_RE = re.compile( + r"\b(?PBearer|Basic|Digest|Negotiate|NTLM|SSWS|JWT|ApiKey|Token|OAuth" + r"|GoogleLogin|AWS4-HMAC-SHA256)[ \t]+(?P[A-Za-z0-9\-._~+/=]+)" +) +# Scheme names that are also ordinary English words, so a bare occurrence needs a +# credential long enough that "Token 12345 expired" cannot trip it. +_AMBIGUOUS_BARE_SCHEMES = {"Token", "OAuth"} +_MIN_AMBIGUOUS_SCHEME_CREDENTIAL_CHARS = 16 +# Upper bound on backend-supplied error detail rendered on a single line. +_MAX_BACKEND_DETAIL_CHARS = 500 +# Character classes for locating an embedded ``scheme://...`` run. The scan +# anchors on the ``://`` literal and walks outward, because a regex of the form +# ``[a-zA-Z][a-zA-Z0-9+.\-]*://`` re-scans every alphanumeric run once per +# starting offset and so costs O(n^2) on a body that is one long run. +_URL_SCHEME_CHARS = frozenset(string.ascii_letters + string.digits + "+.-") +_URL_SCHEME_START_CHARS = frozenset(string.ascii_letters) +_URL_SEPARATOR = "://" +_URL_STOP_CHARS = frozenset("'\"<>") +# Punctuation that ordinarily terminates a sentence rather than a URL. +_URL_TRAILING_PUNCTUATION = ").,;'\"" +# ``user:pass@host`` userinfo carries the credential before the ``@``; the ``@`` +# is the anchor, and the host side is never part of the secret. +_USERINFO_STOP_CHARS = frozenset("/@") +# Percent-encoding layers peeled off a displayed leaf while looking for a hidden +# credential assignment. Encoding ``access_token=x`` again hides the separator +# from a scanner that decodes once, so one layer is not enough; the cap is what +# keeps the walk linear, since each peel is O(len) and there is a fixed number of +# them. A leaf still changing under decode when the cap runs out is dropped. +_MAX_DECODE_LAYERS = 4 _REDACTED = "" _REDACTED_DOCUMENT = "" +_REDACTED_DEEP = "" +# Structure nesting kept when redacting a parsed JSON body. Error detail is +# collapsed onto one bounded line anyway, so nothing legible is lost past this +# point, and a hostile body cannot make an always-on error path recurse. +_MAX_JSON_DEPTH = 32 _OPAQUE_DOCUMENT_KEY_NAMES = { "component_yaml", "dockerfile", @@ -40,38 +200,221 @@ def tangle_verbose_enabled() -> bool: return value.strip().lower() in {"1", "true", "yes", "on"} +def _field_name_tokens(name: str) -> list[str]: + """Split *name* on non-alphanumerics and camelCase/PascalCase boundaries.""" + + split = _FIELD_NAME_CAMEL_BOUNDARY_RE.sub(" ", name).lower() + return [token for token in _FIELD_NAME_TOKEN_SPLIT_RE.split(split) if token] + + +def _names_credential(tokens: list[str]) -> bool: + """Do *tokens* end in a word, or pair of words, that names a credential?""" + + if tokens[-1] in _CREDENTIAL_WORDS: + return True + return len(tokens) >= 2 and tuple(tokens[-2:]) in _CREDENTIAL_PHRASES + + +def _is_url_valued_credential_field(name: str) -> bool: + """Is *name* a credential field whose value is expected to be a URL?""" + + tokens = _field_name_tokens(name) + return len(tokens) >= 2 and tuple(tokens[-2:]) in _URL_VALUED_CREDENTIAL_PHRASES + + +def _is_sensitive_field_name(name: str) -> bool: + """Is *name* a header or body field whose value must never be displayed? + + The name is split into tokens on any non-alphanumeric character and on + camelCase/PascalCase word boundaries, then judged by its final token (or + final two), so an affixed credential such as ``access_token``, + ``accessToken``, ``client_secret``, or ``myApiKey`` is caught without the + predicate having to enumerate prefixes or spellings. A trailing identifier + word is stripped and the remainder judged again, which is what catches + ``access_key_id``, ``AwsAccessKeyId``, and ``GoogleAccessId``. + + Judging tokens rather than substrings is what keeps ordinary fields + readable: ``max_tokens``, ``tokenCount``, ``function_signature``, + ``signatureVersion``, ``password_policy``, ``private_key_path``, and + ``secretaryEmail`` all end in a token that names something other than a + credential. ``access_key_id_format`` is readable for the same reason -- its + identifier word is not trailing, so nothing is stripped. + """ + + tokens = _field_name_tokens(name) + if not tokens: + return False + if _names_credential(tokens): + return True + while tokens[-1] in _IDENTIFIER_SUFFIX_WORDS: + tokens = tokens[:-1] + if not tokens: + return False + if tokens[-1] in _CREDENTIAL_ID_WORDS or _names_credential(tokens): + return True + return False + + def _redact_headers(headers: dict[str, Any] | None) -> dict[str, Any]: redacted: dict[str, Any] = {} for name, value in (headers or {}).items(): normalized_name = name.lower() redacted[name] = ( _REDACTED - if normalized_name in _SENSITIVE_HEADER_NAMES or _SENSITIVE_KEY_RE.search(name) + if normalized_name in _SENSITIVE_HEADER_NAMES or _is_sensitive_field_name(name) else value ) return redacted def _redact_sensitive_values(value: Any, key: str | None = None) -> Any: - if key and _SENSITIVE_KEY_RE.search(key): - return _REDACTED - if key and key.lower() in _OPAQUE_DOCUMENT_KEY_NAMES and isinstance(value, str) and value: - return _REDACTED_DOCUMENT - if isinstance(value, dict): - return {str(k): _redact_sensitive_values(v, str(k)) for k, v in value.items()} - if isinstance(value, list): - return [_redact_sensitive_values(item) for item in value] - return value + """Rebuild *value* with credential-bearing fields and string leaves redacted. + + Iterative rather than recursive, and bounded by :data:`_MAX_JSON_DEPTH`: this + runs on every failed request, on a body an untrusted backend controls, and a + few kilobytes of ``[[[[...]]]]`` is enough to exhaust the interpreter stack. + Anything nested deeper than the bound is replaced wholesale, so the bound + fails closed -- an unexamined subtree is never emitted. + """ + + holder: list[Any] = [None] + # (container, slot, node, node_key, depth); the result is written into the + # slot the parent reserved, which keeps dict insertion order intact. + pending: list[tuple[Any, Any, Any, str | None, int]] = [(holder, 0, value, key, 0)] + while pending: + container, slot, node, node_key, depth = pending.pop() + if ( + node_key + and isinstance(node, str) + and _URL_SEPARATOR in node + and _is_url_valued_credential_field(node_key) + ): + container[slot] = _redact_text_secrets(node) + elif node_key and _is_sensitive_field_name(node_key): + container[slot] = _REDACTED + elif ( + node_key + and node_key.lower() in _OPAQUE_DOCUMENT_KEY_NAMES + and isinstance(node, str) + and node + ): + container[slot] = _REDACTED_DOCUMENT + elif isinstance(node, dict): + if depth >= _MAX_JSON_DEPTH: + container[slot] = _REDACTED_DEEP + continue + branch: dict[str, Any] = {} + container[slot] = branch + for child_key in node: + branch[str(child_key)] = None + # Reversed so the stack pops in source order and a duplicated + # ``str(key)`` resolves to the last occurrence, as a dict would. + for child_key, child in reversed(list(node.items())): + name = str(child_key) + pending.append((branch, name, child, name, depth + 1)) + elif isinstance(node, list): + if depth >= _MAX_JSON_DEPTH: + container[slot] = _REDACTED_DEEP + continue + items: list[Any] = [None] * len(node) + container[slot] = items + for index, child in enumerate(node): + pending.append((items, index, child, None, depth + 1)) + elif isinstance(node, str): + # A non-sensitive field can still quote a sensitive assignment back at + # us (``{"detail": "invalid token=... supplied"}``), so string leaves + # get the same free-text scrub as an unparseable body. + container[slot] = _redact_text_secrets(node) + else: + container[slot] = node + return holder[0] def _safe_json_text(value: Any) -> str: redacted = _redact_sensitive_values(value) try: return json.dumps(redacted, indent=2, sort_keys=True, default=str) - except TypeError: + except (TypeError, ValueError, RecursionError): return str(redacted) +def _looks_like_opaque_credential(value: str) -> bool: + """Heuristic: does *value* resemble an opaque credential rather than a word? + + Only consulted for a bare auth scheme, where there is no field name to judge + and ``Basic authentication failed`` must survive. Tokens carry digits, + base64/JWT punctuation, or are long; an English word does not. + """ + + return len(value) >= 20 or any(ch.isdigit() or ch in "+/=._~-" for ch in value) + + +def _redact_assignments(text: str) -> str: + """Redact credential assignments and auth-scheme credentials in *text*. + + Each ``=``/``:`` separator is found once, the field name preceding it is + recovered by a bounded lookbehind, and :func:`_is_sensitive_query_key` + decides -- so any spelling of a credential field is covered, including + affixed names such as ``access_token`` or ``my_api_key`` that no fixed + alternation would list. Only the value is replaced, so non-sensitive + assignments (``page=2``) and the surrounding diagnostic prose survive. An + auth scheme keeps its name (``Bearer ``) because the scheme is + useful and the credential is not. + + Work is bounded per separator and per match, with no nested quantifier, so + cost stays linear in ``len(text)``. Nothing here parses a URL or re-enters + :func:`sanitize_url`, which is what lets the URL sanitizer reuse it as a leaf. + """ + + def _replace_bare_scheme(match: re.Match[str]) -> str: + scheme, value = match.group("scheme"), match.group("value") + if scheme in _AMBIGUOUS_BARE_SCHEMES: + if len(value) < _MIN_AMBIGUOUS_SCHEME_CREDENTIAL_CHARS: + return match.group(0) + elif not _looks_like_opaque_credential(value): + return match.group(0) + return f"{scheme} {_REDACTED}" + + chunks: list[str] = [] + cursor = 0 + for separator in _ASSIGNMENT_SEPARATOR_RE.finditer(text): + start = separator.start() + if start < cursor: + # Inside a value that was already redacted. + continue + if start == 0 or not ( + text[start - 1].isalnum() or text[start - 1] in _FIELD_NAME_TAIL_CHARS + ): + # Nothing that could end a field name; skip the lookbehind entirely. + continue + preceding = text[max(0, start - _MAX_FIELD_NAME_CHARS) : start] + name = _FIELD_NAME_RE.search(preceding.rstrip(_FIELD_NAME_PADDING)) + if name is None or not _is_sensitive_query_key(name.group(0)): + continue + value = _ASSIGNMENT_VALUE_RE.match(text, separator.end()) + if value is None: + continue + chunks.append(text[cursor : value.start("value")]) + chunks.append(_REDACTED) + cursor = value.end("value") + chunks.append(text[cursor:]) + return _BARE_AUTH_SCHEME_RE.sub(_replace_bare_scheme, "".join(chunks)) + + +def _redact_text_secrets(text: str) -> str: + """Redact credentials in free text. + + Handles every representation the structured JSON path cannot: form-encoded + bodies, plain text, HTML error pages, truncated/unparseable JSON, and string + leaves inside otherwise well-formed JSON. Credentials carried in a URL or in + bare ``user:pass@host`` userinfo are stripped first by + :func:`_scrub_secret_text`, the same primitive used on exception text; what + survives as a plain assignment is then caught by :func:`_redact_assignments`. + """ + + return _redact_assignments(_scrub_secret_text(text)) + + def _content_to_text(content: bytes | str | None) -> str: if content is None: return "" @@ -86,10 +429,430 @@ def _content_to_text(content: bytes | str | None) -> str: try: parsed = json.loads(text) except Exception: - return text + return _redact_text_secrets(text) return _safe_json_text(parsed) +def _is_sensitive_query_key(key: str) -> bool: + """Is *key* a query parameter whose value must never be displayed? + + Stricter than the header/body predicate: a query string is also where a + presigned-URL grant lives, and those parameter names (``sig``, ``signature``, + ``X-Amz-*``) are credentials in a way the same word is not when it names a + body field such as ``function_signature``. + """ + + stripped = key.strip() + return _is_sensitive_field_name(stripped) or bool( + _SIGNED_URL_QUERY_RE.fullmatch(stripped) + ) + + +def _redact_parameter_credentials(query: str) -> str: + """Redact credential-named parameters in *query*, or drop it wholesale. + + The leaf of the sanitizer, and where the descent stops. It judges parameter + names and strips userinfo, but it will not parse a value as a URL in turn — + so a value that still looks like one is redacted whole rather than emitted + unexamined. That is what makes the depth bound safe instead of merely finite: + nesting a credential one level deeper than the sanitizer looks buries it + rather than smuggling it out. + """ + + if not query: + return "" + try: + pairs = urllib.parse.parse_qsl(query, keep_blank_values=True) + except ValueError: + return _REDACTED + if not pairs or urllib.parse.urlencode(pairs) != query: + return _REDACTED + return urllib.parse.urlencode( + [ + (_redact_bare_userinfo(key), _redact_leaf_value(key, value)) + for key, value in pairs + ], + safe="<>", + ) + + +def _redact_leaf_value(key: str, value: str) -> str: + """Redact *value* if its name is sensitive or it hides a further URL.""" + + if _is_sensitive_query_key(key) or _URL_SEPARATOR in value: + return _REDACTED + return _redact_assignments(_redact_bare_userinfo(value)) + + +def _redact_nested_url_credentials(value: str) -> str: + """Redact credential parameters inside a nested absolute URL. + + A return URL is routinely carried inside another URL + (``?next=https%3A%2F%2Fhost%2Fcb%3Faccess_token%3D...``), and once + :func:`urllib.parse.parse_qsl` has decoded it the inner credential is plainly + visible. Its scheme, host, and path are kept so the destination stays + diagnosable; only its own parameters are judged, by + :func:`_redact_parameter_credentials`, which does not descend again. Going + exactly one level deep is deliberate: :func:`sanitize_url` must not be + re-entered here, or a URL nested inside a URL inside a URL would drive the + recursion as deep as an untrusted body cared to nest it. + """ + + if _URL_SEPARATOR not in value: + return value + try: + parsed = urllib.parse.urlsplit(value) + except ValueError: + return _REDACTED + if not parsed.scheme or not (parsed.query or parsed.fragment): + return value + return urllib.parse.urlunsplit( + ( + parsed.scheme, + parsed.netloc, + parsed.path, + _redact_parameter_credentials(parsed.query), + _redact_parameter_credentials(parsed.fragment), + ) + ) + + +def _hides_credential(value: str) -> bool: + """Report whether bounded percent-decoding of *value* reveals a credential. + + ``state=access_token%3D...`` is legible after the single decode + :func:`urllib.parse.parse_qsl` already performed. Encoding it again hides the + separator from :func:`_redact_assignments`, and encoding it a third time hides + it from any check that decodes only once more. Layers are therefore peeled up + to ``_MAX_DECODE_LAYERS`` and every intermediate form is scanned. + + Every shape the caller redacts at the surface is looked for at each layer, not + just assignments: ``alice%253Apw%2540host`` is userinfo one decode further + down, and a scan for ``=`` alone walks straight past it. + + The walk is bounded rather than run to a fixpoint: an attacker supplies the + value, so the number of layers must not be theirs to choose. When the cap runs + out on a value that is still changing under decode, encoding remains that was + never looked behind, and that is reported as hiding a credential -- burying one + deeper than the sanitizer looks drops the value instead of publishing it. + """ + + for _ in range(_MAX_DECODE_LAYERS): + decoded = urllib.parse.unquote(value) + if decoded == value: + return False + if ( + _redact_assignments(decoded) != decoded + or _redact_bare_userinfo(decoded) != decoded + ): + return True + value = decoded + return urllib.parse.unquote(value) != value + + +def _scrub_parameter_value(value: str) -> str: + """Scrub a decoded parameter value in every shape a credential arrives in. + + An absolute URL is sanitized structurally, so its host and path survive. Any + other shape -- a relative return path (``/cb?access_token=...``), an opaque + ``mailto:``/``data:`` URI, or a bare ``access_token=...`` pair that a caller + round-trips through ``state`` -- has no structure worth preserving, so it is + scanned for credential assignments instead. The scan runs last either way: + ahead of :func:`_redact_nested_url_credentials` it would rewrite the nested + query out of its exact-round-trip form and cost that URL its host and path. + """ + + value = _redact_bare_userinfo(value) + if _URL_SEPARATOR in value: + value = _redact_nested_url_credentials(value) + value = _redact_assignments(value) + if _hides_credential(value): + # Encoded past what the scans above can read (``state=access_token%253D...``, + # ``next=alice%253Apw%2540host``). Emitting it would publish a value never + # examined, so fail closed. + return _REDACTED + return value + + +def _sanitize_query_pairs(pairs: list[tuple[str, str]]) -> str: + """Re-encode *pairs*, redacting credential names and scrubbing what remains. + + :func:`urllib.parse.parse_qsl` has already percent-decoded both halves of a + pair, so a credential smuggled through a harmless-looking name is legible + here: as userinfo (``next=https%3A%2F%2Fu%3Apw%40host``), as a parameter of a + nested URL (``next=...%3Faccess_token%3D...``), as a bare assignment + (``state=access_token%3D...``), or in the parameter name itself, either as + userinfo (``alice%3Apw%40host=1``) or as a whole assignment encoded into the + name (``access_token%3D...=1``). A name is displayed just as a value is, so + both halves go through the same leaf scrub. + + Sensitivity is judged on the name as parsed, before that scrub, so rewriting + a name cannot change the verdict on its value. + + Only non-recursive primitives are used. :func:`_scrub_secret_text` in + particular is not, because it routes embedded URLs back through + :func:`sanitize_url`, which arrives here again. + """ + + return urllib.parse.urlencode( + [ + ( + _scrub_parameter_value(key), + _REDACTED + if _is_sensitive_query_key(key) + else _scrub_parameter_value(value), + ) + for key, value in pairs + ], + safe="<>", + ) + + +def _sanitize_fragment(fragment: str) -> str: + """Return *fragment* with credentials removed, or drop it wholesale. + + A fragment is not merely an anchor: the OAuth implicit flow returns + ``#access_token=...&token_type=Bearer`` there precisely so the credential + stays out of the query, and it reaches this function on the always-on error + display path. + + A fragment carrying neither ``=`` nor ``&`` is an anchor. Its userinfo is + still stripped (``#u:pw@host`` is a valid anchor and a leak), and if it + embeds a whole URL it is dropped instead, because sanitizing that properly + would mean re-entering :func:`sanitize_url` from inside itself. Otherwise + the fragment is parsed as a query string and re-encoded from the parsed + pairs, so every value emitted is one that was examined. If the re-encoding + does not reproduce the original exactly, some separator this does not model + (``;``, a stray encoding) could be hiding a value that was never examined, + so the fragment is dropped rather than guessed at. + """ + + if not fragment: + return "" + if "=" not in fragment and "&" not in fragment: + if _URL_SEPARATOR in fragment: + return _REDACTED + anchor = _redact_assignments(_redact_bare_userinfo(fragment)) + if _hides_credential(anchor): + return _REDACTED + return anchor + try: + pairs = urllib.parse.parse_qsl(fragment, keep_blank_values=True) + except ValueError: + return _REDACTED + if not pairs or urllib.parse.urlencode(pairs) != fragment: + return _REDACTED + return _sanitize_query_pairs(pairs) + + +def sanitize_url(url: Any) -> str: + """Return *url* with credentials removed so it is safe to display or log. + + Strips any ``user:password@`` userinfo and redacts the values of query + parameters that look like tokens, credentials, or presigned/SAS-URL + signatures. A parameter kept under a non-sensitive name is scrubbed in turn, + since a credential rides through one either as userinfo + (``next=https%3A%2F%2Fu%3Apw%40host``) or as a parameter of the URL nested + inside it (``next=...%3Faccess_token%3D...``); the nested URL is descended + exactly one level. The scheme, host, port, path, and non-sensitive parameter + names are preserved so the target stays recognizable. The fragment is + sanitized on the same terms by :func:`_sanitize_fragment`, because the OAuth + implicit flow delivers its access token there. + """ + + text = str(url) + try: + parsed = urllib.parse.urlsplit(text) + except ValueError: + return _REDACTED + if not parsed.scheme and not parsed.netloc: + return text + host = parsed.hostname or "" + # ``hostname`` unwraps IPv6 literals, so re-bracket them before appending an + # optional port; otherwise ``[2001:db8::1]:8443`` becomes ambiguous garbage. + if ":" in host: + host = f"[{host}]" + netloc = f"{_REDACTED}@{host}" if (parsed.username or parsed.password) else host + if parsed.port is not None: + netloc = f"{netloc}:{parsed.port}" + query = parsed.query + if query: + query = _sanitize_query_pairs( + urllib.parse.parse_qsl(query, keep_blank_values=True) + ) + fragment = _sanitize_fragment(parsed.fragment) + return urllib.parse.urlunsplit((parsed.scheme, netloc, parsed.path, query, fragment)) + + +def _bounded_detail(text: str | None) -> str: + """Collapse whitespace and cap length of backend-supplied error detail.""" + + if not text: + return "" + collapsed = " ".join(str(text).split()) + if len(collapsed) > _MAX_BACKEND_DETAIL_CHARS: + collapsed = collapsed[:_MAX_BACKEND_DETAIL_CHARS].rstrip() + "…" + return collapsed + + +def http_status_line(exc: httpx.HTTPStatusError) -> str: + """Return the ``HTTP `` summary for a status error.""" + + response = exc.response + return f"HTTP {response.status_code} {response.reason_phrase}".strip() + + +def format_http_status_error(exc: httpx.HTTPStatusError, *, include_detail: bool = True) -> str: + """Build a concise one-line message for an httpx HTTP status error. + + Includes the status, request method, and a credential-safe URL. When + *include_detail* is set, the response body is first run through the same + secret redaction used for verbose logging -- key-by-key for JSON, and + ``key=value``/``key: value`` assignment scrubbing for form, plain-text, and + HTML bodies -- and only then whitespace normalized and length bounded, so a + secret can never survive by sitting past the truncation point. Backend + messages remain visible without leaking reflected credentials or dumping + multi-line/oversized payloads. + """ + + request = exc.request + method = request.method if request is not None else "?" + url = sanitize_url(request.url) if request is not None else "?" + message = f"{http_status_line(exc)} for {method} {url}" + if include_detail: + try: + body = exc.response.text + except Exception: # pragma: no cover - defensive: streamed/undecodable body + body = "" + # Backends and proxies can reflect submitted fields (tokens, passwords) + # into validation/authentication errors -- as JSON, form data, or an HTML + # error page -- so redact before the body reaches stderr and CI logs. + # Redaction precedes bounding so truncation cannot expose a secret. + detail = _bounded_detail(_content_to_text(body)) if body else "" + if detail: + message = f"{message}: {detail}" + return message + + +def _redact_embedded_urls(text: str) -> str: + """Route every ``scheme://...`` run in *text* through :func:`sanitize_url`. + + Anchored on the ``://`` literal: the scheme is recovered by walking back over + the scheme characters and the target by walking forward to the first + delimiter. Consecutive runs cannot overlap, so every character is visited a + bounded number of times whatever shape the input has. + """ + + chunks: list[str] = [] + cursor = 0 + separator = text.find(_URL_SEPARATOR) + while separator != -1: + start = separator + while start > cursor and text[start - 1] in _URL_SCHEME_CHARS: + start -= 1 + while start < separator and text[start] not in _URL_SCHEME_START_CHARS: + start += 1 + if start < separator: + end = separator + len(_URL_SEPARATOR) + while end < len(text) and not ( + text[end].isspace() or text[end] in _URL_STOP_CHARS + ): + end += 1 + raw = text[start:end] + trailing = "" + while raw and raw[-1] in _URL_TRAILING_PUNCTUATION: + trailing = raw[-1] + trailing + raw = raw[:-1] + chunks.append(text[cursor:start]) + chunks.append(sanitize_url(raw) + trailing) + cursor = end + separator = text.find(_URL_SEPARATOR, max(separator + 1, cursor)) + chunks.append(text[cursor:]) + return "".join(chunks) + + +def _redact_bare_userinfo(text: str) -> str: + """Redact schemeless ``user:pass@host`` userinfo, keeping the host. + + Anchored on ``@`` and walking back only to the nearest delimiter, so the cost + is linear even when the text is one unbroken run. + """ + + chunks: list[str] = [] + cursor = 0 + at = text.find("@") + while at != -1: + start = at + while start > cursor and not ( + text[start - 1].isspace() or text[start - 1] in _USERINFO_STOP_CHARS + ): + start -= 1 + userinfo = text[start:at] + colon = userinfo.find(":") + # An empty username is legal userinfo (``:password@host``), so the colon + # may sit at offset zero; it may not sit last, or there is no credential. + if 0 <= colon < len(userinfo) - 1: + chunks.append(text[cursor:start]) + chunks.append(_REDACTED) + cursor = at + at = text.find("@", at + 1) + chunks.append(text[cursor:]) + return "".join(chunks) + + +def _scrub_secret_text(text: str) -> str: + """Redact URLs and bare userinfo embedded in free-form text. + + A crafted or third-party ``httpx`` exception -- and equally a reflected + response body -- can carry a proxy URL, a signed query, or ``user:pass@host`` + inside its message. Never emit that raw: route every ``scheme://`` run + through :func:`sanitize_url` and strip any remaining schemeless userinfo, + while leaving benign diagnostics (errno, TLS reason) intact. + """ + + return _redact_bare_userinfo(_redact_embedded_urls(text)) + + +def describe_request_error(exc: httpx.RequestError) -> str: + """Return an actionable, credential-safe reason for an httpx request error. + + Connection, timeout, proxy, and TLS failures are labeled so the user knows + what to check; the underlying detail is included when it adds information. + Any URL or userinfo embedded in the exception text is redacted first. + """ + + detail = _scrub_secret_text(" ".join(str(exc).split())) + lowered = detail.lower() + if isinstance(exc, httpx.ProxyError): + return f"proxy error: {detail}" if detail else "proxy error" + if isinstance(exc, httpx.TimeoutException): + label = { + httpx.ConnectTimeout: "connection timed out", + httpx.ReadTimeout: "read timed out", + httpx.WriteTimeout: "write timed out", + httpx.PoolTimeout: "connection pool timed out", + }.get(type(exc), "request timed out") + return f"{label}: {detail}" if detail and label not in lowered else label + if isinstance(exc, httpx.ConnectError): + if any(token in lowered for token in ("ssl", "certificate", "tls", "handshake")): + return f"TLS error: {detail}" if detail else "TLS error" + return f"connection failed: {detail}" if detail else "connection failed" + if detail: + return detail + return exc.__class__.__name__ + + +def format_request_error(exc: httpx.RequestError) -> str: + """Build a concise one-line message for an httpx connection-level error.""" + + request = getattr(exc, "request", None) + reason = describe_request_error(exc) + if request is None: + return f"Failed to reach the backend: {reason}" + url = sanitize_url(request.url) + return f"Failed to reach {request.method} {url}: {reason}" + + def log_http_exchange( logger: Any, *, diff --git a/tests/test_api_cli.py b/tests/test_api_cli.py index c93a575..8115b20 100644 --- a/tests/test_api_cli.py +++ b/tests/test_api_cli.py @@ -1681,6 +1681,37 @@ def fake_get(url, **kwargs): assert "secret-token" not in message +def test_refresh_connection_error_is_clean_and_actionable(monkeypatch): + def fake_get(url, **kwargs): + raise httpx.ConnectError("connection refused", request=httpx.Request("GET", url)) + + monkeypatch.setattr(api_cli.httpx, "get", fake_get) + app = api_cli.build_app(SCHEMA) + + with pytest.raises(SystemExit) as exc_info: + app(["refresh", "--base-url", "http://api.test"]) + + message = str(exc_info.value) + assert "\n" not in message + assert message.startswith("Failed to fetch http://api.test/openapi.json:") + assert "connection failed: connection refused" in message + + +def test_refresh_error_redacts_base_url_credentials(monkeypatch): + def fake_get(url, **kwargs): + raise httpx.ConnectError("connection refused", request=httpx.Request("GET", url)) + + monkeypatch.setattr(api_cli.httpx, "get", fake_get) + app = api_cli.build_app(SCHEMA) + + with pytest.raises(SystemExit) as exc_info: + app(["refresh", "--base-url", "http://alice:hunter2@api.test"]) + + message = str(exc_info.value) + assert "hunter2" not in message + assert "@api.test" in message + + def test_nested_refs_are_resolved_for_simple_array_body_fields(monkeypatch): schema = { "openapi": "3.1.0", @@ -1725,9 +1756,48 @@ def fake_request(method, url, **kwargs): assert json.loads(requests[-1]["content"].decode()) == {"names": ["alice"]} -def test_http_error_prints_body_and_exits_with_status(monkeypatch, capsys): +@pytest.mark.parametrize( + ("status_code", "reason", "body", "expected_detail"), + [ + (401, "Unauthorized", "not authorized", "not authorized"), + # JSON bodies are re-serialized by the structured redaction pass, so the + # detail is asserted on the normalized field rather than the raw bytes. + (404, "Not Found", '{"detail": "missing"}', '"detail": "missing"'), + (500, "Internal Server Error", '{"detail": "boom"}', '"detail": "boom"'), + (512, "", "unused status multiple of 256", "unused status multiple of 256"), + ], +) +def test_dynamic_operation_http_error_is_clean_one_line_nonzero( + monkeypatch, capsys, status_code, reason, body, expected_detail +): + def fake_request(method, url, **kwargs): + return text_response(method, url, body, status_code=status_code) + + monkeypatch.setattr(api_cli.httpx, "request", fake_request) + app = api_cli.build_app(SCHEMA) + + with pytest.raises(SystemExit) as exc_info: + app(["pipeline-runs", "list", "--base-url", "http://api.test"]) + + # Non-zero exit that is never the raw HTTP status: exit codes are 8-bit, so a + # status would truncate (404 -> 148) and multiples of 256 would report success. + code = exc_info.value.code + assert code not in (0, None, status_code) + message = str(code) + assert "\n" not in message + assert f"HTTP {status_code}" in message + if reason: + assert reason in message + assert "GET http://api.test/api/pipeline_runs/" in message + assert expected_detail in message + assert capsys.readouterr().err == "" + + +def test_dynamic_operation_http_error_bounds_huge_multiline_body(monkeypatch): + body = "detail:\n\n" + "\t".join("x" * 200 for _ in range(50)) + def fake_request(method, url, **kwargs): - return text_response(method, url, "not authorized", status_code=401) + return text_response(method, url, body, status_code=500) monkeypatch.setattr(api_cli.httpx, "request", fake_request) app = api_cli.build_app(SCHEMA) @@ -1735,11 +1805,58 @@ def fake_request(method, url, **kwargs): with pytest.raises(SystemExit) as exc_info: app(["pipeline-runs", "list", "--base-url", "http://api.test"]) - assert exc_info.value.code == 401 - assert "not authorized" in capsys.readouterr().err + message = str(exc_info.value.code) + assert "\n" not in message and "\t" not in message + assert message.endswith("…") + assert len(message) < 700 + +def test_dynamic_operation_http_error_redacts_reflected_body_secrets(monkeypatch, capsys): + body = '{"detail": "invalid token", "credential": "BODYSECRET", "api_key": "AK123"}' -def test_network_error_message_includes_url(monkeypatch): + def fake_request(method, url, **kwargs): + return text_response(method, url, body, status_code=401) + + monkeypatch.setattr(api_cli.httpx, "request", fake_request) + app = api_cli.build_app(SCHEMA) + + with pytest.raises(SystemExit) as exc_info: + app(["pipeline-runs", "list", "--base-url", "http://api.test"]) + + message = str(exc_info.value.code) + assert "BODYSECRET" not in message + assert "AK123" not in message + assert "" in message + assert "invalid token" in message + assert capsys.readouterr().err == "" + + +def test_dynamic_operation_http_error_redacts_url_credentials(monkeypatch): + def fake_request(method, url, **kwargs): + return text_response(method, url, "denied", status_code=403) + + monkeypatch.setattr(api_cli.httpx, "request", fake_request) + app = api_cli.build_app(SCHEMA) + + with pytest.raises(SystemExit) as exc_info: + app( + [ + "pipeline-runs", + "list", + "--filter", + "active", + "--base-url", + "http://alice:hunter2@api.test", + ] + ) + + message = str(exc_info.value.code) + assert "hunter2" not in message + assert "alice" not in message + assert "@api.test" in message + + +def test_network_error_message_includes_url_and_redacts_credentials(monkeypatch): def fake_request(method, url, **kwargs): request = httpx.Request(method, url) raise httpx.ConnectError("connection refused", request=request) @@ -1748,11 +1865,51 @@ def fake_request(method, url, **kwargs): app = api_cli.build_app(SCHEMA) with pytest.raises(SystemExit) as exc_info: - app(["pipeline-runs", "list", "--base-url", "http://api.test"]) + app(["pipeline-runs", "list", "--base-url", "http://alice:pw@api.test"]) message = str(exc_info.value) - assert "http://api.test/api/pipeline_runs/" in message + assert "\n" not in message + assert message.startswith("Failed to reach GET ") + assert "/api/pipeline_runs/" in message assert "connection refused" in message + assert "pw" not in message + + +def test_dynamic_operation_connection_refused_real_socket(monkeypatch): + # A refused connection against a closed local port exercises the real httpx + # transport (no monkeypatched request) and must still exit cleanly. + import socket + + probe = socket.socket() + probe.bind(("127.0.0.1", 0)) + refused_port = probe.getsockname()[1] + probe.close() + + app = api_cli.build_app(SCHEMA) + + with pytest.raises(SystemExit) as exc_info: + app(["pipeline-runs", "list", "--base-url", f"http://127.0.0.1:{refused_port}"]) + + message = str(exc_info.value) + assert exc_info.value.code not in (0, None) + assert "\n" not in message + assert message.startswith("Failed to reach GET ") + assert f"127.0.0.1:{refused_port}" in message + + +def test_dynamic_operation_timeout_is_clean(monkeypatch): + def fake_request(method, url, **kwargs): + raise httpx.ConnectTimeout("timed out", request=httpx.Request(method, url)) + + monkeypatch.setattr(api_cli.httpx, "request", fake_request) + app = api_cli.build_app(SCHEMA) + + with pytest.raises(SystemExit) as exc_info: + app(["pipeline-runs", "list", "--base-url", "http://api.test"]) + + message = str(exc_info.value) + assert "\n" not in message + assert "connection timed out" in message def test_custom_headers_apply_to_schema_fetch_and_generated_requests(monkeypatch): diff --git a/tests/test_api_transport.py b/tests/test_api_transport.py index c3ea83a..ea79cd9 100644 --- a/tests/test_api_transport.py +++ b/tests/test_api_transport.py @@ -1,13 +1,22 @@ +import json +import time +import urllib.parse from types import SimpleNamespace import httpx import pytest from tangle_cli.api_transport import ( + _MAX_BACKEND_DETAIL_CHARS, + _MAX_JSON_DEPTH, _redact_headers, build_operation_request, default_base_url, + describe_request_error, + format_http_status_error, + format_request_error, request_operation, + sanitize_url, tangle_verbose_enabled, ) @@ -241,3 +250,1463 @@ def test_build_operation_request_allows_relative_paths() -> None: assert url == "https://api.tangle.test/api/components/{id}" assert headers["Authorization"] == "Bearer secret-token" assert content is None + + +def test_sanitize_url_strips_userinfo() -> None: + sanitized = sanitize_url("https://alice:hunter2@api.tangle.test:8443/api/x?limit=5") + + assert "hunter2" not in sanitized + assert "alice" not in sanitized + assert sanitized == "https://@api.tangle.test:8443/api/x?limit=5" + + +@pytest.mark.parametrize( + "param", + ["token", "access_token", "api_key", "signature", "X-Amz-Signature", "sig"], +) +def test_sanitize_url_redacts_credential_query_params(param: str) -> None: + sanitized = sanitize_url(f"https://api.tangle.test/api/x?{param}=SECRETVALUE&limit=5") + + assert "SECRETVALUE" not in sanitized + assert "" in sanitized + assert "limit=5" in sanitized + + +def test_sanitize_url_redacts_presigned_url_signature() -> None: + signed = ( + "https://bucket.s3.amazonaws.com/object?" + "X-Amz-Credential=AKIA_LEAK&X-Amz-Signature=DEADBEEFSIG&X-Amz-Expires=900" + ) + + sanitized = sanitize_url(signed) + + assert "AKIA_LEAK" not in sanitized + assert "DEADBEEFSIG" not in sanitized + assert "bucket.s3.amazonaws.com" in sanitized + + +def test_sanitize_url_preserves_plain_url() -> None: + assert sanitize_url("http://api.test/api/pipeline_runs/") == "http://api.test/api/pipeline_runs/" + + +@pytest.mark.parametrize( + ("url", "expected"), + [ + ("https://[2001:db8::1]/api/x", "https://[2001:db8::1]/api/x"), + ("https://[2001:db8::1]:8443/api/x", "https://[2001:db8::1]:8443/api/x"), + ( + "https://alice:hunter2@[2001:db8::1]:8443/api/x", + "https://@[2001:db8::1]:8443/api/x", + ), + ( + "https://alice:hunter2@[2001:db8::1]:8443/api/x?token=SECRET&limit=5", + "https://@[2001:db8::1]:8443/api/x?token=&limit=5", + ), + ], +) +def test_sanitize_url_rebrackets_ipv6_literals(url: str, expected: str) -> None: + sanitized = sanitize_url(url) + + assert sanitized == expected + assert "hunter2" not in sanitized + assert "SECRET" not in sanitized + + +def test_format_http_status_error_is_one_line_with_status_method_and_url() -> None: + request = httpx.Request("GET", "https://alice:pw@api.tangle.test/api/x") + response = httpx.Response(404, text='{"detail": "missing"}', request=request) + exc = httpx.HTTPStatusError("client error", request=request, response=response) + + message = format_http_status_error(exc) + + assert "\n" not in message + assert "HTTP 404 Not Found for GET" in message + assert "pw" not in message + assert "missing" in message + + +def test_format_http_status_error_bounds_and_normalizes_detail() -> None: + request = httpx.Request("POST", "https://api.tangle.test/api/x") + body = "line one\n\n line two\t" + "A" * 5000 + response = httpx.Response(500, text=body, request=request) + exc = httpx.HTTPStatusError("server error", request=request, response=response) + + message = format_http_status_error(exc) + + assert "\n" not in message and "\t" not in message + assert "line one line two" in message + assert message.endswith("…") + assert len(message) < _MAX_BACKEND_DETAIL_CHARS + 200 + + +def test_format_http_status_error_redacts_reflected_json_secrets() -> None: + request = httpx.Request("POST", "https://api.tangle.test/api/x") + body = '{"detail": "invalid credential", "credential": "BODYSECRET", "token": "abc123"}' + response = httpx.Response(401, text=body, request=request) + exc = httpx.HTTPStatusError("unauthorized", request=request, response=response) + + message = format_http_status_error(exc) + + assert "\n" not in message + assert "BODYSECRET" not in message + assert "abc123" not in message + assert "" in message + # Non-sensitive detail stays visible so the backend message is still useful. + assert "invalid credential" in message + + +def test_format_http_status_error_preserves_non_json_detail() -> None: + request = httpx.Request("POST", "https://api.tangle.test/api/x") + response = httpx.Response(502, text="upstream unavailable", request=request) + exc = httpx.HTTPStatusError("bad gateway", request=request, response=response) + + message = format_http_status_error(exc) + + assert message.endswith("upstream unavailable") + + +def _status_error(body: str | bytes, *, status: int = 401) -> httpx.HTTPStatusError: + request = httpx.Request("POST", "https://api.tangle.test/api/x") + kwargs: dict[str, object] = {"request": request} + if isinstance(body, bytes): + kwargs["content"] = body + else: + kwargs["text"] = body + response = httpx.Response(status, **kwargs) # type: ignore[arg-type] + return httpx.HTTPStatusError("error", request=request, response=response) + + +@pytest.mark.parametrize( + ("body", "leaked", "expected"), + [ + ("token=sk-live-0123456789ABCDEF&foo=bar", "sk-live-0123456789ABCDEF", "token="), + ("password: hunter2secretvalue", "hunter2secretvalue", "password: "), + ( + "api_key=AKIA0123456789ABCD failed", + "AKIA0123456789ABCD", + "api_key=", + ), + ("credential=BODYSECRET&foo=bar", "BODYSECRET", "credential="), + ("X-Amz-Signature=DEADBEEF0123456789 expired", "DEADBEEF0123456789", "X-Amz-Signature="), + ("oauth_token=OAUTH0123456&keep=1", "OAUTH0123456", "oauth_token="), + ("Signature: PLAINSIG12345", "PLAINSIG12345", "Signature: "), + ("awsaccesskeyid=AKIAX0123&googleaccessid=GOOGLE0123", "AKIAX0123", "awsaccesskeyid="), + ], +) +def test_format_http_status_error_redacts_secrets_in_non_json_body( + body: str, leaked: str, expected: str +) -> None: + # Form, plain-text, and HTML bodies bypass the structured JSON path, so the + # free-text assignment fallback is what keeps a reflected secret off stderr. + message = format_http_status_error(_status_error(body)) + + assert leaked not in message + assert expected in message + + +@pytest.mark.parametrize( + ("body", "leaked"), + [ + # Truncated/unparseable JSON never reaches the structured redactor, so the + # quoted assignments must still be scrubbed by the text fallback. + ('{"detail": "bad", "token": "abc123", ', "abc123"), + ('{"password": "secret"', "secret"), + ('{"a": {"api_key": "AKIA0123"} ', "AKIA0123"), + ], +) +def test_format_http_status_error_redacts_secrets_in_malformed_json_body( + body: str, leaked: str +) -> None: + message = format_http_status_error(_status_error(body)) + + assert leaked not in message + assert "" in message + + +def test_format_http_status_error_redacts_secrets_in_undecodable_body() -> None: + # Invalid UTF-8 is decoded with replacement rather than raising; the surviving + # text must still be scrubbed. + message = format_http_status_error(_status_error(b"\xff\xfetoken=SECRET0123456789xyz")) + + assert "SECRET0123456789xyz" not in message + assert "token=" in message + + +@pytest.mark.parametrize( + ("body", "leaked"), + [ + # Secret past the bound: truncation must not be what hides it. + ("A" * (_MAX_BACKEND_DETAIL_CHARS + 200) + " token=TAILSECRET0123456789", "TAILSECRET0123456789"), + # Secret before the bound in an oversized body: redaction still applies. + ("token=HEADSECRET0123456789 " + "B" * (_MAX_BACKEND_DETAIL_CHARS + 200), "HEADSECRET0123456789"), + ], +) +def test_format_http_status_error_redacts_before_truncation(body: str, leaked: str) -> None: + message = format_http_status_error(_status_error(body, status=500)) + + assert leaked not in message + assert len(message) < _MAX_BACKEND_DETAIL_CHARS + 200 + + +def test_format_http_status_error_preserves_non_sensitive_non_json_detail() -> None: + # Sensitive words that are not the name of an assignment, and assignments + # whose name is not sensitive, stay intact so backend diagnostics stay useful. + message = format_http_status_error( + _status_error( + "Invalid credentials supplied; authentication required;" + " status=failed detail=useful page=2 attempt=3/5", + status=502, + ) + ) + + assert message.endswith( + ": Invalid credentials supplied; authentication required;" + " status=failed detail=useful page=2 attempt=3/5" + ) + + +def test_sensitive_field_name_redacts_regardless_of_how_the_value_looks() -> None: + # A short, purely alphabetic value is still a credential -- ``password: + # swordfish`` must not survive on the grounds that it reads like a word. The + # field name decides, so the surrounding prose is the only thing preserved. + message = format_http_status_error( + _status_error("rejected password: swordfish for user alice", status=403) + ) + + assert "swordfish" not in message + assert message.endswith(": rejected password: for user alice") + + +@pytest.mark.parametrize( + ("body", "leaked", "expected"), + [ + # Affixed credential names: the sensitive word is welded to a prefix or + # suffix, so nothing sits on a word boundary for an alternation to catch. + ("access_token=ghp_AAAABBBBCCCCDDDD", "ghp_AAAABBBBCCCCDDDD", "access_token="), + ("refresh_token=rt_zzzzzzzz9999", "rt_zzzzzzzz9999", "refresh_token="), + ("client_secret=cs_abcdefgh12345678", "cs_abcdefgh12345678", "client_secret="), + ("id_token=eyJhbGciOiJIUzI1NiJ9.e30.x", "eyJhbGciOiJIUzI1NiJ9.e30.x", "id_token="), + ("session_token=st_112233445566", "st_112233445566", "session_token="), + ("api_secret=sk_9988776655", "sk_9988776655", "api_secret="), + ("my_api_key=mk_5544332211", "mk_5544332211", "my_api_key="), + ("user.password=hunter2", "hunter2", "user.password="), + ("X-Refresh-Token: rt_998877", "rt_998877", "X-Refresh-Token: "), + ], +) +def test_affixed_sensitive_field_names_are_redacted( + body: str, leaked: str, expected: str +) -> None: + message = format_http_status_error(_status_error(body)) + + assert leaked not in message + assert expected in message + + +@pytest.mark.parametrize( + ("body", "leaked", "expected"), + [ + # Short, purely alphabetic values carry no digits or base64 punctuation, + # so a "does this look opaque?" test on the value alone misses them. + ("password:swordfish", "swordfish", "password:"), + ("token:abcdefgh", "abcdefgh", "token:"), + ("secret: hunter", "hunter", "secret: "), + ('{"api_key": "abc"', "abc", '"api_key": "'), + ], +) +def test_short_alphabetic_values_are_redacted(body: str, leaked: str, expected: str) -> None: + message = format_http_status_error(_status_error(body)) + + assert leaked not in message + assert expected in message + + +@pytest.mark.parametrize( + ("body", "leaked", "expected"), + [ + # The credential follows an auth scheme rather than the separator, so the + # scheme name is kept as a diagnostic and only the credential is cut. + ( + "Authorization: Bearer abcdefghijklmnop", + "abcdefghijklmnop", + "Authorization: Bearer ", + ), + ('invalid header "Authorization: Bearer sk-XYZ"', "sk-XYZ", "Bearer "), + ("authorization=Basic YWxpY2U6c2VjcmV0", "YWxpY2U6c2VjcmV0", "Basic "), + # No sensitive field name at all -- the scheme is the only signal. + ("rejected header: Bearer sk-live-0123456789", "sk-live-0123456789", "Bearer "), + ("proxy said Basic dXNlcjpwdw==", "dXNlcjpwdw==", "Basic "), + ], +) +def test_authorization_scheme_credentials_are_redacted( + body: str, leaked: str, expected: str +) -> None: + message = format_http_status_error(_status_error(body)) + + assert leaked not in message + assert expected in message + + +@pytest.mark.parametrize( + ("body", "leaked"), + [ + # Well-formed JSON whose *string leaves* quote a sensitive assignment + # back at us. Key-by-key redaction never inspects the leaf text. + ('{"detail": "invalid token=ghp_SECRET1234 supplied"}', "ghp_SECRET1234"), + ('{"errors": [{"msg": "password=hunter2 rejected"}]}', "hunter2"), + ('{"a": {"b": {"c": "access_token=at_9988776655 expired"}}}', "at_9988776655"), + ('"Authorization: Bearer sk-0123456789"', "sk-0123456789"), + ('{"detail": "bad client_secret=cs_5544332211"}', "cs_5544332211"), + ], +) +def test_reflected_assignments_inside_json_string_leaves_are_redacted( + body: str, leaked: str +) -> None: + message = format_http_status_error(_status_error(body)) + + assert leaked not in message + assert "" in message + + +def test_json_string_leaf_scrub_keeps_surrounding_prose() -> None: + body = '{"detail": "invalid token=ghp_SECRET1234 supplied for run abc123"}' + + message = format_http_status_error(_status_error(body)) + + assert "ghp_SECRET1234" not in message + assert "invalid token= supplied for run abc123" in message + + +@pytest.mark.parametrize( + "body", + [ + # Non-sensitive assignments, sensitive words used as prose rather than as + # a field name, and values that merely follow a sensitive-looking word. + "page=2 of 5 results", + "token count: 42 exceeded", + "status=failed detail=useful", + "Basic authentication failed for this endpoint", + "run abc123 not found", + "retry after 30 seconds; attempt=3", + ], +) +def test_non_sensitive_diagnostics_survive_redaction(body: str) -> None: + message = format_http_status_error(_status_error(body, status=502)) + + assert message.endswith(f": {body}") + assert "" not in message + + +@pytest.mark.parametrize( + ("body", "leaked"), + [ + # A harmless outer field must not consume the sensitive assignment nested + # in its value: declining a separator has to cost zero characters, or the + # inner name is never even looked at. + ('{"detail": "token=SECRET0123456789"', "SECRET0123456789"), + ('{"detail": "access_token=SECRET0123456789"', "SECRET0123456789"), + ('{"msg": "d", "err": "password:swordfish"', "swordfish"), + ("outer=inner_secret=SECRET0123456789", "SECRET0123456789"), + ], +) +def test_nested_assignment_under_harmless_outer_field_is_reached( + body: str, leaked: str +) -> None: + message = format_http_status_error(_status_error(body)) + + assert leaked not in message + assert "" in message + + +@pytest.mark.parametrize( + "body", + [ + # Aligned/log-style padding between the name and its separator. + "api_key : SECRET0123456789", + 'api_key" : "SECRET0123456789', + "api_key\t=\tSECRET0123456789", + ], +) +def test_padding_between_field_name_and_separator_still_redacts(body: str) -> None: + message = format_http_status_error(_status_error(body)) + + assert "SECRET0123456789" not in message + assert "" in message + + +def test_ambiguous_bare_scheme_needs_a_long_credential() -> None: + # "Token"/"OAuth" are ordinary words, so a bare occurrence redacts only when + # what follows is long enough to be a credential rather than an identifier. + redacted = format_http_status_error(_status_error("rejected: Token SECRET0123456789xyz")) + prose = format_http_status_error(_status_error("Token 12345 expired", status=502)) + + assert "SECRET0123456789xyz" not in redacted + assert "Token " in redacted + assert prose.endswith(": Token 12345 expired") + + +def test_url_in_detail_is_not_swallowed_as_an_assignment() -> None: + # ``https:`` reads like ``key: value``; the scheme must not consume the query + # string, or the credential inside it would never be scanned. + body = "callback https://cb.tangle.test/hook?token=sk-0123456789&page=2 rejected" + + message = format_http_status_error(_status_error(body)) + + assert "sk-0123456789" not in message + assert "https://cb.tangle.test/hook?token=&page=2" in message + + +@pytest.mark.parametrize( + ("body", "leaked", "expected"), + [ + # A credential can ride in a URL or in bare userinfo rather than in an + # assignment, and a reflected body carries those just as an exception does. + ( + "callback to https://alice:hunter2@cb.tangle.test/hook failed", + "hunter2", + "https://@cb.tangle.test/hook", + ), + ( + "proxy alice:hunter2@proxy.internal:8080 refused", + "hunter2", + "proxy @proxy.internal:8080 refused", + ), + ( + "upstream https://bucket.s3.test/o?X-Amz-Signature=DEADBEEFSIG expired", + "DEADBEEFSIG", + "X-Amz-Signature=", + ), + ], +) +def test_urls_and_userinfo_in_body_text_are_scrubbed( + body: str, leaked: str, expected: str +) -> None: + message = format_http_status_error(_status_error(body, status=502)) + + assert leaked not in message + assert "alice" not in message + assert expected in message + + +@pytest.mark.parametrize( + ("body", "leaked"), + [ + ('{"detail": "fetch https://alice:hunter2@storage.test/o failed"}', "hunter2"), + ('{"detail": "proxy alice:hunter2@proxy.internal refused"}', "hunter2"), + ('{"a": {"b": [{"msg": "https://alice:hunter2@h.test/x"}]}}', "hunter2"), + ( + '{"a": {"b": {"detail": "https://s.test/o?X-Amz-Signature=DEADBEEFSIG"}}}', + "DEADBEEFSIG", + ), + ], +) +def test_urls_and_userinfo_in_nested_json_leaves_are_scrubbed(body: str, leaked: str) -> None: + # Key-by-key redaction never inspects leaf text, so the leaf scrub is the only + # thing standing between a reflected presigned URL and stderr. + message = format_http_status_error(_status_error(body)) + + assert leaked not in message + assert "" in message + + +_CREDENTIAL_FIELD_NAMES = [ + "access_token", + "api_key", + "api_secret", + "authorization", + "client_secret", + "cookie", + "credential", + "id_token", + "my_api_key", + "password", + "private_key", + "refresh_token", + "session_token", + "signed_url", + "token", + "user.password", + "X-Api-Key", + "X-Gateway-Auth", + "X-Refresh-Token", +] +# Names that merely contain a credential word. Substring matching redacts every +# one of them; matching the trailing token of the name does not. +_CREDENTIAL_LOOKALIKE_FIELD_NAMES = [ + "auth_method", + "completion_tokens", + "function_signature", + "max_tokens", + "oauth_provider", + "password_policy", + "prompt_tokens", + "secretary_email", + "signature_version", + "token_count", + "tokenizer", + "X-Author", +] + + +@pytest.mark.parametrize("name", _CREDENTIAL_FIELD_NAMES) +def test_credential_field_names_are_redacted(name: str) -> None: + json_message = format_http_status_error(_status_error(f'{{"{name}": "SECRETVALUE"}}')) + text_message = format_http_status_error(_status_error(f"{name}=SECRETVALUE")) + + assert "SECRETVALUE" not in json_message + assert "SECRETVALUE" not in text_message + + +@pytest.mark.parametrize("name", _CREDENTIAL_LOOKALIKE_FIELD_NAMES) +def test_credential_lookalike_field_names_keep_their_values(name: str) -> None: + json_message = format_http_status_error(_status_error(f'{{"{name}": "PLAINVALUE"}}')) + text_message = format_http_status_error(_status_error(f"{name}=PLAINVALUE", status=502)) + + assert "PLAINVALUE" in json_message + assert text_message.endswith(f": {name}=PLAINVALUE") + + +def test_credential_lookalike_header_names_keep_their_values() -> None: + headers = {name: "PLAINVALUE" for name in _CREDENTIAL_LOOKALIKE_FIELD_NAMES} + + assert _redact_headers(headers) == headers + + +def test_credential_headers_are_redacted() -> None: + redacted = _redact_headers({name: "SECRETVALUE" for name in _CREDENTIAL_FIELD_NAMES}) + + assert set(redacted.values()) == {""} + + +def test_signature_is_a_credential_in_a_query_but_not_as_a_body_field() -> None: + # A presigned grant lives in the query string, so ``signature`` there is the + # credential itself; a body field named ``function_signature`` is not. + assert "SIGVALUE" not in sanitize_url("https://api.test/x?signature=SIGVALUE") + assert ( + sanitize_url("https://api.test/x?function_signature=f(x)") + == "https://api.test/x?function_signature=f%28x%29" + ) + + +@pytest.mark.parametrize( + "body", + [ + # Adversarial shapes with no exponential structure, all bounded classes. + "=" * 20000, + "a" * 20000 + "=", + ("token" + "_" * 200 + "=abc ") * 2000, + ("Bearer " + "a" * 100 + " ") * 2000, + '{"k": "' + ("password=x " * 5000) + '"}', + ], +) +def test_redaction_stays_linear_on_adversarial_bodies(body: str) -> None: + start = time.perf_counter() + message = format_http_status_error(_status_error(body, status=500)) + elapsed = time.perf_counter() - start + + assert elapsed < 2.0 + assert len(message) < _MAX_BACKEND_DETAIL_CHARS + 200 + + +def test_non_json_redaction_is_linear_on_large_bodies() -> None: + # The assignment pattern uses only bounded/negated classes with no nested + # quantifier, so a large adversarial body must not trigger catastrophic + # backtracking. + body = ("token=" + "a" * 50 + " ") * 20000 + "password: " + "x" * 40 + start = time.perf_counter() + message = format_http_status_error(_status_error(body)) + elapsed = time.perf_counter() - start + + assert elapsed < 2.0 + assert "aaaaaaaaaa" not in message + + +@pytest.mark.parametrize( + "body", + [ + # Shapes that make a ``scheme://`` or ``user:pass@`` regex re-scan the same + # run once per starting offset. The scans anchor on ``://`` and ``@`` + # instead, so each character is visited a bounded number of times. + "a" * 200000 + "=", + "a@" * 100000, + "a:" * 100000, + "://" * 60000, + "https://u:p@h.test/x " * 10000, + ], +) +def test_url_and_userinfo_scrubbing_stays_linear(body: str) -> None: + start = time.perf_counter() + message = format_http_status_error(_status_error(body, status=500)) + elapsed = time.perf_counter() - start + + assert elapsed < 2.0 + assert len(message) < _MAX_BACKEND_DETAIL_CHARS + 200 + + +def test_schema_fetch_no_body_policy_survives_text_redaction() -> None: + # The /openapi.json path omits the body entirely rather than relying on + # redaction, so a reflected credential must not appear even as a marker. + request = httpx.Request("GET", "https://api.tangle.test/openapi.json") + response = httpx.Response(401, text="token=REFLECTED0123456789", request=request) + exc = httpx.HTTPStatusError("unauthorized", request=request, response=response) + + message = format_http_status_error(exc, include_detail=False) + + assert message == "HTTP 401 Unauthorized for GET https://api.tangle.test/openapi.json" + assert "REFLECTED0123456789" not in message + assert "" not in message + + +def test_format_http_status_error_can_omit_detail() -> None: + request = httpx.Request("GET", "https://api.tangle.test/openapi.json") + response = httpx.Response(401, text="secret-token", request=request) + exc = httpx.HTTPStatusError("unauthorized", request=request, response=response) + + message = format_http_status_error(exc, include_detail=False) + + assert message == "HTTP 401 Unauthorized for GET https://api.tangle.test/openapi.json" + assert "secret-token" not in message + + +@pytest.mark.parametrize( + ("exc", "expected"), + [ + (httpx.ConnectError("connection refused"), "connection failed: connection refused"), + (httpx.ConnectTimeout("timed out"), "connection timed out"), + (httpx.ReadTimeout("slow"), "read timed out"), + (httpx.PoolTimeout("busy"), "connection pool timed out"), + (httpx.ProxyError("bad proxy"), "proxy error: bad proxy"), + ( + httpx.ConnectError("[SSL: CERTIFICATE_VERIFY_FAILED] self-signed certificate"), + "TLS error", + ), + ], +) +def test_describe_request_error_is_actionable(exc: httpx.RequestError, expected: str) -> None: + assert expected in describe_request_error(exc) + + +def test_format_request_error_is_one_line_and_redacts_url() -> None: + request = httpx.Request("GET", "https://alice:pw@api.tangle.test/api/x?token=SECRET") + exc = httpx.ConnectError("connection refused", request=request) + + message = format_request_error(exc) + + assert "\n" not in message + assert message.startswith("Failed to reach GET ") + assert "pw" not in message + assert "SECRET" not in message + assert "connection refused" in message + + +@pytest.mark.parametrize( + ("exc", "secrets"), + [ + ( + httpx.ProxyError( + "unable to connect to proxy http://proxyuser:proxypass@proxy.internal:8080" + ), + ["proxyuser", "proxypass"], + ), + ( + httpx.ConnectError( + "connection failed while fetching " + "https://bucket.s3.amazonaws.com/o?X-Amz-Signature=DEADBEEFSIG&X-Amz-Expires=900" + ), + ["DEADBEEFSIG"], + ), + ( + httpx.ConnectError("refused for user:secretpw@10.0.0.5"), + ["secretpw"], + ), + ], +) +def test_describe_request_error_scrubs_embedded_secrets( + exc: httpx.RequestError, secrets: list[str] +) -> None: + message = describe_request_error(exc) + + assert "\n" not in message + assert "" in message + for secret in secrets: + assert secret not in message + + +def test_describe_request_error_preserves_benign_diagnostics() -> None: + refused = describe_request_error(httpx.ConnectError("[Errno 111] Connection refused")) + assert "connection failed" in refused + assert "[Errno 111] Connection refused" in refused + + tls = describe_request_error( + httpx.ConnectError("[SSL: CERTIFICATE_VERIFY_FAILED] self-signed certificate") + ) + assert "TLS error" in tls + assert "CERTIFICATE_VERIFY_FAILED" in tls + + +def test_format_request_error_scrubs_secrets_in_exception_text() -> None: + request = httpx.Request("POST", "https://api.tangle.test/api/x") + exc = httpx.ProxyError( + "proxy http://proxyuser:proxypass@proxy.internal:8080 rejected", request=request + ) + + message = format_request_error(exc) + + assert "\n" not in message + assert "proxyuser" not in message + assert "proxypass" not in message + assert message.startswith("Failed to reach POST https://api.tangle.test/api/x:") + + +_CAMEL_CASE_CREDENTIAL_NAMES = [ + "accessToken", + "AccessToken", + "refreshToken", + "idToken", + "sessionToken", + "authToken", + "oauth2Token", + "clientSecret", + "ClientSecret", + "apiSecret", + "apiKey", + "ApiKey", + "APIKey", + "XApiKey", + "myApiKey", + "privateKey", + "secretKey", + "userPassword", + "sessionCookie", +] + +_CAMEL_CASE_DIAGNOSTIC_NAMES = [ + "maxTokens", + "MaxTokens", + "promptTokens", + "completionTokens", + "tokenCount", + "TokenCount", + "tokenizer", + "Tokenizer", + "functionSignature", + "signatureVersion", + "passwordPolicy", + "authMethod", + "oauthProvider", + "secretaryEmail", +] + + +@pytest.mark.parametrize("name", _CAMEL_CASE_CREDENTIAL_NAMES) +def test_camel_case_credential_names_are_redacted_everywhere(name: str) -> None: + body = json.dumps({name: "PLAINSECRET"}) + + assert "PLAINSECRET" not in format_http_status_error(_status_error(body)) + assert _redact_headers({name: "PLAINSECRET"})[name] == "" + assert "PLAINSECRET" not in format_http_status_error(_status_error(f"{name}=PLAINSECRET")) + + +@pytest.mark.parametrize("name", _CAMEL_CASE_DIAGNOSTIC_NAMES) +def test_camel_case_diagnostic_names_keep_their_values(name: str) -> None: + body = json.dumps({name: "KEEPME"}) + + assert "KEEPME" in format_http_status_error(_status_error(body)) + assert _redact_headers({name: "KEEPME"})[name] == "KEEPME" + assert "KEEPME" in format_http_status_error(_status_error(f"{name}=KEEPME")) + + +@pytest.mark.parametrize("depth", [1, 8, 32, 900, 1000, 5000, 9000]) +def test_deeply_nested_json_body_stays_one_bounded_line(depth: int) -> None: + message = format_http_status_error(_status_error("[" * depth + "1" + "]" * depth, status=500)) + + assert "\n" not in message + assert len(message) < _MAX_BACKEND_DETAIL_CHARS + 200 + + +@pytest.mark.parametrize("depth", [8, 900, 3000]) +def test_secret_nested_beyond_reach_is_never_emitted(depth: int) -> None: + node: dict[str, object] = {"access_token": "DEEPSECRET"} + for _ in range(depth): + node = {"a": node} + + message = format_http_status_error(_status_error(json.dumps(node), status=500)) + + assert "DEEPSECRET" not in message + assert "\n" not in message + + +def test_nesting_past_the_depth_bound_is_replaced_wholesale() -> None: + node: dict[str, object] = {"leaf": "visible-detail"} + for _ in range(_MAX_JSON_DEPTH + 5): + node = {"a": node} + + message = format_http_status_error(_status_error(json.dumps(node), status=500)) + + assert "visible-detail" not in message + assert "nesting too deep" in message + + +def test_shallow_json_is_unaffected_by_the_depth_bound() -> None: + body = json.dumps({"errors": [{"loc": ["body", "name"], "msg": "field required"}]}) + + message = format_http_status_error(_status_error(body)) + + assert "field required" in message + assert "nesting too deep" not in message + + +@pytest.mark.parametrize( + "text", + [ + ":hunter2@proxy.internal", + "proxy :hunter2@proxy.internal refused", + "https://:hunter2@proxy.internal/x", + ], +) +def test_empty_username_userinfo_is_redacted(text: str) -> None: + message = describe_request_error(httpx.ConnectError(text)) + + assert "hunter2" not in message + assert "proxy.internal" in message + + +def test_sanitize_url_keeps_non_credential_presigned_parameters() -> None: + signed = ( + "https://bucket.s3.test/object?X-Amz-Algorithm=AWS4-HMAC-SHA256" + "&X-Amz-Date=20240101T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host" + "&X-Amz-Credential=AKIA_LEAK&X-Amz-Signature=DEADBEEFSIG" + "&X-Amz-Security-Token=SESSIONTOKEN" + ) + + sanitized = sanitize_url(signed) + + assert "AKIA_LEAK" not in sanitized + assert "DEADBEEFSIG" not in sanitized + assert "SESSIONTOKEN" not in sanitized + assert "X-Amz-Algorithm=AWS4-HMAC-SHA256" in sanitized + assert "X-Amz-Date=20240101T000000Z" in sanitized + assert "X-Amz-Expires=900" in sanitized + assert "X-Amz-SignedHeaders=host" in sanitized + + +_ACCESS_KEY_NAMES = [ + "access_key_id", + "access-key-id", + "accessKeyId", + "AccessKeyID", + "AccessKeyIdentifier", + "aws_access_key_id", + "AwsAccessKeyId", + "AWSAccessKeyId", + "google_access_id", + "GoogleAccessId", + "access_id", + "accessId", + "api_key_id", + "apiKeyId", + "secret_key", + "secretKey", + "private_key", + "privateKey", +] + +_ACCESS_KEY_LOOKALIKES = [ + "access_key_id_format", + "accessKeyIdFormat", + "access_key_id_prefix", + "private_key_path", + "privateKeyPath", + "signature_version", + "signatureVersion", + "function_signature", + "session_id", + "sessionId", + "request_id", + "requestId", + "user_id", + "userId", + "key_id", + "keyId", + "correlation_id", + "traceId", +] + + +@pytest.mark.parametrize("name", _ACCESS_KEY_NAMES) +def test_access_key_identifiers_are_redacted_everywhere(name: str) -> None: + body = json.dumps({"error": {name: "AKIA_LEAK"}}) + + assert "AKIA_LEAK" not in format_http_status_error(_status_error(body)) + assert _redact_headers({name: "AKIA_LEAK"})[name] == "" + assert "AKIA_LEAK" not in format_http_status_error(_status_error(f"rejected {name}=AKIA_LEAK")) + assert "AKIA_LEAK" not in sanitize_url(f"https://api.tangle.test/x?{name}=AKIA_LEAK") + + +@pytest.mark.parametrize("name", _ACCESS_KEY_LOOKALIKES) +def test_access_key_lookalikes_keep_their_values(name: str) -> None: + body = json.dumps({"error": {name: "KEEPME"}}) + + assert "KEEPME" in format_http_status_error(_status_error(body)) + assert _redact_headers({name: "KEEPME"})[name] == "KEEPME" + assert "KEEPME" in format_http_status_error(_status_error(f"rejected {name}=KEEPME")) + assert "KEEPME" in sanitize_url(f"https://api.tangle.test/x?{name}=KEEPME") + + +def test_signed_url_body_value_keeps_its_location_and_loses_its_signature() -> None: + body = json.dumps( + {"signed_url": "https://bucket.s3.test/report.csv?X-Amz-Expires=900&X-Amz-Signature=SIGLEAK"} + ) + + message = format_http_status_error(_status_error(body)) + + assert "SIGLEAK" not in message + assert "bucket.s3.test/report.csv" in message + assert "X-Amz-Expires=900" in message + + +def test_signed_url_without_a_url_value_is_redacted_whole() -> None: + message = format_http_status_error(_status_error(json.dumps({"signed_url": "OPAQUEGRANT"}))) + + assert "OPAQUEGRANT" not in message + + +def test_signed_url_header_and_query_are_redacted_whole() -> None: + assert _redact_headers({"X-Signed-Url": "https://b.test/o?sig=SIGLEAK"})["X-Signed-Url"] == ( + "" + ) + sanitized = sanitize_url("https://api.tangle.test/x?signed_url=https%3A%2F%2Fb.test%2Fo") + assert "b.test" not in sanitized + + +_UNBROKEN_RUN_CREDENTIAL_NAMES = [ + "accesskeyid", + "ACCESSKEYID", + "accesstoken", + "accessid", + "accesskey", + "apikey", + "apisecret", + "apitoken", + "authtoken", + "bearertoken", + "clientsecret", + "idtoken", + "privatekey", + "refreshtoken", + "secretkey", + "sessionkey", + "sessiontoken", + "awsaccesskeyid", + "googleaccessid", +] + +_UNBROKEN_RUN_LOOKALIKES = [ + "accesskeyidformat", + "privatekeypath", + "maxtokens", + "sessionid", + "keyid", + "tokenizer", + "secretaryemail", + "signatureversion", + "functionsignature", + "requestid", + "traceid", + "identifier", +] + + +@pytest.mark.parametrize("name", _UNBROKEN_RUN_CREDENTIAL_NAMES) +def test_unbroken_lowercase_credential_runs_are_redacted_everywhere(name: str) -> None: + body = json.dumps({"error": {name: "AKIA_LEAK"}}) + + assert "AKIA_LEAK" not in format_http_status_error(_status_error(body)) + assert _redact_headers({name: "AKIA_LEAK"})[name] == "" + assert "AKIA_LEAK" not in format_http_status_error(_status_error(f"denied {name}=AKIA_LEAK")) + assert "AKIA_LEAK" not in sanitize_url(f"https://api.tangle.test/x?{name}=AKIA_LEAK") + + +@pytest.mark.parametrize("name", _UNBROKEN_RUN_LOOKALIKES) +def test_unbroken_lowercase_lookalikes_keep_their_values(name: str) -> None: + body = json.dumps({"error": {name: "KEEPME"}}) + + assert "KEEPME" in format_http_status_error(_status_error(body)) + assert _redact_headers({name: "KEEPME"})[name] == "KEEPME" + assert "KEEPME" in format_http_status_error(_status_error(f"bad {name}=KEEPME")) + assert "KEEPME" in sanitize_url(f"https://api.tangle.test/x?{name}=KEEPME") + + +@pytest.mark.parametrize( + "fragment", + [ + "access_token=FRAGSECRET&token_type=Bearer&expires_in=3600", + "accesstoken=FRAGSECRET", + "sig=FRAGSECRET", + "signature=FRAGSECRET", + "credential=FRAGSECRET", + "X-Amz-Signature=FRAGSECRET", + "id_token=FRAGSECRET&state=abc", + ], +) +def test_sanitize_url_redacts_credentials_in_the_fragment(fragment: str) -> None: + sanitized = sanitize_url(f"https://h.test/callback#{fragment}") + + assert "FRAGSECRET" not in sanitized + assert "h.test/callback" in sanitized + + +def test_sanitize_url_keeps_non_credential_fragment_parameters() -> None: + sanitized = sanitize_url("https://h.test/x#X-Amz-Signature=SIG&X-Amz-Expires=900&page=3") + + assert "SIG" not in sanitized + assert "X-Amz-Expires=900" in sanitized + assert "page=3" in sanitized + + +@pytest.mark.parametrize("fragment", ["section-2", "L42", "installation", ""]) +def test_sanitize_url_preserves_a_plain_anchor_fragment(fragment: str) -> None: + url = f"https://h.test/guide#{fragment}" if fragment else "https://h.test/guide" + + assert sanitize_url(url) == url + + +def test_sanitize_url_drops_a_fragment_it_cannot_structurally_parse() -> None: + sanitized = sanitize_url("https://h.test/x#state=ok;access_token=HIDDEN") + + assert "HIDDEN" not in sanitized + assert sanitized == "https://h.test/x#" + + +def test_request_url_fragment_credential_is_redacted_in_the_status_line() -> None: + request = httpx.Request("POST", "https://api.tangle.test/x#access_token=REQFRAG") + response = httpx.Response(401, request=request, text="denied") + + message = format_http_status_error( + httpx.HTTPStatusError("error", request=request, response=response) + ) + + assert "REQFRAG" not in message + + +def test_body_url_fragment_credential_is_redacted() -> None: + body = json.dumps({"detail": "retry at https://h.test/cb#access_token=FRAGSECRET"}) + + assert "FRAGSECRET" not in format_http_status_error(_status_error(body)) + + +@pytest.mark.parametrize( + "fragment", + [ + "alice:ANCHORSECRET@evil.test", + ":ANCHORSECRET@evil.test", + "https://alice:ANCHORSECRET@evil.test", + "https://alice:ANCHORSECRET@evil.test/path", + "go/to/alice:ANCHORSECRET@evil.test", + ], +) +def test_sanitize_url_strips_userinfo_from_an_anchor_fragment(fragment: str) -> None: + sanitized = sanitize_url(f"https://h.test/guide#{fragment}") + + assert "ANCHORSECRET" not in sanitized + assert sanitized.startswith("https://h.test/guide#") + + +@pytest.mark.parametrize("fragment", ["user@example.com", "a.b.c", "step-3/details"]) +def test_sanitize_url_keeps_an_anchor_without_userinfo(fragment: str) -> None: + url = f"https://h.test/guide#{fragment}" + + assert sanitize_url(url) == url + + +@pytest.mark.parametrize( + "value", + [ + "https%3A%2F%2Falice%3AVALSECRET%40evil.test", + "alice%3AVALSECRET%40evil.test", + "https://alice:VALSECRET@evil.test", + ], +) +def test_sanitize_url_strips_userinfo_from_a_non_credential_query_value(value: str) -> None: + sanitized = sanitize_url(f"https://h.test/x?page=3&next={value}") + + assert "VALSECRET" not in sanitized + assert "page=3" in sanitized + assert "evil.test" in sanitized + + +@pytest.mark.parametrize( + "value", + [ + "https%3A%2F%2Falice%3AVALSECRET%40evil.test", + "alice%3AVALSECRET%40evil.test", + ], +) +def test_sanitize_url_strips_userinfo_from_a_non_credential_fragment_value(value: str) -> None: + sanitized = sanitize_url(f"https://h.test/x#page=3&next={value}") + + assert "VALSECRET" not in sanitized + assert "page=3" in sanitized + + +def test_sanitize_url_keeps_query_values_that_carry_no_userinfo() -> None: + url = "https://h.test/x?next=https%3A%2F%2Fdocs.test%2Fa&page=3&q=user%40example.com" + + assert sanitize_url(url) == url + + +def test_request_url_anchor_userinfo_is_redacted_in_the_failure_message() -> None: + request = httpx.Request("GET", "https://api.tangle.test/v1/jobs#alice:REQANCHOR@evil.test") + exc = httpx.ConnectError("connection refused", request=request) + + message = format_request_error(exc) + + assert "REQANCHOR" not in message + assert "\n" not in message + + +def test_request_url_encoded_query_userinfo_is_redacted_in_the_status_line() -> None: + request = httpx.Request( + "POST", "https://api.tangle.test/x?next=https%3A%2F%2Falice%3AREQQUERY%40evil.test" + ) + response = httpx.Response(401, request=request, text="denied") + + message = format_http_status_error( + httpx.HTTPStatusError("error", request=request, response=response) + ) + + assert "REQQUERY" not in message + + +@pytest.mark.parametrize( + "url", + [ + "https://h.test/cb#alice:BODYANCHOR@evil.test", + "https://h.test/cb#next=https%3A%2F%2Falice%3ABODYANCHOR%40evil.test", + "https://h.test/cb?next=https%3A%2F%2Falice%3ABODYANCHOR%40evil.test", + "https://h.test/cb?next=alice%3ABODYANCHOR%40evil.test", + ], +) +def test_body_url_userinfo_is_redacted_wherever_it_hides(url: str) -> None: + body = json.dumps({"detail": f"retry at {url}"}) + + assert "BODYANCHOR" not in format_http_status_error(_status_error(body)) + + +def _nested(target: str, *, key: str = "next", separator: str = "?") -> str: + return f"https://h.test/x{separator}{key}={urllib.parse.quote(target, safe='')}" + + +@pytest.mark.parametrize( + "parameter", + [ + "access_token=NESTSECRET", + "oauth=NESTSECRET", + "id_token=NESTSECRET", + "credential=NESTSECRET", + "security-token=NESTSECRET", + "X-Amz-Signature=NESTSECRET", + "x-amz-security-token=NESTSECRET", + ], +) +def test_sanitize_url_redacts_credentials_of_a_nested_url_value(parameter: str) -> None: + sanitized = sanitize_url(_nested(f"https://evil.test/cb?{parameter}")) + + assert "NESTSECRET" not in sanitized + assert "evil.test" in sanitized + + +@pytest.mark.parametrize("separator", ["?", "#"]) +def test_sanitize_url_redacts_a_nested_url_credential_in_query_and_fragment( + separator: str, +) -> None: + sanitized = sanitize_url( + _nested("https://evil.test/cb?access_token=NESTSECRET", separator=separator) + ) + + assert "NESTSECRET" not in sanitized + + +def test_sanitize_url_redacts_a_credential_in_the_nested_urls_own_fragment() -> None: + sanitized = sanitize_url(_nested("https://evil.test/cb#access_token=NESTSECRET")) + + assert "NESTSECRET" not in sanitized + + +def test_sanitize_url_redacts_a_nested_url_that_was_not_percent_encoded() -> None: + sanitized = sanitize_url("https://h.test/x?next=https://evil.test/cb?access_token=NESTSECRET") + + assert "NESTSECRET" not in sanitized + + +def test_sanitize_url_redacts_both_userinfo_and_parameters_of_a_nested_url() -> None: + sanitized = sanitize_url(_nested("https://bob:NESTUSER@evil.test/cb?access_token=NESTPARAM")) + + assert "NESTUSER" not in sanitized + assert "NESTPARAM" not in sanitized + + +def test_sanitize_url_keeps_a_nested_url_that_carries_no_credential() -> None: + url = _nested("https://docs.test/guide?page=3&lang=en") + + assert sanitize_url(url) == url + + +def test_sanitize_url_stops_descending_after_one_nested_level() -> None: + """The bound is what keeps an untrusted body from driving unbounded recursion.""" + + nested = "https://u:DEEPSECRET@evil.test" + for _ in range(30): + nested = f"https://h.test/y?next={urllib.parse.quote(nested, safe='')}" + + sanitized = sanitize_url(nested) + + assert sanitized.startswith("https://h.test/y?next=") + assert "DEEPSECRET" not in sanitized + + +@pytest.mark.parametrize( + "url", + [ + "https://h.test/x?alice:KEYSECRET@evil.test=1", + "https://h.test/x?alice%3AKEYSECRET%40evil.test=1", + "https://h.test/x#alice%3AKEYSECRET%40evil.test=1", + "https://h.test/x#alice:KEYSECRET@evil.test=1", + ], +) +def test_sanitize_url_strips_userinfo_from_a_parameter_name(url: str) -> None: + sanitized = sanitize_url(url) + + assert "KEYSECRET" not in sanitized + + +def test_nested_url_credential_is_redacted_in_the_status_line() -> None: + url = _nested("https://evil.test/cb?access_token=REQNESTED") + request = httpx.Request("POST", url) + response = httpx.Response(401, request=request, text="denied") + + message = format_http_status_error( + httpx.HTTPStatusError("error", request=request, response=response) + ) + + assert "REQNESTED" not in message + + +def test_nested_url_credential_is_redacted_in_the_connection_failure() -> None: + exc = httpx.ConnectError( + "refused", + request=httpx.Request("GET", _nested("https://evil.test/cb?access_token=REQNESTED")), + ) + + message = format_request_error(exc) + + assert "REQNESTED" not in message + assert "\n" not in message + + +@pytest.mark.parametrize( + "target", + [ + "https://evil.test/cb?access_token=BODYNESTED", + "https://evil.test/cb#access_token=BODYNESTED", + "https://bob:BODYNESTED@evil.test/cb?page=1", + ], +) +def test_nested_url_credential_is_redacted_in_a_reflected_body(target: str) -> None: + body = json.dumps({"detail": f"redirecting to {_nested(target)}"}) + + assert "BODYNESTED" not in format_http_status_error(_status_error(body)) + + +@pytest.mark.parametrize("name", ["oauth", "OAuth", "oauth_token"]) +def test_bare_oauth_parameter_is_redacted(name: str) -> None: + assert "OAUTHSECRET" not in sanitize_url(f"https://h.test/x?{name}=OAUTHSECRET") + + +@pytest.mark.parametrize("name", ["oauthlib", "oauth_callback", "oauth_version"]) +def test_oauth_lookalike_names_keep_their_values(name: str) -> None: + assert "keepme" in sanitize_url(f"https://h.test/x?{name}=keepme") + + +@pytest.mark.parametrize( + "carried", + [ + "access_token=CARRIEDSECRET", + "/cb?access_token=CARRIEDSECRET", + "mailto:ops@evil.test?password=CARRIEDSECRET", + "data:text/plain,access_token=CARRIEDSECRET", + "custom:api_key=CARRIEDSECRET", + "Bearer CARRIEDSECRETCARRIEDSECRET", + ], +) +@pytest.mark.parametrize("separator", ["?", "#"]) +def test_sanitize_url_redacts_an_assignment_carried_in_a_value( + carried: str, separator: str +) -> None: + """A credential need not be a parameter of its own to be displayed.""" + + url = f"https://h.test/x{separator}state={urllib.parse.quote(carried, safe='')}" + + assert "CARRIEDSECRET" not in sanitize_url(url) + + +def test_sanitize_url_redacts_an_unencoded_assignment_carried_in_a_value() -> None: + assert "CARRIEDSECRET" not in sanitize_url("https://h.test/x?state=access_token:CARRIEDSECRET") + + +def _encoded(times: int, plain: str = "access_token=CARRIEDSECRET") -> str: + for _ in range(times): + plain = urllib.parse.quote(plain, safe="") + return plain + + +# Every shape a credential arrives in, to be run against every placement and every +# encoding depth. Scanning a decoded layer for one shape and not the others lets the +# missing shape ride out of the sanitizer, so the cross-product is the point. +_CARRIED_PAYLOADS = [ + "access_token=CARRIEDSECRET", + "alice:pwCARRIEDSECRET@host.test", + "https://e.test/cb?access_token=CARRIEDSECRET", + "https://alice:pwCARRIEDSECRET@e.test/cb", + "/cb?api_key=CARRIEDSECRET", +] + + +@pytest.mark.parametrize("payload", _CARRIED_PAYLOADS) +@pytest.mark.parametrize("depth", [0, 1, 2, 3, 4, 5, 6]) +@pytest.mark.parametrize("separator", ["?", "#"]) +def test_sanitize_url_redacts_a_carried_credential_at_any_encoding_depth( + depth: int, separator: str, payload: str +) -> None: + """Encoding again must bury the credential, never carry it past the scan.""" + + url = f"https://h.test/x{separator}next={_encoded(depth, payload)}" + + assert "CARRIEDSECRET" not in sanitize_url(url) + + +@pytest.mark.parametrize("payload", _CARRIED_PAYLOADS) +@pytest.mark.parametrize("depth", [1, 2, 3]) +@pytest.mark.parametrize("separator", ["?", "#"]) +def test_sanitize_url_redacts_a_credential_encoded_into_a_parameter_name( + depth: int, separator: str, payload: str +) -> None: + """A name is displayed just as a value is, so it gets the same leaf scrub.""" + + url = f"https://h.test/x{separator}{_encoded(depth, payload)}=1" + + assert "CARRIEDSECRET" not in sanitize_url(url) + + +@pytest.mark.parametrize("payload", _CARRIED_PAYLOADS) +@pytest.mark.parametrize("depth", [0, 1, 2, 3, 4, 5, 6]) +def test_plain_anchor_redacts_a_carried_credential_at_any_encoding_depth( + depth: int, payload: str +) -> None: + """An anchor is not parsed as pairs, so it needs the same per-layer scan.""" + + encoded = _encoded(depth, payload) + if "=" in encoded or "&" in encoded: + pytest.skip("carries a separator, so it is parsed as pairs rather than an anchor") + + assert "CARRIEDSECRET" not in sanitize_url(f"https://h.test/x#{encoded}") + + +@pytest.mark.parametrize("payload", _CARRIED_PAYLOADS) +@pytest.mark.parametrize("depth", [1, 2, 3, 4, 5, 6]) +def test_carried_credential_at_depth_is_redacted_on_every_display_path( + depth: int, payload: str +) -> None: + """The bound holds on the exception paths and on a reflected body alike.""" + + url = f"https://h.test/x?next={_encoded(depth, payload)}" + request = httpx.Request("GET", url) + + rendered = [ + format_request_error(httpx.ConnectError("refused", request=request)), + format_request_error(httpx.ConnectTimeout("slow", request=request)), + format_http_status_error(_status_error(json.dumps({"detail": f"see {url}"}))), + format_http_status_error(_status_error(f"redirect to {url} failed")), + format_http_status_error(_status_error(f"

go {url}

")), + ] + + assert not [text for text in rendered if "CARRIEDSECRET" in text] + + +@pytest.mark.parametrize("payload", _CARRIED_PAYLOADS) +@pytest.mark.parametrize("depth", [2, 3, 4, 5, 6]) +def test_deeply_encoded_credential_is_dropped_whole(depth: int, payload: str) -> None: + """Past the layers actually decoded the value is unexamined, so it is dropped.""" + + sanitized = sanitize_url(f"https://h.test/x?next={_encoded(depth, payload)}") + + assert sanitized == "https://h.test/x?next=" + + +@pytest.mark.parametrize( + "chain", + ["%25" * 20000, "%2525" * 12000, ("%25" * 3 + "a") * 8000, "%2540" * 12000], +) +def test_hostile_percent_chains_stay_bounded(chain: str) -> None: + """Decoding is capped, so a hostile value cannot choose how many layers we peel.""" + + start = time.perf_counter() + + sanitize_url(f"https://h.test/x?next={chain}") + + assert time.perf_counter() - start < 2.0 + + +def test_scrubbing_a_parameter_name_does_not_change_the_verdict_on_its_value() -> None: + """Sensitivity is judged on the name as parsed, before it is rewritten.""" + + assert sanitize_url("https://h.test/x?access_token=SECRET") == ( + "https://h.test/x?access_token=" + ) + assert sanitize_url("https://h.test/x?session_id=sess_1") == ( + "https://h.test/x?session_id=sess_1" + ) + + +@pytest.mark.parametrize( + "url", + [ + "https://h.test/x?page=2&limit=10", + "https://h.test/x?dotted.key=1&dash-key=2", + "https://h.test/x?state=abc%3Ddef", + "https://h.test/x?state=abc%253Ddef", + "https://h.test/x?pct=100%25", + ], +) +def test_benign_names_and_encodings_survive_the_bounded_scrub(url: str) -> None: + assert sanitize_url(url) == url + + +def test_sanitize_url_redacts_an_assignment_in_a_plain_anchor() -> None: + assert "CARRIEDSECRET" not in sanitize_url("https://h.test/guide#api_key:CARRIEDSECRET") + + +@pytest.mark.parametrize( + "value", + ["/cb?page=3", "returning", "state:ok", "abc%3D", "next=/dashboard"], +) +def test_sanitize_url_keeps_a_value_carrying_no_credential(value: str) -> None: + url = f"https://h.test/x?state={urllib.parse.quote(value, safe='')}" + + assert sanitize_url(url) == url + + +def test_carried_assignment_is_redacted_in_the_connection_failure() -> None: + url = f"https://h.test/x?state={urllib.parse.quote('access_token=CARRIEDSECRET', safe='')}" + exc = httpx.ConnectError("refused", request=httpx.Request("GET", url)) + + message = format_request_error(exc) + + assert "CARRIEDSECRET" not in message + assert "\n" not in message + + +def test_carried_assignment_is_redacted_in_a_reflected_body() -> None: + url = f"https://h.test/x?state={urllib.parse.quote('access_token=CARRIEDSECRET', safe='')}" + body = json.dumps({"detail": f"redirecting to {url}"}) + + assert "CARRIEDSECRET" not in format_http_status_error(_status_error(body))