diff --git a/.github/workflows/pinned-https-coverage.yml b/.github/workflows/pinned-https-coverage.yml new file mode 100644 index 000000000..6149a566b --- /dev/null +++ b/.github/workflows/pinned-https-coverage.yml @@ -0,0 +1,52 @@ +name: Pinned HTTPS Coverage + +on: + pull_request: + branches: [develop, main] + push: + branches: [develop, main] + +permissions: + contents: read + +jobs: + exact-pinned-https-coverage: + runs-on: ubuntu-latest + timeout-minutes: 15 + env: + FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: true + steps: + - name: Checkout repository + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Set up Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: '3.13' + + - name: Install hash-locked test dependencies + run: python -m pip install --disable-pip-version-check --no-cache-dir --require-hashes -r requirements-test.txt + + - name: Run DNS-pinned HTTPS contracts + run: | + python -m pytest -q \ + tests/test_pinned_https_resolution.py \ + tests/test_pinned_https_redirects.py \ + tests/test_pinned_https_cli_integration.py \ + tests/test_pinned_https_validation_edges.py \ + tests/test_pinned_https_coverage_edges.py \ + tests/test_pinned_https_release_contract.py \ + tests/test_ssrf_protection.py + + - name: Verify exact unrounded statement coverage + run: | + python -m scripts.ci.verify_module_coverage \ + --module appguardrail_core/pinned_https.py \ + --test tests/test_pinned_https_resolution.py \ + --test tests/test_pinned_https_redirects.py \ + --test tests/test_pinned_https_cli_integration.py \ + --test tests/test_pinned_https_validation_edges.py \ + --test tests/test_pinned_https_coverage_edges.py \ + --test tests/test_pinned_https_release_contract.py diff --git a/CHANGELOG.d/892-pinned-control-plane-delivery.md b/CHANGELOG.d/892-pinned-control-plane-delivery.md new file mode 100644 index 000000000..514897093 --- /dev/null +++ b/CHANGELOG.d/892-pinned-control-plane-delivery.md @@ -0,0 +1,6 @@ +### Security + +- Added DNS-pinned public HTTPS delivery for bearer-authenticated control-plane scan uploads, preserving the original hostname for TLS SNI and certificate verification while connecting only to the exact globally routable addresses admitted by validation. +- Restricted credential-bearing redirects to HTTP 307 and 308, revalidated every hop, and removed `Authorization` and `Proxy-Authorization` case-insensitively whenever the normalized HTTPS origin changes. +- Added bounded timeout, redirect-count, response-size, JSON, and non-secret error contracts plus exact statement-coverage and operator documentation for standalone, organization-service, and naruon reuse. +- Reject ambiguous control-plane base URLs and validate receipt identifiers and deploy-blocking counts before rendering terminal output, preventing query/fragment confusion and forged control-sequence messages. diff --git a/appguardrail_core/__init__.py b/appguardrail_core/__init__.py index a8aab08f3..2c0047e18 100644 --- a/appguardrail_core/__init__.py +++ b/appguardrail_core/__init__.py @@ -67,6 +67,16 @@ parse_project_matches as parse_openssf_project_matches, ) from appguardrail_core.openssf_report import augment_buyer_diligence_report +from appguardrail_core.pinned_https import ( + DestinationValidationError, + HTTPSDestination, + PinnedHTTPSConnection, + PinnedHTTPSFailure, + PinnedHTTPSResponse, + ResolvedAddress, + post_json_pinned_https, + resolve_public_https_destination, +) from appguardrail_core.org_intelligence import ( BuyerEvidenceMetric, BuyerEvidencePack, @@ -146,10 +156,12 @@ def render_buyer_diligence_report( "CURRENT_SCHEMA_VERSION", "DEFAULT_RETENTION_DAYS", "DEPLOY_BLOCKING_SEVERITIES", + "DestinationValidationError", "DriftAssessment", "ExternalEngineDecision", "ExternalScanPlan", "GENESIS_EVENT_HASH", + "HTTPSDestination", "MAX_RETENTION_DAYS", "MIN_RETENTION_DAYS", "MetricResult", @@ -158,9 +170,13 @@ def render_buyer_diligence_report( "OrgInventory", "PurgePreview", "PurgeReceipt", + "PinnedHTTPSConnection", + "PinnedHTTPSFailure", + "PinnedHTTPSResponse", "PullRequestGateSummary", "RETENTION_CATEGORIES", "ReportContext", + "ResolvedAddress", "RepositoryGateSummary", "RetentionPolicy", "RetentionPolicyConflict", @@ -198,9 +214,11 @@ def render_buyer_diligence_report( "normalize_finding", "normalize_findings", "parse_openssf_project_matches", + "post_json_pinned_https", "recompute_event_hash", "render_buyer_diligence_report", "render_org_readiness_report", + "resolve_public_https_destination", "safe_report_snippet", "sanitize_audit_summary", "score_sale_readiness", diff --git a/appguardrail_core/pinned_https.py b/appguardrail_core/pinned_https.py new file mode 100644 index 000000000..ef1563f9c --- /dev/null +++ b/appguardrail_core/pinned_https.py @@ -0,0 +1,512 @@ +"""Deliver credential-bearing JSON over DNS-pinned public HTTPS connections. + +The transport closes the gap between validating a hostname and later asking the +operating system to resolve that hostname again for the actual TCP connection. +Each request hop resolves exactly once, rejects every non-global answer, and +connects only to the captured socket addresses while retaining the original +hostname for TLS Server Name Indication and certificate identity verification. +""" + +from __future__ import annotations + +import http.client +import ipaddress +import json +import re +import socket +import ssl +import urllib.parse +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from typing import Any + + +DEFAULT_TIMEOUT_SECONDS = 15.0 +DEFAULT_MAX_REDIRECTS = 5 +DEFAULT_MAX_RESPONSE_BYTES = 1_000_000 +_REDIRECT_STATUSES = frozenset({307, 308}) +_SENSITIVE_HEADERS = frozenset({"authorization", "proxy-authorization"}) +_FORBIDDEN_CALLER_HEADERS = frozenset( + {"connection", "content-length", "host", "transfer-encoding"} +) +_HEADER_NAME_RE = re.compile(r"\A[!#$%&'*+.^_`|~0-9A-Za-z-]+\Z") + + +class PinnedHTTPSFailure(RuntimeError): + """Report one bounded DNS-pinned HTTPS transport failure.""" + + +class DestinationValidationError(PinnedHTTPSFailure): + """Report an invalid or non-public credential-bearing destination.""" + + +@dataclass(frozen=True) +class ResolvedAddress: + """One validated TCP address captured from a single resolver decision.""" + + family: int + socket_type: int + protocol: int + socket_address: tuple[Any, ...] + ip_address: str + + +@dataclass(frozen=True) +class HTTPSDestination: + """A canonical HTTPS origin and its immutable validated address set.""" + + url: str + hostname: str + port: int + request_target: str + addresses: tuple[ResolvedAddress, ...] + + @property + def origin(self) -> tuple[str, str, int]: + """Return the normalized origin used for redirect credential scoping.""" + return ("https", self.hostname, self.port) + + +@dataclass(frozen=True) +class PinnedHTTPSResponse: + """A bounded HTTP response returned by the pinned transport.""" + + status: int + reason: str + headers: tuple[tuple[str, str], ...] + body: bytes + + def get_header(self, name: str) -> str | None: + """Return the first matching response header without case sensitivity.""" + expected = name.lower() + for header_name, value in self.headers: + if header_name.lower() == expected: + return value + return None + + +Resolver = Callable[ + [str, int, int, int, int], + Sequence[tuple[int, int, int, str, tuple[Any, ...]]], +] +SocketFactory = Callable[[int, int, int], Any] +ConnectionFactory = Callable[[HTTPSDestination, float], Any] + + +def _has_control_character(value: str) -> bool: + """Return whether a URL or header contains an ASCII control character.""" + return any(ord(character) < 0x20 or ord(character) == 0x7F for character in value) + + +def _is_permitted_global_address(value: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: + """Return whether an address is globally routable and not IPv4-mapped.""" + if isinstance(value, ipaddress.IPv6Address) and value.ipv4_mapped is not None: + return False + return not ( + value.is_loopback + or value.is_private + or value.is_link_local + or value.is_multicast + or value.is_unspecified + or value.is_reserved + or not value.is_global + ) + + +def _canonical_hostname(hostname: str) -> str: + """Return a lowercase ASCII service identity suitable for TLS verification.""" + candidate = hostname.strip().rstrip(".") + if not candidate or "%" in candidate or _has_control_character(candidate): + raise DestinationValidationError("HTTPS destination has an invalid hostname") + try: + literal = ipaddress.ip_address(candidate) + except ValueError: + try: + candidate = candidate.encode("idna").decode("ascii").lower() + except UnicodeError as exc: + raise DestinationValidationError( + "HTTPS destination hostname is not valid IDNA" + ) from exc + if not candidate or len(candidate) > 253: + raise DestinationValidationError("HTTPS destination hostname is invalid") + return candidate + if not _is_permitted_global_address(literal): + raise DestinationValidationError( + "HTTPS destination resolved to a non-global address" + ) + return literal.compressed.lower() + + +def _quoted_request_target(parts: urllib.parse.SplitResult) -> str: + """Return an origin-form request target while preserving valid percent escapes.""" + path = urllib.parse.quote( + parts.path or "/", + safe="/%:@!$&'()*+,;=-._~", + ) + if not path.startswith("/"): + path = "/" + path + if not parts.query: + return path + query = urllib.parse.quote( + parts.query, + safe="/%?:@!$&'()*+,;=-._~", + ) + return f"{path}?{query}" + + +def _canonical_url(hostname: str, port: int, request_target: str) -> str: + """Build the canonical absolute URL retained for relative redirect resolution.""" + authority_host = f"[{hostname}]" if ":" in hostname else hostname + authority = authority_host if port == 443 else f"{authority_host}:{port}" + return f"https://{authority}{request_target}" + + +def _resolved_address( + result: tuple[int, int, int, str, tuple[Any, ...]], + *, + expected_port: int, +) -> ResolvedAddress | None: + """Validate one getaddrinfo result and return its immutable representation.""" + try: + family, socket_type, protocol, _canonical_name, socket_address = result + except (TypeError, ValueError): + return None + if family not in {socket.AF_INET, socket.AF_INET6}: + return None + if socket_type != socket.SOCK_STREAM or protocol not in {0, socket.IPPROTO_TCP}: + return None + if not isinstance(socket_address, tuple) or len(socket_address) < 2: + return None + if socket_address[1] != expected_port: + raise DestinationValidationError("resolver returned an unexpected port") + raw_ip = str(socket_address[0]) + if "%" in raw_ip: + raise DestinationValidationError("zone-scoped addresses are not permitted") + try: + parsed_ip = ipaddress.ip_address(raw_ip) + except ValueError as exc: + raise DestinationValidationError("resolver returned an invalid IP address") from exc + if not _is_permitted_global_address(parsed_ip): + raise DestinationValidationError( + "HTTPS destination resolved to a non-global address" + ) + if family == socket.AF_INET6 and len(socket_address) >= 4 and socket_address[3] != 0: + raise DestinationValidationError("zone-scoped addresses are not permitted") + return ResolvedAddress( + family=family, + socket_type=socket.SOCK_STREAM, + protocol=socket.IPPROTO_TCP, + socket_address=socket_address, + ip_address=parsed_ip.compressed, + ) + + +def resolve_public_https_destination( + url: str, + *, + resolver: Resolver = socket.getaddrinfo, +) -> HTTPSDestination: + """Resolve one strict public HTTPS destination exactly once. + + Every returned TCP address must be globally routable. A mixed answer set is + rejected in full instead of silently selecting only the public subset. + """ + if not isinstance(url, str) or not url or _has_control_character(url): + raise DestinationValidationError("HTTPS destination URL is invalid") + try: + parts = urllib.parse.urlsplit(url) + except ValueError as exc: + raise DestinationValidationError("HTTPS destination URL is invalid") from exc + if parts.scheme.lower() != "https": + raise DestinationValidationError("control-plane delivery requires public HTTPS") + if parts.username is not None or parts.password is not None: + raise DestinationValidationError("HTTPS destination must not contain credentials") + if parts.fragment: + raise DestinationValidationError("HTTPS destination must not contain a fragment") + if not parts.hostname: + raise DestinationValidationError("HTTPS destination must include a hostname") + hostname = _canonical_hostname(parts.hostname) + try: + parsed_port = parts.port + port = 443 if parsed_port is None else parsed_port + except ValueError as exc: + raise DestinationValidationError("HTTPS destination port is invalid") from exc + if isinstance(port, bool) or not 1 <= port <= 65535: + raise DestinationValidationError("HTTPS destination port is invalid") + request_target = _quoted_request_target(parts) + + try: + answers = resolver( + hostname, + port, + socket.AF_UNSPEC, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + ) + except (OSError, socket.gaierror) as exc: + raise DestinationValidationError("HTTPS destination resolution failed") from exc + + validated: list[ResolvedAddress] = [] + seen: set[tuple[int, int, int, tuple[Any, ...]]] = set() + for result in answers: + address = _resolved_address(result, expected_port=port) + if address is None: + continue + identity = ( + address.family, + address.socket_type, + address.protocol, + address.socket_address, + ) + if identity not in seen: + seen.add(identity) + validated.append(address) + if not validated: + raise DestinationValidationError( + "HTTPS destination resolution returned no TCP addresses" + ) + canonical_url = _canonical_url(hostname, port, request_target) + return HTTPSDestination( + url=canonical_url, + hostname=hostname, + port=port, + request_target=request_target, + addresses=tuple(validated), + ) + + +class PinnedHTTPSConnection(http.client.HTTPSConnection): + """Connect an HTTPS request only to a prevalidated immutable address set.""" + + def __init__( + self, + destination: HTTPSDestination, + *, + timeout: float = DEFAULT_TIMEOUT_SECONDS, + context: ssl.SSLContext | Any | None = None, + socket_factory: SocketFactory = socket.socket, + ) -> None: + """Create a connection retaining the URL hostname for TLS verification.""" + super().__init__( + destination.hostname, + destination.port, + timeout=timeout, + context=context, + ) + self.destination = destination + self._socket_factory = socket_factory + + def connect(self) -> None: + """Open TCP directly to the validated addresses and then negotiate TLS.""" + if self._tunnel_host: + raise OSError("HTTP tunneling is not supported by pinned HTTPS") + last_error: OSError | None = None + for address in self.destination.addresses: + raw_socket = None + try: + raw_socket = self._socket_factory( + address.family, + address.socket_type, + address.protocol, + ) + raw_socket.settimeout(self.timeout) + if self.source_address: + raw_socket.bind(self.source_address) + raw_socket.connect(address.socket_address) + self.sock = self._context.wrap_socket( + raw_socket, + server_hostname=self.destination.hostname, + ) + return + except OSError as exc: + last_error = exc + if raw_socket is not None: + raw_socket.close() + if last_error is not None: + raise last_error + raise OSError("HTTPS destination has no validated addresses") + + +def _validated_bound( + value: int | float, + *, + name: str, + allow_zero: bool, + integer_only: bool, +) -> int | float: + """Validate one resource bound before serialization, resolution, or I/O.""" + if isinstance(value, bool): + raise ValueError(f"{name} must be a valid resource bound") + if integer_only and not isinstance(value, int): + raise ValueError(f"{name} must be an integer") + if not integer_only and not isinstance(value, (int, float)): + raise ValueError(f"{name} must be numeric") + if value < 0 or (not allow_zero and value == 0): + raise ValueError(f"{name} must be positive") + return value + + +def _prepared_headers(headers: Mapping[str, str] | None) -> dict[str, str]: + """Validate caller headers and install deterministic JSON defaults.""" + prepared: dict[str, str] = {} + observed_names: set[str] = set() + for name, value in (headers or {}).items(): + if not isinstance(name, str) or not _HEADER_NAME_RE.fullmatch(name): + raise ValueError("HTTP header name is invalid") + if not isinstance(value, str) or _has_control_character(value): + raise ValueError("HTTP header value is invalid") + lowered = name.lower() + if lowered in _FORBIDDEN_CALLER_HEADERS: + raise ValueError(f"caller cannot set the {name} header") + if lowered in observed_names: + raise ValueError("duplicate HTTP header names are not permitted") + observed_names.add(lowered) + prepared[name] = value + if "content-type" not in observed_names: + prepared["Content-Type"] = "application/json" + if "accept" not in observed_names: + prepared["Accept"] = "application/json" + return prepared + + +def _without_sensitive_headers(headers: Mapping[str, str]) -> dict[str, str]: + """Remove credential headers using their actual case-preserved names.""" + return { + name: value + for name, value in headers.items() + if name.lower() not in _SENSITIVE_HEADERS + } + + +def _default_connection_factory( + destination: HTTPSDestination, + timeout: float, +) -> PinnedHTTPSConnection: + """Create the production pinned HTTPS connection for one validated hop.""" + return PinnedHTTPSConnection(destination, timeout=timeout) + + +def post_json_pinned_https( + url: str, + payload: Any, + *, + headers: Mapping[str, str] | None = None, + timeout: float = DEFAULT_TIMEOUT_SECONDS, + max_redirects: int = DEFAULT_MAX_REDIRECTS, + max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES, + resolver: Resolver = socket.getaddrinfo, + connection_factory: ConnectionFactory | None = None, +) -> PinnedHTTPSResponse: + """POST JSON through strict DNS-pinned HTTPS with bounded redirects. + + Only method-preserving 307 and 308 redirects are followed. Sensitive + authorization headers remain on the same normalized HTTPS origin and are + stripped case-insensitively before any cross-origin request. + """ + timeout = float( + _validated_bound( + timeout, + name="timeout", + allow_zero=False, + integer_only=False, + ) + ) + max_redirects = int( + _validated_bound( + max_redirects, + name="max_redirects", + allow_zero=True, + integer_only=True, + ) + ) + max_response_bytes = int( + _validated_bound( + max_response_bytes, + name="max_response_bytes", + allow_zero=False, + integer_only=True, + ) + ) + request_headers = _prepared_headers(headers) + try: + body = json.dumps( + payload, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + except (TypeError, ValueError, RecursionError) as exc: + raise PinnedHTTPSFailure("JSON serialization failed") from exc + + destination = resolve_public_https_destination(url, resolver=resolver) + create_connection = connection_factory or _default_connection_factory + redirects_followed = 0 + + while True: + connection = create_connection(destination, timeout) + response = None + try: + connection.request( + "POST", + destination.request_target, + body=body, + headers=request_headers, + ) + response = connection.getresponse() + response_body = response.read(max_response_bytes + 1) + if len(response_body) > max_response_bytes: + raise PinnedHTTPSFailure( + f"HTTPS response exceeds {max_response_bytes} bytes" + ) + result = PinnedHTTPSResponse( + status=int(response.status), + reason=str(response.reason or ""), + headers=tuple( + (str(name), str(value)) for name, value in response.getheaders() + ), + body=response_body, + ) + except PinnedHTTPSFailure: + raise + except (OSError, http.client.HTTPException) as exc: + raise PinnedHTTPSFailure("pinned HTTPS delivery failed") from exc + finally: + connection.close() + + if not 300 <= result.status < 400: + return result + if result.status not in _REDIRECT_STATUSES: + raise PinnedHTTPSFailure( + f"unsupported redirect status {result.status}" + ) + if redirects_followed >= max_redirects: + raise PinnedHTTPSFailure("too many redirects") + location = result.get_header("Location") + if location is None or not location.strip(): + raise PinnedHTTPSFailure("redirect response is missing Location") + if _has_control_character(location): + raise PinnedHTTPSFailure("redirect Location is invalid") + redirected_url = urllib.parse.urljoin(destination.url, location) + redirected_destination = resolve_public_https_destination( + redirected_url, + resolver=resolver, + ) + if redirected_destination.origin != destination.origin: + request_headers = _without_sensitive_headers(request_headers) + destination = redirected_destination + redirects_followed += 1 + + +__all__ = [ + "DEFAULT_MAX_REDIRECTS", + "DEFAULT_MAX_RESPONSE_BYTES", + "DEFAULT_TIMEOUT_SECONDS", + "DestinationValidationError", + "HTTPSDestination", + "PinnedHTTPSConnection", + "PinnedHTTPSFailure", + "PinnedHTTPSResponse", + "ResolvedAddress", + "post_json_pinned_https", + "resolve_public_https_destination", +] diff --git a/docs/pinned-https-control-plane-delivery.md b/docs/pinned-https-control-plane-delivery.md new file mode 100644 index 000000000..78131a1d2 --- /dev/null +++ b/docs/pinned-https-control-plane-delivery.md @@ -0,0 +1,118 @@ +# DNS-pinned control-plane delivery + +AppGuardrail sends normalized scan findings to an optional control plane with a bearer credential. That path has a stricter trust boundary than unauthenticated webhook delivery: every initial destination and redirect must use public HTTPS, the exact DNS answers admitted by validation must also be the addresses used by the TCP connection, and the original URL hostname must remain the service identity used for TLS SNI and certificate verification. + +This module is independently importable from `appguardrail_core` so standalone installations, ContextualWisdomLab organization services, `naruon`, and other modular MSA consumers can reuse the same transport without importing the scanner CLI. + +## Threat model + +A preflight-only URL check is insufficient for a credential-bearing request. If the application validates a hostname and the HTTP client resolves it again later, an attacker controlling DNS can return a public address during validation and a private, loopback, link-local, metadata-service, or otherwise disallowed address during connection. This time-of-check/time-of-use difference is a DNS rebinding form of server-side request forgery. + +The transport also treats redirects as a new authorization boundary. A redirect may change scheme, host, or port. Credential headers must not follow that change merely because the HTTP library constructs another request. + +## Architecture decision + +```mermaid +flowchart LR + A[CLI normalized findings] --> B[Strict HTTPS URL parser] + B --> C[Single DNS decision] + C --> D{Every answer globally routable?} + D -->|No| E[Fail closed before credential I/O] + D -->|Yes| F[Immutable validated address set] + F --> G[Direct TCP connect to captured address] + G --> H[TLS with original hostname for SNI and certificate identity] + H --> I[POST JSON] + I --> J{HTTP response} + J -->|2xx or non-redirect| K[Bounded response] + J -->|307 or 308| L[Resolve and validate next hop once] + L --> M{Same HTTPS origin?} + M -->|Yes| G + M -->|No| N[Remove Authorization and Proxy-Authorization] + N --> G + J -->|Other 3xx| E +``` + +The implementation uses the following contract: + +1. Parse an absolute HTTPS URL and reject credentials, fragments, invalid ports—including an explicit port `0`—ASCII control characters, and zone-scoped IPv6 identities before DNS is consulted. +2. Resolve the normalized hostname once with TCP-only `getaddrinfo` parameters. +3. Reject the complete answer set when any returned address is private, loopback, link-local, multicast, unspecified, reserved, non-global, IPv4-mapped IPv6, or zone-scoped. +4. Connect only to the immutable socket addresses captured by that resolver call. +5. Pass the original normalized hostname to `SSLContext.wrap_socket(server_hostname=...)`, preserving TLS SNI and hostname-based certificate verification. +6. Follow only method-preserving HTTP 307 and 308 redirects. +7. Resolve each redirect target exactly once under the same public-address policy. +8. Preserve credential headers only on the same normalized `https` origin. Remove `Authorization` and `Proxy-Authorization` case-insensitively before a cross-origin request. +9. Bound timeout, redirect count, JSON serialization, and response bytes before data is exposed to callers. +10. Return bounded error categories without copying response bodies, credentials, raw socket errors, or tenant data into scanner output. + +## Separation from webhook delivery + +`appguardrail_core.controlplane._send_alert` remains a generic unauthenticated webhook path. It can preserve its existing public HTTP/HTTPS compatibility and destination policy because it does not carry the AppGuardrail control-plane bearer token. `appguardrail scan --push` uses the dedicated DNS-pinned HTTPS transport. The two policies must not be silently combined. + +## Redirect semantics + +HTTP 307 and 308 preserve the method and request content. Other 3xx responses are rejected because changing a credential-bearing POST into a GET or applying ambiguous user-agent rewrite behavior would weaken the delivery contract. Relative `Location` values are resolved according to URI reference resolution, then reparsed and revalidated as a new HTTPS destination. + +A normalized origin is the tuple `(scheme, hostname, port)`. Explicit port `443` and an omitted HTTPS port are equivalent. Any host or port change removes both sensitive headers by comparing actual header names case-insensitively; this avoids relying on `urllib`'s title-casing behavior. + +## TLS service identity + +The TCP connection uses the validated IP address, but the TLS handshake continues to use the original hostname. This preserves: + +- Server Name Indication for virtual hosting; +- certificate chain validation through the default Python SSL context; and +- reference-identifier matching against the service hostname rather than the numeric peer address. + +The transport does not permit HTTP proxy tunneling. Deployments requiring an outbound proxy should terminate this path behind a separately reviewed egress service that enforces equivalent private-address denial and preserves end-to-end TLS service identity. + +## Operational egress control + +Process-level address pinning is defense in depth, not a replacement for network controls. Enterprise deployments should also deny access from scanner workloads to loopback, link-local, cloud metadata, RFC 1918, unique-local IPv6, internal service ranges, and administrative networks unless an explicitly reviewed destination is required. DNS, proxy, firewall, service-mesh, and container-network policy should agree on the same boundary. + +## Failure and recovery semantics + +A failed delivery does not turn a completed local scan into a failed scan. The CLI reports a bounded delivery status and retains the local findings output. Retrying is safe only when the control-plane endpoint provides its own idempotency contract; the transport itself does not assume remote deduplication. + +Rollback consists of reverting the integration commit. The existing unauthenticated webhook path and local scanner remain independently operational. Do not restore preflight-only bearer delivery as a fallback. + +## Response receipt validation + +A successful HTTP status does not make the response body trusted. The CLI accepts a delivery receipt only when the body is a JSON object containing a positive, non-Boolean integer scan identifier. The optional `new_blocking` field must be a nonnegative, non-Boolean integer. Strings, terminal control sequences, negative values, Boolean values, nested objects, and malformed JSON are rejected without copying their contents into terminal output. + +The base control-plane URL is also treated as untrusted configuration. It must be public HTTPS and must not contain user information, a query string, or a fragment. Rejected values are not echoed because they may contain credentials or attacker-controlled terminal sequences. Endpoint construction appends `/api/v1/scans` to the validated base path through structured URI composition rather than string concatenation across an ambiguous query or fragment boundary. + +## Verification + +The dedicated coverage workflow runs deterministic tests for: + +- direct and mixed DNS answers; +- every non-global address class; +- IPv4 and IPv6 socket pinning; +- TLS hostname preservation; +- fallback across only prevalidated addresses; +- initial and redirected DNS rebinding boundaries; +- relative and cross-origin redirects; +- case-insensitive credential removal; +- unsupported redirects and redirect loops; +- malformed headers, URLs, bounds, JSON, and response sizes; +- ambiguous base URLs and forged response receipts; +- scanner CLI payload and non-secret error behavior; and +- public package exports, workflow permissions, documentation, and changelog evidence. + +Changed production statements in `appguardrail_core/pinned_https.py` must remain at exact unrounded 100% coverage. Public and non-obvious behavior is documented in module, class, method, and function docstrings. + +## References + +Berners-Lee, T., Fielding, R., & Masinter, L. (2005). *Uniform Resource Identifier (URI): Generic syntax* (RFC 3986). Internet Engineering Task Force. https://www.rfc-editor.org/rfc/rfc3986 + +Fielding, R., Nottingham, M., & Reschke, J. (2022). *HTTP semantics* (RFC 9110). Internet Engineering Task Force. https://www.rfc-editor.org/rfc/rfc9110 + +Open Worldwide Application Security Project. (2026). *Server side request forgery prevention cheat sheet*. OWASP Cheat Sheet Series. https://cheatsheetseries.owasp.org/cheatsheets/Server_Side_Request_Forgery_Prevention_Cheat_Sheet.html + +Python Software Foundation. (2026a). *http.client—HTTP protocol client* (Python 3.13 documentation). https://docs.python.org/3.13/library/http.client.html + +Python Software Foundation. (2026b). *socket—Low-level networking interface* (Python 3.13 documentation). https://docs.python.org/3.13/library/socket.html + +Python Software Foundation. (2026c). *ssl—TLS/SSL wrapper for socket objects* (Python 3.13 documentation). https://docs.python.org/3.13/library/ssl.html + +Saint-Andre, P., & Hodges, J. (2023). *Service identity in TLS* (RFC 9525). Internet Engineering Task Force. https://www.rfc-editor.org/rfc/rfc9525 diff --git a/scanner/cli/appguardrail.py b/scanner/cli/appguardrail.py index a62ecb2ff..c655195aa 100644 --- a/scanner/cli/appguardrail.py +++ b/scanner/cli/appguardrail.py @@ -59,6 +59,11 @@ from appguardrail_core.config import load_config from appguardrail_core.controlplane import SafeRedirectHandler +from appguardrail_core.pinned_https import ( + DestinationValidationError, + PinnedHTTPSFailure, + post_json_pinned_https, +) from appguardrail_core.external import build_external_scan_plan from appguardrail_core.findings import NON_BLOCKING_CONTEXTS from appguardrail_core.findings import is_deploy_blocking as core_is_deploy_blocking @@ -1671,8 +1676,8 @@ def is_bad_ip(ip) -> bool: def _push_findings(url, findings): - """POST normalized findings to a control-plane /api/v1/scans endpoint.""" - import urllib.request + """POST normalized findings through DNS-pinned public HTTPS.""" + import urllib.parse api_key = os.environ.get("APPGUARDRAIL_API_KEY", "") if not api_key: @@ -1681,48 +1686,102 @@ def _push_findings(url, findings): file=sys.stderr, ) return - if not _is_safe_url(url): + try: + parsed = urllib.parse.urlsplit(url) + hostname = parsed.hostname + parsed.port + except (TypeError, ValueError): + parsed = None + hostname = None + if ( + parsed is None + or parsed.scheme.lower() != "https" + or not hostname + or parsed.username is not None + or parsed.password is not None + or bool(parsed.query) + or bool(parsed.fragment) + ): _console_print( - f"⚠️ --push URL must be a valid http/https URL and not point to internal infrastructure, got {url}", + "⚠️ --push URL must be a public HTTPS URL without credentials, " + "query, or fragment; skipping push.", file=sys.stderr, ) return + + base_path = parsed.path.rstrip("/") + endpoint_path = f"{base_path}/api/v1/scans" if base_path else "/api/v1/scans" + endpoint = urllib.parse.urlunsplit( + ("https", parsed.netloc, endpoint_path, "", "") + ) payload = { "findings": list(normalize_findings(findings)), "repo": os.environ.get("GITHUB_REPOSITORY"), "commit": os.environ.get("GITHUB_SHA"), } - endpoint = url.rstrip("/") + "/api/v1/scans" - req = urllib.request.Request( # noqa: S310 - Safe URL scheme validated - endpoint, - data=json.dumps(payload).encode("utf-8"), - method="POST", - headers={ - "Content-Type": "application/json", - "Authorization": f"Bearer {api_key}", - }, - ) try: - opener = urllib.request.build_opener(SafeRedirectHandler()) - with ( - opener.open( # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected - req, timeout=15 - ) as resp - ): # noqa: S310 - Safe URL scheme validated - body = json.loads(resp.read() or b"{}") - drift = body.get("new_blocking") - extra = f", {drift} newly deploy-blocking" if drift else "" - _console_print(f"📡 Pushed scan #{body.get('id')} to control plane{extra}.") - except urllib.error.HTTPError as exc: + response = post_json_pinned_https( + endpoint, + payload, + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {api_key}", + }, + timeout=15, + ) + except DestinationValidationError: + _console_print( + "⚠️ --push URL must be a public HTTPS URL without credentials, " + "query, or fragment; skipping push.", + file=sys.stderr, + ) + return + except PinnedHTTPSFailure: _console_print( - f"⚠️ Control-plane push failed ({exc.code}); scan still completed.", + "⚠️ Control-plane push failed; scan still completed.", file=sys.stderr, ) - except (urllib.error.URLError, OSError, ValueError) as exc: + return + + if not 200 <= response.status < 300: _console_print( - f"⚠️ Control-plane push failed ({exc}); scan still completed.", + f"⚠️ Control-plane push failed ({response.status}); scan still completed.", file=sys.stderr, ) + return + try: + body = json.loads(response.body or b"{}") + except (UnicodeDecodeError, json.JSONDecodeError, RecursionError, ValueError): + _console_print( + "⚠️ Control-plane push returned an invalid response; scan still completed.", + file=sys.stderr, + ) + return + if not isinstance(body, dict): + _console_print( + "⚠️ Control-plane push returned an invalid response; scan still completed.", + file=sys.stderr, + ) + return + + scan_id = body.get("id") + new_blocking = body.get("new_blocking", 0) + if ( + not isinstance(scan_id, int) + or isinstance(scan_id, bool) + or scan_id <= 0 + or not isinstance(new_blocking, int) + or isinstance(new_blocking, bool) + or new_blocking < 0 + ): + _console_print( + "⚠️ Control-plane push returned an invalid response; scan still completed.", + file=sys.stderr, + ) + return + + extra = f", {new_blocking} newly deploy-blocking" if new_blocking else "" + _console_print(f"📡 Pushed scan #{scan_id} to control plane{extra}.") def _write_sarif(findings, output_path: Path): diff --git a/tests/test_pinned_https_cli_integration.py b/tests/test_pinned_https_cli_integration.py new file mode 100644 index 000000000..f3bef3d23 --- /dev/null +++ b/tests/test_pinned_https_cli_integration.py @@ -0,0 +1,197 @@ +"""CLI integration contracts for bearer-authenticated pinned HTTPS delivery.""" + +from __future__ import annotations + +import json + +import pytest + +from appguardrail_core.pinned_https import PinnedHTTPSFailure, PinnedHTTPSResponse +from scanner.cli import appguardrail as cli + + +@pytest.mark.parametrize( + "url", + [ + "http://api.example.com", + "https://user:secret@api.example.com/root", + "https://api.example.com/root?token=secret-query", + "https://api.example.com/root#secret-fragment", + "https://[broken", + ], +) +def test_push_findings_requires_unambiguous_public_https_before_transport( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + url: str, +) -> None: + """Unsafe base identities are rejected without echoing attacker-controlled text.""" + monkeypatch.setenv("APPGUARDRAIL_API_KEY", "credential-value") + monkeypatch.setattr( + cli, + "post_json_pinned_https", + lambda *_args, **_kwargs: pytest.fail("transport must not run"), + ) + + cli._push_findings(url, []) + + captured = capsys.readouterr() + assert "public HTTPS URL" in captured.err + assert url not in captured.err + assert "secret" not in captured.err + + +def test_push_findings_sends_normalized_payload_through_pinned_transport( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """The CLI supplies one bounded authenticated request and renders its receipt.""" + observed: dict[str, object] = {} + + def deliver( + url: str, + payload: dict[str, object], + *, + headers: dict[str, str], + timeout: float, + ) -> PinnedHTTPSResponse: + observed.update( + url=url, + payload=payload, + headers=headers, + timeout=timeout, + ) + return PinnedHTTPSResponse( + status=201, + reason="Created", + headers=(("Content-Type", "application/json"),), + body=json.dumps({"id": 41, "new_blocking": 2}).encode("utf-8"), + ) + + monkeypatch.setenv("APPGUARDRAIL_API_KEY", "credential-value") + monkeypatch.setenv("GITHUB_REPOSITORY", "ContextualWisdomLab/appguardrail") + monkeypatch.setenv("GITHUB_SHA", "a" * 40) + monkeypatch.setattr(cli, "post_json_pinned_https", deliver) + + cli._push_findings( + "https://api.example.com/root/", + [ + { + "rule_id": "example-rule", + "severity": "HIGH", + "message": "example", + "file": "src/app.py", + "line": 1, + "snippet": "", + "source": "test", + } + ], + ) + + assert observed["url"] == "https://api.example.com/root/api/v1/scans" + assert observed["timeout"] == 15 + headers = observed["headers"] + assert isinstance(headers, dict) + assert headers["Authorization"] == "Bearer credential-value" + assert headers["Content-Type"] == "application/json" + payload = observed["payload"] + assert isinstance(payload, dict) + assert payload["repo"] == "ContextualWisdomLab/appguardrail" + assert payload["commit"] == "a" * 40 + assert len(payload["findings"]) == 1 + captured = capsys.readouterr() + assert "Pushed scan #41" in captured.out + assert "2 newly deploy-blocking" in captured.out + + +def test_push_findings_reports_non_success_without_copying_response_body( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """A remote error exposes only its status and never an untrusted response body.""" + monkeypatch.setenv("APPGUARDRAIL_API_KEY", "credential-value") + monkeypatch.setattr( + cli, + "post_json_pinned_https", + lambda *_args, **_kwargs: PinnedHTTPSResponse( + status=503, + reason="Service Unavailable", + headers=(), + body=b"internal tenant data", + ), + ) + + cli._push_findings("https://api.example.com", []) + + captured = capsys.readouterr() + assert "(503)" in captured.err + assert "internal tenant data" not in captured.err + + +def test_push_findings_bounds_transport_and_response_parse_failures( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """Network and malformed receipt failures remain non-secret operational messages.""" + monkeypatch.setenv("APPGUARDRAIL_API_KEY", "credential-value") + + def fail_transport(*_args: object, **_kwargs: object) -> PinnedHTTPSResponse: + raise PinnedHTTPSFailure("private implementation detail") + + monkeypatch.setattr(cli, "post_json_pinned_https", fail_transport) + cli._push_findings("https://api.example.com", []) + first = capsys.readouterr() + assert "Control-plane push failed" in first.err + assert "private implementation detail" not in first.err + + monkeypatch.setattr( + cli, + "post_json_pinned_https", + lambda *_args, **_kwargs: PinnedHTTPSResponse( + status=200, + reason="OK", + headers=(), + body=b"not-json", + ), + ) + cli._push_findings("https://api.example.com", []) + second = capsys.readouterr() + assert "invalid response" in second.err.lower() + + +@pytest.mark.parametrize( + "receipt", + [ + {"id": "\x1b[31mforged", "new_blocking": 0}, + {"id": 0, "new_blocking": 0}, + {"id": True, "new_blocking": 0}, + {"id": 41, "new_blocking": "\x1b[31mforged"}, + {"id": 41, "new_blocking": -1}, + {"id": 41, "new_blocking": True}, + ], +) +def test_push_findings_rejects_untrusted_receipt_metadata( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + receipt: dict[str, object], +) -> None: + """Untrusted response fields cannot become terminal output or forged receipts.""" + monkeypatch.setenv("APPGUARDRAIL_API_KEY", "credential-value") + monkeypatch.setattr( + cli, + "post_json_pinned_https", + lambda *_args, **_kwargs: PinnedHTTPSResponse( + status=201, + reason="Created", + headers=(), + body=json.dumps(receipt).encode("utf-8"), + ), + ) + + cli._push_findings("https://api.example.com", []) + + captured = capsys.readouterr() + assert captured.out == "" + assert "invalid response" in captured.err.lower() + assert "forged" not in captured.err + assert "\x1b" not in captured.err diff --git a/tests/test_pinned_https_coverage_edges.py b/tests/test_pinned_https_coverage_edges.py new file mode 100644 index 000000000..092698f3f --- /dev/null +++ b/tests/test_pinned_https_coverage_edges.py @@ -0,0 +1,150 @@ +"""Exact statement-coverage edges for DNS-pinned HTTPS delivery.""" + +from __future__ import annotations + +import socket +import urllib.parse + +import pytest + +from appguardrail_core import pinned_https as transport + + +PUBLIC_IPV4 = "8.8.8.8" + + +def _answer( + ip_address: str, + port: int = 443, + *, + family: int = socket.AF_INET, + socket_address: tuple[object, ...] | None = None, +): + """Return one TCP resolver answer with optionally hostile socket metadata.""" + return ( + family, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + socket_address + if socket_address is not None + else ( + (ip_address, port, 0, 0) + if family == socket.AF_INET6 + else (ip_address, port) + ), + ) + + +def _resolver_with(*answers: tuple[object, ...]): + """Return a deterministic resolver containing the supplied answers.""" + + def resolve(*_args: object): + return list(answers) + + return resolve + + +def test_hostname_rejects_overlong_identity_after_idna_normalization() -> None: + """A syntactically encodable hostname cannot exceed the DNS identity bound.""" + hostname = "a." * 127 + "a" + assert len(hostname) == 255 + + with pytest.raises(transport.DestinationValidationError, match="hostname is invalid"): + transport.resolve_public_https_destination( + f"https://{hostname}/path", + resolver=lambda *_args: pytest.fail("resolver must not run"), + ) + + +@pytest.mark.parametrize("url", ["https://127.0.0.1/", "https://[::1]/"]) +def test_non_global_ip_literal_is_rejected_before_resolution(url: str) -> None: + """A private or loopback literal cannot rely on a later resolver decision.""" + with pytest.raises(transport.DestinationValidationError, match="non-global"): + transport.resolve_public_https_destination( + url, + resolver=lambda *_args: pytest.fail("resolver must not run"), + ) + + +def test_request_target_normalizer_handles_relative_defensive_input() -> None: + """The defensive target helper always returns HTTP origin-form syntax.""" + parts = urllib.parse.SplitResult( + scheme="https", + netloc="example.com", + path="relative path", + query="value=one two", + fragment="", + ) + + assert transport._quoted_request_target(parts) == "/relative%20path?value=one%20two" + + +def test_resolver_answer_with_short_socket_tuple_is_ignored() -> None: + """An incomplete sockaddr never becomes a connectable validated address.""" + with pytest.raises(transport.DestinationValidationError, match="no TCP addresses"): + transport.resolve_public_https_destination( + "https://example.com", + resolver=_resolver_with( + _answer(PUBLIC_IPV4, socket_address=(PUBLIC_IPV4,)) + ), + ) + + +def test_resolver_answer_with_textual_zone_is_rejected() -> None: + """A percent-encoded or textual interface scope cannot bypass zone rejection.""" + with pytest.raises(transport.DestinationValidationError, match="zone-scoped"): + transport.resolve_public_https_destination( + "https://example.com", + resolver=_resolver_with( + _answer( + "2606:4700:4700::1111%eth0", + family=socket.AF_INET6, + socket_address=("2606:4700:4700::1111%eth0", 443, 0, 0), + ) + ), + ) + + +def test_url_parser_failure_is_bounded() -> None: + """Malformed bracketed authority syntax is translated to a validation error.""" + with pytest.raises(transport.DestinationValidationError, match="URL is invalid"): + transport.resolve_public_https_destination( + "https://[broken/path", + resolver=lambda *_args: pytest.fail("resolver must not run"), + ) + + +@pytest.mark.parametrize( + "kwargs", + [ + {"timeout": "fast"}, + {"max_redirects": 1.5}, + {"max_response_bytes": 1.5}, + ], +) +def test_non_numeric_or_non_integer_bounds_fail_before_resolution( + kwargs: dict[str, object], +) -> None: + """Type-invalid resource bounds never reach JSON serialization or DNS.""" + with pytest.raises(ValueError): + transport.post_json_pinned_https( + "https://example.com", + {}, + resolver=lambda *_args: pytest.fail("resolver must not run"), + **kwargs, + ) + + +def test_default_connection_factory_returns_the_pinned_connection() -> None: + """The production factory preserves destination identity and timeout.""" + destination = transport.resolve_public_https_destination( + "https://example.com/path", + resolver=_resolver_with(_answer(PUBLIC_IPV4)), + ) + + connection = transport._default_connection_factory(destination, 9.5) + + assert isinstance(connection, transport.PinnedHTTPSConnection) + assert connection.destination is destination + assert connection.timeout == 9.5 diff --git a/tests/test_pinned_https_redirects.py b/tests/test_pinned_https_redirects.py new file mode 100644 index 000000000..137ea95a8 --- /dev/null +++ b/tests/test_pinned_https_redirects.py @@ -0,0 +1,324 @@ +"""Redirect and response contracts for DNS-pinned HTTPS delivery.""" + +from __future__ import annotations + +import json +import socket + +import pytest + +from appguardrail_core.pinned_https import ( + HTTPSDestination, + PinnedHTTPSFailure, + PinnedHTTPSResponse, + post_json_pinned_https, +) + + +PUBLIC_IPV4 = "8.8.8.8" +SECOND_PUBLIC_IPV4 = "1.1.1.1" + + +def _answer(ip_address: str, port: int = 443): + """Return one TCP IPv4 resolver result.""" + return ( + socket.AF_INET, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + (ip_address, port), + ) + + +def _resolver(host: str, port: int, *_args: object): + """Resolve the two fixture origins to distinct public addresses.""" + addresses = { + "api.example.com": PUBLIC_IPV4, + "other.example.com": SECOND_PUBLIC_IPV4, + } + return [_answer(addresses[host], port)] + + +class _Response: + """Small http.client-compatible response used by redirect tests.""" + + def __init__( + self, + status: int, + body: bytes = b"{}", + *, + reason: str = "", + headers: tuple[tuple[str, str], ...] = (), + ) -> None: + self.status = status + self.reason = reason + self._body = body + self._headers = headers + + def read(self, limit: int = -1) -> bytes: + """Return at most the requested number of bytes.""" + return self._body if limit < 0 else self._body[:limit] + + def getheaders(self) -> list[tuple[str, str]]: + """Return response headers in wire order.""" + return list(self._headers) + + def getheader(self, name: str, default: str | None = None) -> str | None: + """Return one response header using case-insensitive matching.""" + for header_name, value in self._headers: + if header_name.lower() == name.lower(): + return value + return default + + +class _ScriptedConnection: + """Capture one request and return one scripted response.""" + + def __init__( + self, + destination: HTTPSDestination, + timeout: float, + response: _Response, + requests: list[dict[str, object]], + ) -> None: + self.destination = destination + self.timeout = timeout + self.response = response + self.requests = requests + self.closed = False + + def request( + self, + method: str, + target: str, + body: bytes, + headers: dict[str, str], + ) -> None: + """Record method, target, body, headers, origin, and timeout.""" + self.requests.append( + { + "origin": self.destination.origin, + "method": method, + "target": target, + "body": body, + "headers": dict(headers), + "timeout": self.timeout, + } + ) + + def getresponse(self) -> _Response: + """Return the scripted response.""" + return self.response + + def close(self) -> None: + """Record deterministic connection cleanup.""" + self.closed = True + + +def _factory( + responses: list[_Response], + requests: list[dict[str, object]], + connections: list[_ScriptedConnection], +): + """Build a connection factory that consumes responses in order.""" + response_iter = iter(responses) + + def create(destination: HTTPSDestination, timeout: float) -> _ScriptedConnection: + connection = _ScriptedConnection( + destination, + timeout, + next(response_iter), + requests, + ) + connections.append(connection) + return connection + + return create + + +def _header_map(request: dict[str, object]) -> dict[str, str]: + """Normalize recorded request headers for case-insensitive assertions.""" + headers = request["headers"] + assert isinstance(headers, dict) + return {str(name).lower(): str(value) for name, value in headers.items()} + + +def test_cross_origin_307_removes_sensitive_headers_case_insensitively() -> None: + """Bearer and proxy authorization metadata never crosses an origin boundary.""" + requests: list[dict[str, object]] = [] + connections: list[_ScriptedConnection] = [] + response = post_json_pinned_https( + "https://api.example.com/api/v1/scans", + {"findings": []}, + headers={ + "Content-Type": "application/json", + "aUtHoRiZaTiOn": "Bearer credential-value", + "pRoXy-AuThOrIzAtIoN": "Basic proxy-value", + "X-Request-ID": "request-1", + }, + resolver=_resolver, + connection_factory=_factory( + [ + _Response( + 307, + headers=(("Location", "https://other.example.com/ingest"),), + ), + _Response(201, body=b'{"id": 7}'), + ], + requests, + connections, + ), + ) + + assert response.status == 201 + assert json.loads(response.body) == {"id": 7} + assert len(requests) == 2 + assert "authorization" in _header_map(requests[0]) + assert "proxy-authorization" in _header_map(requests[0]) + redirected_headers = _header_map(requests[1]) + assert "authorization" not in redirected_headers + assert "proxy-authorization" not in redirected_headers + assert redirected_headers["content-type"] == "application/json" + assert redirected_headers["x-request-id"] == "request-1" + assert requests[1]["target"] == "/ingest" + assert all(connection.closed for connection in connections) + + +def test_same_origin_308_preserves_authorization_and_relative_location() -> None: + """A method-preserving redirect on the same HTTPS origin retains credentials.""" + requests: list[dict[str, object]] = [] + connections: list[_ScriptedConnection] = [] + response = post_json_pinned_https( + "https://api.example.com/v1/scans", + {"findings": []}, + headers={"Authorization": "Bearer credential-value"}, + resolver=_resolver, + connection_factory=_factory( + [ + _Response(308, headers=(("location", "/v2/scans?source=v1"),)), + _Response(202, body=b'{"accepted": true}'), + ], + requests, + connections, + ), + ) + + assert response.status == 202 + assert _header_map(requests[1])["authorization"] == "Bearer credential-value" + assert requests[1]["target"] == "/v2/scans?source=v1" + + +@pytest.mark.parametrize("status", [301, 302, 303, 305, 306]) +def test_unsupported_redirect_statuses_fail_closed(status: int) -> None: + """Credential-bearing POST delivery follows only 307 and 308 semantics.""" + with pytest.raises(PinnedHTTPSFailure, match=f"redirect status {status}"): + post_json_pinned_https( + "https://api.example.com/v1/scans", + {}, + headers={"Authorization": "Bearer credential-value"}, + resolver=_resolver, + connection_factory=_factory( + [_Response(status, headers=(("Location", "/elsewhere"),))], + [], + [], + ), + ) + + +def test_redirect_requires_location_and_respects_maximum_hops() -> None: + """Missing targets and redirect loops terminate with bounded errors.""" + with pytest.raises(PinnedHTTPSFailure, match="missing Location"): + post_json_pinned_https( + "https://api.example.com/v1/scans", + {}, + headers={}, + resolver=_resolver, + connection_factory=_factory([_Response(307)], [], []), + ) + + with pytest.raises(PinnedHTTPSFailure, match="too many redirects"): + post_json_pinned_https( + "https://api.example.com/v1/scans", + {}, + headers={}, + max_redirects=1, + resolver=_resolver, + connection_factory=_factory( + [ + _Response(307, headers=(("Location", "/two"),)), + _Response(307, headers=(("Location", "/three"),)), + ], + [], + [], + ), + ) + + +def test_response_size_is_bounded_and_connection_closes_on_error() -> None: + """An oversized response cannot exhaust the scanner or leak the connection.""" + connections: list[_ScriptedConnection] = [] + with pytest.raises(PinnedHTTPSFailure, match="response exceeds"): + post_json_pinned_https( + "https://api.example.com/v1/scans", + {}, + headers={}, + max_response_bytes=4, + resolver=_resolver, + connection_factory=_factory( + [_Response(200, body=b"12345")], + [], + connections, + ), + ) + + assert connections[0].closed + + +def test_response_header_lookup_is_case_insensitive() -> None: + """Callers inspect bounded response metadata without depending on casing.""" + response = PinnedHTTPSResponse( + status=202, + reason="Accepted", + headers=(("Content-Type", "application/json"),), + body=b"{}", + ) + + assert response.get_header("content-type") == "application/json" + assert response.get_header("missing") is None + + +@pytest.mark.parametrize( + ("kwargs", "message"), + [ + ({"timeout": 0}, "timeout"), + ({"timeout": True}, "timeout"), + ({"max_redirects": -1}, "max_redirects"), + ({"max_redirects": True}, "max_redirects"), + ({"max_response_bytes": 0}, "max_response_bytes"), + ({"max_response_bytes": True}, "max_response_bytes"), + ], +) +def test_transport_bounds_are_validated_before_resolution( + kwargs: dict[str, object], + message: str, +) -> None: + """Invalid resource bounds fail before DNS or credential-bearing I/O.""" + with pytest.raises(ValueError, match=message): + post_json_pinned_https( + "https://api.example.com/v1/scans", + {}, + headers={}, + resolver=lambda *_args: pytest.fail("resolver must not run"), + **kwargs, + ) + + +def test_unserializable_json_is_a_bounded_transport_failure() -> None: + """Serialization errors do not reach DNS or expose Python tracebacks.""" + with pytest.raises(PinnedHTTPSFailure, match="JSON serialization"): + post_json_pinned_https( + "https://api.example.com/v1/scans", + {"bad": object()}, + headers={}, + resolver=lambda *_args: pytest.fail("resolver must not run"), + ) diff --git a/tests/test_pinned_https_release_contract.py b/tests/test_pinned_https_release_contract.py new file mode 100644 index 000000000..17e08c795 --- /dev/null +++ b/tests/test_pinned_https_release_contract.py @@ -0,0 +1,84 @@ +"""Release and documentation contracts for DNS-pinned HTTPS delivery.""" + +from __future__ import annotations + +from pathlib import Path + +import appguardrail_core + + +ROOT = Path(__file__).resolve().parents[1] +WORKFLOW = ROOT / ".github" / "workflows" / "pinned-https-coverage.yml" +DOCUMENTATION = ROOT / "docs" / "pinned-https-control-plane-delivery.md" +CHANGELOG = ROOT / "CHANGELOG.d" / "892-pinned-control-plane-delivery.md" + + +def test_core_package_exports_pinned_https_boundary() -> None: + """Standalone and MSA consumers receive the reviewed transport API.""" + expected = { + "DestinationValidationError", + "HTTPSDestination", + "PinnedHTTPSConnection", + "PinnedHTTPSFailure", + "PinnedHTTPSResponse", + "ResolvedAddress", + "post_json_pinned_https", + "resolve_public_https_destination", + } + + assert expected <= set(appguardrail_core.__all__) + assert all(hasattr(appguardrail_core, name) for name in expected) + + +def test_pinned_https_workflow_enforces_exact_read_only_coverage() -> None: + """The transport has an immutable least-privilege exact-coverage gate.""" + workflow = WORKFLOW.read_text(encoding="utf-8") + + assert "pull_request:" in workflow + assert "push:" in workflow + assert "permissions:\n contents: read" in workflow + assert "python-version: '3.13'" in workflow + assert "persist-credentials: false" in workflow + assert "appguardrail_core/pinned_https.py" in workflow + for test_name in ( + "test_pinned_https_resolution.py", + "test_pinned_https_redirects.py", + "test_pinned_https_cli_integration.py", + "test_pinned_https_validation_edges.py", + "test_pinned_https_coverage_edges.py", + "test_pinned_https_release_contract.py", + ): + assert test_name in workflow + + +def test_operator_documentation_records_security_boundary_and_apa_sources() -> None: + """Operators receive a complete architecture, rollback, and source record.""" + documentation = DOCUMENTATION.read_text(encoding="utf-8") + + required = ( + "DNS rebinding", + "TLS SNI", + "certificate", + "307", + "308", + "Proxy-Authorization", + "egress", + "naruon", + "## References", + "RFC 3986", + "RFC 9110", + "RFC 9525", + "Python Software Foundation", + ) + assert all(item in documentation for item in required) + assert "```mermaid" in documentation + + +def test_changelog_records_buyer_visible_transport_hardening() -> None: + """The release fragment describes the user-visible security guarantee.""" + changelog = CHANGELOG.read_text(encoding="utf-8") + + assert changelog.startswith("### Security\n") + assert "DNS-pinned" in changelog + assert "Authorization" in changelog + assert "Proxy-Authorization" in changelog diff --git a/tests/test_pinned_https_resolution.py b/tests/test_pinned_https_resolution.py new file mode 100644 index 000000000..71c8f32b7 --- /dev/null +++ b/tests/test_pinned_https_resolution.py @@ -0,0 +1,285 @@ +"""Resolution and socket contracts for DNS-pinned HTTPS delivery.""" + +from __future__ import annotations + +import socket +import ssl +from collections.abc import Callable +from typing import Any + +import pytest + +from appguardrail_core.pinned_https import ( + DestinationValidationError, + HTTPSDestination, + PinnedHTTPSConnection, + resolve_public_https_destination, +) + + +PUBLIC_IPV4 = "8.8.8.8" +SECOND_PUBLIC_IPV4 = "1.1.1.1" +PUBLIC_IPV6 = "2606:4700:4700::1111" + + +def _answer( + ip_address: str, + port: int = 443, +) -> tuple[int, int, int, str, tuple[Any, ...]]: + """Return one TCP resolver result for an IPv4 or IPv6 address.""" + if ":" in ip_address: + return ( + socket.AF_INET6, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + (ip_address, port, 0, 0), + ) + return ( + socket.AF_INET, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + (ip_address, port), + ) + + +def _resolver_for( + answers_by_host: dict[str, list[tuple[int, int, int, str, tuple[Any, ...]]]], + calls: list[tuple[object, ...]] | None = None, +) -> Callable[..., list[tuple[int, int, int, str, tuple[Any, ...]]]]: + """Build a deterministic resolver and optionally record exact arguments.""" + + def resolve( + host: str, + port: int, + family: int, + socket_type: int, + protocol: int, + ) -> list[tuple[int, int, int, str, tuple[Any, ...]]]: + if calls is not None: + calls.append((host, port, family, socket_type, protocol)) + return answers_by_host[host] + + return resolve + + +def test_destination_pins_normalized_public_tcp_answers() -> None: + """Resolution produces one stable origin, target, and deduplicated address set.""" + calls: list[tuple[object, ...]] = [] + destination = resolve_public_https_destination( + "https://EXAMPLE.com:8443/api/v1/scans?mode=full", + resolver=_resolver_for( + { + "example.com": [ + _answer(PUBLIC_IPV4, 8443), + _answer(PUBLIC_IPV6, 8443), + _answer(PUBLIC_IPV4, 8443), + ] + }, + calls, + ), + ) + + assert destination.hostname == "example.com" + assert destination.port == 8443 + assert destination.request_target == "/api/v1/scans?mode=full" + assert destination.origin == ("https", "example.com", 8443) + assert [address.ip_address for address in destination.addresses] == [ + PUBLIC_IPV4, + PUBLIC_IPV6, + ] + assert calls == [ + ( + "example.com", + 8443, + socket.AF_UNSPEC, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + ) + ] + + +@pytest.mark.parametrize( + "url", + [ + "http://example.com/api", + "file:///tmp/report", + "https://user:password@example.com/api", + "https://example.com/api#fragment", + "https://[fe80::1%25eth0]/api", + "https://example.com:0/api", + "https://example.com:65536/api", + "https:///missing-host", + ], +) +def test_destination_rejects_ambiguous_or_non_https_identity(url: str) -> None: + """Credential-bearing requests require an unambiguous public HTTPS identity.""" + with pytest.raises(DestinationValidationError): + resolve_public_https_destination( + url, + resolver=_resolver_for({"example.com": [_answer(PUBLIC_IPV4)]}), + ) + + +def test_destination_rejects_failed_empty_and_mixed_resolution() -> None: + """DNS failure, no answers, or one non-global answer rejects the destination.""" + + def unavailable(*_args: object) -> list[object]: + raise socket.gaierror("not found") + + with pytest.raises(DestinationValidationError, match="resolution failed"): + resolve_public_https_destination("https://example.com", resolver=unavailable) + + with pytest.raises(DestinationValidationError, match="no TCP addresses"): + resolve_public_https_destination( + "https://example.com", + resolver=_resolver_for({"example.com": []}), + ) + + with pytest.raises(DestinationValidationError, match="non-global"): + resolve_public_https_destination( + "https://example.com", + resolver=_resolver_for( + {"example.com": [_answer(PUBLIC_IPV4), _answer("127.0.0.1")]} + ), + ) + + +@pytest.mark.parametrize( + "ip_address", + [ + "10.0.0.1", + "127.0.0.1", + "169.254.169.254", + "224.0.0.1", + "0.0.0.0", + "255.255.255.255", + "::1", + "::", + "ff02::1", + "::ffff:127.0.0.1", + ], +) +def test_destination_rejects_non_global_address_classes(ip_address: str) -> None: + """Private, local, metadata, multicast, reserved, and mapped IPs fail closed.""" + with pytest.raises(DestinationValidationError, match="non-global"): + resolve_public_https_destination( + "https://example.com", + resolver=_resolver_for({"example.com": [_answer(ip_address)]}), + ) + + +class _RawSocket: + """Record one direct socket connection used by the pinned TLS connection.""" + + def __init__(self, *, failure: OSError | None = None) -> None: + self.failure = failure + self.timeout: float | None = None + self.connected_to: tuple[Any, ...] | None = None + self.closed = False + + def settimeout(self, timeout: float | None) -> None: + """Record the requested timeout.""" + self.timeout = timeout + + def connect(self, socket_address: tuple[Any, ...]) -> None: + """Connect to the captured address or raise the fixture error.""" + self.connected_to = socket_address + if self.failure is not None: + raise self.failure + + def close(self) -> None: + """Record cleanup of failed raw sockets.""" + self.closed = True + + +class _TLSContext: + """Record the original hostname passed to TLS verification and SNI.""" + + def __init__(self) -> None: + self.verify_mode = ssl.CERT_REQUIRED + self.check_hostname = True + self.server_hostnames: list[str] = [] + + def wrap_socket(self, raw_socket: _RawSocket, *, server_hostname: str): + """Return the fixture after recording the verified hostname.""" + self.server_hostnames.append(server_hostname) + return raw_socket + + +def _destination_with_addresses(*ip_addresses: str) -> HTTPSDestination: + """Build a destination through the production normalization boundary.""" + return resolve_public_https_destination( + "https://api.example.com:8443/api/v1/scans", + resolver=_resolver_for( + {"api.example.com": [_answer(value, 8443) for value in ip_addresses]} + ), + ) + + +def test_connection_uses_pinned_socket_and_original_tls_hostname() -> None: + """The TCP peer is pinned while SNI and certificate identity use the URL host.""" + destination = _destination_with_addresses(PUBLIC_IPV4) + created: list[tuple[int, int, int]] = [] + raw_socket = _RawSocket() + + def socket_factory(family: int, socket_type: int, protocol: int) -> _RawSocket: + created.append((family, socket_type, protocol)) + return raw_socket + + context = _TLSContext() + connection = PinnedHTTPSConnection( + destination, + timeout=7.5, + context=context, + socket_factory=socket_factory, + ) + connection.connect() + + assert created == [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)] + assert raw_socket.timeout == 7.5 + assert raw_socket.connected_to == (PUBLIC_IPV4, 8443) + assert context.server_hostnames == ["api.example.com"] + assert connection.sock is raw_socket + + +def test_connection_tries_only_prevalidated_addresses() -> None: + """Fallback advances through the captured address set without re-resolution.""" + destination = _destination_with_addresses(PUBLIC_IPV4, SECOND_PUBLIC_IPV4) + first = _RawSocket(failure=OSError("unreachable")) + second = _RawSocket() + sockets = iter((first, second)) + context = _TLSContext() + + connection = PinnedHTTPSConnection( + destination, + timeout=3, + context=context, + socket_factory=lambda *_args: next(sockets), + ) + connection.connect() + + assert first.connected_to == (PUBLIC_IPV4, 8443) + assert first.closed + assert second.connected_to == (SECOND_PUBLIC_IPV4, 8443) + assert context.server_hostnames == ["api.example.com"] + + +def test_connection_closes_failed_sockets_and_raises_last_error() -> None: + """Exhausting the captured set raises instead of performing another lookup.""" + destination = _destination_with_addresses(PUBLIC_IPV4, SECOND_PUBLIC_IPV4) + first = _RawSocket(failure=OSError("first")) + second = _RawSocket(failure=OSError("second")) + sockets = iter((first, second)) + connection = PinnedHTTPSConnection( + destination, + timeout=3, + context=_TLSContext(), + socket_factory=lambda *_args: next(sockets), + ) + + with pytest.raises(OSError, match="second"): + connection.connect() + + assert first.closed and second.closed diff --git a/tests/test_pinned_https_validation_edges.py b/tests/test_pinned_https_validation_edges.py new file mode 100644 index 000000000..c2988cf74 --- /dev/null +++ b/tests/test_pinned_https_validation_edges.py @@ -0,0 +1,402 @@ +"""Edge and failure contracts for DNS-pinned HTTPS delivery.""" + +from __future__ import annotations + +import http.client +import socket +import ssl +from collections.abc import Iterator, Mapping +from typing import Any + +import pytest + +from appguardrail_core import pinned_https as transport + + +PUBLIC_IPV4 = "8.8.8.8" + + +def _answer( + ip_address: str, + port: int = 443, + *, + family: int = socket.AF_INET, + socket_type: int = socket.SOCK_STREAM, + protocol: int = socket.IPPROTO_TCP, + scope_id: int = 0, +) -> tuple[int, int, int, str, tuple[Any, ...]]: + """Return one configurable resolver answer.""" + socket_address: tuple[Any, ...] + if family == socket.AF_INET6: + socket_address = (ip_address, port, 0, scope_id) + else: + socket_address = (ip_address, port) + return family, socket_type, protocol, "", socket_address + + +def _resolver_with(*answers: tuple[int, int, int, str, tuple[Any, ...]]): + """Return a deterministic resolver containing the supplied answers.""" + + def resolve(*_args: object): + return list(answers) + + return resolve + + +@pytest.mark.parametrize( + "url", + [ + "", + "https://example.com/\nnext", + "https://example.com:invalid/path", + "https://example.com./path#fragment", + ], +) +def test_destination_rejects_invalid_url_forms(url: str) -> None: + """Malformed URL identity fails before resolver output can be trusted.""" + with pytest.raises(transport.DestinationValidationError): + transport.resolve_public_https_destination( + url, + resolver=_resolver_with(_answer(PUBLIC_IPV4)), + ) + + +def test_destination_rejects_non_string_and_invalid_idna() -> None: + """Non-text and unencodable host identities fail closed.""" + with pytest.raises(transport.DestinationValidationError): + transport.resolve_public_https_destination( # type: ignore[arg-type] + None, + resolver=_resolver_with(_answer(PUBLIC_IPV4)), + ) + with pytest.raises(transport.DestinationValidationError, match="IDNA"): + transport.resolve_public_https_destination( + "https://\udcff.example/path", + resolver=_resolver_with(_answer(PUBLIC_IPV4)), + ) + + +def test_destination_supports_public_ip_literal_and_default_target() -> None: + """A public IP literal remains strict HTTPS and uses the root request target.""" + destination = transport.resolve_public_https_destination( + "https://8.8.8.8", + resolver=_resolver_with(_answer(PUBLIC_IPV4)), + ) + + assert destination.hostname == PUBLIC_IPV4 + assert destination.port == 443 + assert destination.request_target == "/" + assert destination.url == "https://8.8.8.8/" + + +def test_destination_skips_non_tcp_results_but_rejects_an_empty_tcp_set() -> None: + """Only IPv4/IPv6 TCP stream answers can enter the immutable address set.""" + invalid_answers = [ + (socket.AF_UNIX, socket.SOCK_STREAM, 0, "", ("ignored",)), + _answer(PUBLIC_IPV4, socket_type=socket.SOCK_DGRAM), + _answer(PUBLIC_IPV4, protocol=socket.IPPROTO_UDP), + (socket.AF_INET,), + ] + with pytest.raises(transport.DestinationValidationError, match="no TCP addresses"): + transport.resolve_public_https_destination( + "https://example.com", + resolver=lambda *_args: invalid_answers, + ) + + +@pytest.mark.parametrize( + ("answer", "message"), + [ + (_answer(PUBLIC_IPV4, port=444), "unexpected port"), + (_answer("not-an-ip"), "invalid IP"), + ( + _answer( + "2606:4700:4700::1111", + family=socket.AF_INET6, + scope_id=2, + ), + "zone-scoped", + ), + ], +) +def test_destination_rejects_hostile_resolver_metadata( + answer: tuple[int, int, int, str, tuple[Any, ...]], + message: str, +) -> None: + """Resolver-controlled port, address, and scope metadata cannot widen trust.""" + with pytest.raises(transport.DestinationValidationError, match=message): + transport.resolve_public_https_destination( + "https://example.com", + resolver=_resolver_with(answer), + ) + + +class _Socket: + """Record bind, connect, timeout, and close behavior.""" + + def __init__(self, *, wrap_failure: OSError | None = None) -> None: + self.wrap_failure = wrap_failure + self.bound_to: tuple[str, int] | None = None + self.connected_to: tuple[Any, ...] | None = None + self.timeout: float | None = None + self.closed = False + + def settimeout(self, timeout: float | None) -> None: + self.timeout = timeout + + def bind(self, source_address: tuple[str, int]) -> None: + self.bound_to = source_address + + def connect(self, socket_address: tuple[Any, ...]) -> None: + self.connected_to = socket_address + + def close(self) -> None: + self.closed = True + + +class _Context: + """Return or reject a fixture TLS socket across supported Python versions.""" + + def __init__(self, *, failure: OSError | None = None) -> None: + self.verify_mode = ssl.CERT_REQUIRED + self.check_hostname = True + self.failure = failure + + def wrap_socket(self, raw_socket: _Socket, *, server_hostname: str) -> _Socket: + assert server_hostname == "example.com" + if self.failure is not None: + raise self.failure + return raw_socket + + +def _destination() -> transport.HTTPSDestination: + """Return one production-created destination for connection edge tests.""" + return transport.resolve_public_https_destination( + "https://example.com/path", + resolver=_resolver_with(_answer(PUBLIC_IPV4)), + ) + + +def test_connection_honors_source_address_and_rejects_tunneling() -> None: + """Pinned sockets preserve an explicit source bind and never enable a proxy tunnel.""" + raw_socket = _Socket() + connection = transport.PinnedHTTPSConnection( + _destination(), + timeout=4, + context=_Context(), + socket_factory=lambda *_args: raw_socket, + ) + connection.source_address = ("0.0.0.0", 0) + connection.connect() + + assert raw_socket.bound_to == ("0.0.0.0", 0) + assert raw_socket.connected_to == (PUBLIC_IPV4, 443) + + blocked = transport.PinnedHTTPSConnection( + _destination(), + context=_Context(), + socket_factory=lambda *_args: _Socket(), + ) + blocked._tunnel_host = "proxy.example" # noqa: SLF001 - contract edge + with pytest.raises(OSError, match="tunneling"): + blocked.connect() + + +def test_connection_closes_socket_when_tls_wrap_fails() -> None: + """A failed TLS handshake cannot leak the already connected raw socket.""" + raw_socket = _Socket() + connection = transport.PinnedHTTPSConnection( + _destination(), + context=_Context(failure=OSError("TLS failed")), + socket_factory=lambda *_args: raw_socket, + ) + + with pytest.raises(OSError, match="TLS failed"): + connection.connect() + + assert raw_socket.closed + + +def test_connection_rejects_manually_constructed_empty_address_set() -> None: + """Public dataclass construction cannot trigger an implicit fallback lookup.""" + destination = transport.HTTPSDestination( + url="https://example.com/", + hostname="example.com", + port=443, + request_target="/", + addresses=(), + ) + connection = transport.PinnedHTTPSConnection(destination) + + with pytest.raises(OSError, match="no validated addresses"): + connection.connect() + + +class _DuplicateHeaders(Mapping[str, str]): + """Mapping fixture exposing case-colliding names through item iteration.""" + + def __getitem__(self, key: str) -> str: + return {"Authorization": "first", "authorization": "second"}[key] + + def __iter__(self) -> Iterator[str]: + return iter(("Authorization", "authorization")) + + def __len__(self) -> int: + return 2 + + def items(self): + return (("Authorization", "first"), ("authorization", "second")) + + +@pytest.mark.parametrize( + "headers", + [ + {"Bad Header": "value"}, + {"X-Test": "line\nbreak"}, + {"Host": "attacker.example"}, + {"Content-Length": "1"}, + {"Transfer-Encoding": "chunked"}, + {"Connection": "close"}, + _DuplicateHeaders(), + {1: "value"}, # type: ignore[dict-item] + {"X-Test": 1}, # type: ignore[dict-item] + ], +) +def test_header_validation_fails_before_dns( + headers: Mapping[str, str], +) -> None: + """Ambiguous, smuggled, duplicate, and non-text headers fail before resolution.""" + with pytest.raises(ValueError): + transport.post_json_pinned_https( + "https://example.com", + {}, + headers=headers, + resolver=lambda *_args: pytest.fail("resolver must not run"), + ) + + +class _SimpleResponse: + """Provide one bounded final or redirect response.""" + + def __init__( + self, + status: int, + *, + body: bytes = b"{}", + reason: str = "", + headers: tuple[tuple[str, str], ...] = (), + ) -> None: + self.status = status + self.reason = reason + self.body = body + self.headers = headers + + def read(self, size: int) -> bytes: + return self.body[:size] + + def getheaders(self): + return list(self.headers) + + def getheader(self, name: str, default: str | None = None): + for header_name, value in self.headers: + if header_name.lower() == name.lower(): + return value + return default + + +class _Connection: + """Capture request defaults or raise one configured error.""" + + def __init__( + self, + response: _SimpleResponse, + observed: dict[str, object], + *, + failure: BaseException | None = None, + ) -> None: + self.response = response + self.observed = observed + self.failure = failure + self.closed = False + + def request(self, method: str, target: str, body: bytes, headers: dict[str, str]): + if self.failure is not None: + raise self.failure + self.observed.update(method=method, target=target, body=body, headers=headers) + + def getresponse(self): + return self.response + + def close(self) -> None: + self.closed = True + + +def test_transport_adds_json_defaults_and_accepts_exact_response_boundary() -> None: + """A final response at the byte limit is returned with deterministic request defaults.""" + observed: dict[str, object] = {} + connection = _Connection(_SimpleResponse(204, body=b"1234"), observed) + + result = transport.post_json_pinned_https( + "https://example.com/path", + {"value": 1}, + max_response_bytes=4, + resolver=_resolver_with(_answer(PUBLIC_IPV4)), + connection_factory=lambda *_args: connection, + ) + + assert result.status == 204 + assert result.body == b"1234" + headers = observed["headers"] + assert isinstance(headers, dict) + assert headers["Content-Type"] == "application/json" + assert headers["Accept"] == "application/json" + assert connection.closed + + +@pytest.mark.parametrize("failure", [OSError("socket"), http.client.HTTPException("protocol-private-detail")]) +def test_transport_wraps_connection_failures_without_details(failure: BaseException) -> None: + """Socket and protocol errors become one bounded transport category.""" + connection = _Connection(_SimpleResponse(200), {}, failure=failure) + + with pytest.raises(transport.PinnedHTTPSFailure, match="delivery failed") as error: + transport.post_json_pinned_https( + "https://example.com", + {}, + resolver=_resolver_with(_answer(PUBLIC_IPV4)), + connection_factory=lambda *_args: connection, + ) + + assert str(failure) not in str(error.value) + assert connection.closed + + +def test_redirect_rejects_control_characters_and_non_https_targets() -> None: + """Redirect metadata cannot inject a request line or downgrade transport security.""" + for location in ("/next\nInjected: value", "http://example.com/next"): + connection = _Connection( + _SimpleResponse(307, headers=(("Location", location),)), + {}, + ) + with pytest.raises(transport.PinnedHTTPSFailure): + transport.post_json_pinned_https( + "https://example.com/start", + {}, + resolver=_resolver_with(_answer(PUBLIC_IPV4)), + connection_factory=lambda *_args, connection=connection: connection, + ) + assert connection.closed + + +def test_zero_redirect_budget_rejects_the_first_redirect() -> None: + """A zero-hop policy does not permit a single method-preserving redirect.""" + connection = _Connection( + _SimpleResponse(308, headers=(("Location", "/next"),)), + {}, + ) + with pytest.raises(transport.PinnedHTTPSFailure, match="too many redirects"): + transport.post_json_pinned_https( + "https://example.com/start", + {}, + max_redirects=0, + resolver=_resolver_with(_answer(PUBLIC_IPV4)), + connection_factory=lambda *_args: connection, + ) diff --git a/tests/test_ssrf_protection.py b/tests/test_ssrf_protection.py index 551ff2026..3462f72f5 100644 --- a/tests/test_ssrf_protection.py +++ b/tests/test_ssrf_protection.py @@ -73,7 +73,7 @@ def test_push_findings_unsafe_url_handled_properly(monkeypatch, capsys): _push_findings("http://127.0.0.1/", []) captured = capsys.readouterr() assert ( - "URL must be a valid http/https URL and not point to internal infrastructure" + "URL must be a public HTTPS URL" in captured.err )