urllib3 2.7.0: connectionpool.HTTPSConnectionPool

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify urllib3.connectionpool.HTTPSConnectionPool in pkg:pypi/urllib3@2.7.0
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/urllib3@2.7.0
Demonstrate these symbols/APIs:
  - urllib3.connectionpool.HTTPSConnectionPool

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:4202a0b0eb3b226c4a6764fd706edfafd55bfd47b34b18880914a6a4a28122bb","contract":["urllib3.connectionpool.HTTPSConnectionPool sends HTTP GET requests over TLS and decodes response status, body, and headers","urllib3.connectionpool.HTTPSConnectionPool sends HTTP POST requests with request body over TLS and preserves pool-level default headers","urllib3.connectionpool.HTTPSConnectionPool enforces TLS certificate validation and raises SSLError on untrusted certificates","urllib3.connectionpool.HTTPSConnectionPool.is_same_host matches https URLs for the configured host and port while rejecting mismatched schemes, ports, and hosts"],"goal":"verify urllib3.connectionpool.HTTPSConnectionPool in pkg:pypi/urllib3@2.7.0","kind":"HOW","packages":["pkg:pypi/urllib3@2.7.0"],"schemaVersion":1,"symbols":["urllib3.connectionpool.HTTPSConnectionPool"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/urllib3@2.7.0"],"schemaVersion":1,"subject":"pkg:pypi/urllib3@2.7.0","symbols":["urllib3.connectionpool.HTTPSConnectionPool"],"verifierAdapter":"python@1"}

urllib3==2.7.0

"""Demonstration of HTTPSConnectionPool in urllib3."""

from __future__ import annotations

from typing import Mapping, Optional
from urllib3.connectionpool import HTTPSConnectionPool
from urllib3.response import BaseHTTPResponse


def create_https_pool(
    host: str,
    port: Optional[int] = 443,
    ca_certs: Optional[str] = None,
    cert_reqs: Optional[int | str] = None,
    assert_hostname: Optional[str | bool] = None,
    headers: Optional[Mapping[str, str]] = None,
    maxsize: int = 5,
) -> HTTPSConnectionPool:
    """Create an HTTPSConnectionPool for a specific host and port with TLS configuration."""
    return HTTPSConnectionPool(
        host=host,
        port=port,
        ca_certs=ca_certs,
        cert_reqs=cert_reqs,
        assert_hostname=assert_hostname,
        headers=headers,
        maxsize=maxsize,
    )


def send_https_get_request(
    pool: HTTPSConnectionPool,
    path: str,
    headers: Optional[Mapping[str, str]] = None,
) -> BaseHTTPResponse:
    """Send an HTTP GET request over TLS using the HTTPSConnectionPool."""
    return pool.request("GET", path, headers=headers)


def send_https_post_request(
    pool: HTTPSConnectionPool,
    path: str,
    body: bytes,
    headers: Optional[Mapping[str, str]] = None,
) -> BaseHTTPResponse:
    """Send an HTTP POST request over TLS with payload data using the HTTPSConnectionPool."""
    return pool.request("POST", path, body=body, headers=headers)


def check_https_same_host(pool: HTTPSConnectionPool, url: str) -> bool:
    """Check whether a target URL matches the HTTPSConnectionPool host, port, and scheme."""
    return pool.is_same_host(url)

{
  "schemaVersion": 1,
  "goal": "verify urllib3.connectionpool.HTTPSConnectionPool in pkg:pypi/urllib3@2.7.0",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/urllib3@2.7.0"
  ],
  "symbols": [
    "urllib3.connectionpool.HTTPSConnectionPool"
  ]
}

"""Contract verification test for urllib3.connectionpool.HTTPSConnectionPool."""

from __future__ import annotations

import http.server
import os
import ssl
import subprocess
import sys
import tempfile
import threading

# Ensure root directory is in sys.path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import urllib3.exceptions
from urllib3.connectionpool import HTTPSConnectionPool
from sample import (
    check_https_same_host,
    create_https_pool,
    send_https_get_request,
    send_https_post_request,
)


def _generate_test_certificates(tmpdir: str) -> tuple[str, str]:
    """Generate temporary self-signed certificate and private key using OpenSSL."""
    cert_path = os.path.join(tmpdir, "cert.pem")
    key_path = os.path.join(tmpdir, "key.pem")
    subprocess.run(
        [
            "openssl",
            "req",
            "-x509",
            "-newkey",
            "rsa:2048",
            "-keyout",
            key_path,
            "-out",
            cert_path,
            "-days",
            "1",
            "-nodes",
            "-subj",
            "/CN=localhost",
            "-addext",
            "subjectAltName=DNS:localhost,IP:127.0.0.1",
        ],
        check=True,
        capture_output=True,
    )
    return cert_path, key_path


class MockHTTPSHandler(http.server.BaseHTTPRequestHandler):
    def do_GET(self) -> None:
        if self.path == "/hello":
            self.send_response(200)
            self.send_header("Content-Type", "text/plain")
            self.send_header("X-TLS-Header", "tls-established")
            self.end_headers()
            self.wfile.write(b"Hello HTTPSConnectionPool!")
        else:
            self.send_response(404)
            self.end_headers()

    def do_POST(self) -> None:
        if self.path == "/echo":
            content_len = int(self.headers.get("Content-Length", 0))
            payload = self.rfile.read(content_len)
            default_hdr = self.headers.get("X-Pool-Default", "")
            self.send_response(201)
            self.send_header("Content-Type", "text/plain")
            self.send_header("X-Echo-Default", default_hdr)
            self.end_headers()
            self.wfile.write(b"Echo: " + payload)
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, format: str, *args: object) -> None:
        # Suppress server access logs during test run
        pass


def test_get_requests(port: int, cert_path: str) -> None:
    """Assertion 1: urllib3.connectionpool.HTTPSConnectionPool sends HTTP GET requests over TLS and decodes response status, body, and headers."""
    pool = HTTPSConnectionPool(
        "127.0.0.1",
        port=port,
        ca_certs=cert_path,
        assert_hostname="localhost",
    )
    try:
        assert pool.scheme == "https"
        resp = pool.request("GET", "/hello")
        assert resp.status == 200, f"Expected status 200, got {resp.status}"
        assert resp.data == b"Hello HTTPSConnectionPool!", f"Unexpected response body: {resp.data!r}"
        assert resp.headers.get("X-TLS-Header") == "tls-established", (
            f"Unexpected header: {resp.headers.get('X-TLS-Header')}"
        )

        # Verify helper function
        sample_resp = send_https_get_request(pool, "/hello")
        assert sample_resp.status == 200
        assert sample_resp.data == b"Hello HTTPSConnectionPool!"
    finally:
        pool.close()


def test_post_requests_and_pool_headers(port: int, cert_path: str) -> None:
    """Assertion 2: urllib3.connectionpool.HTTPSConnectionPool sends HTTP POST requests with request body over TLS and preserves pool-level default headers."""
    pool = create_https_pool(
        "127.0.0.1",
        port=port,
        ca_certs=cert_path,
        assert_hostname="localhost",
        headers={"X-Pool-Default": "pool-tls-token"},
    )
    try:
        payload = b"secure payload data"
        resp = pool.request("POST", "/echo", body=payload)
        assert resp.status == 201, f"Expected status 201, got {resp.status}"
        assert resp.data == b"Echo: secure payload data", f"Unexpected response body: {resp.data!r}"
        assert resp.headers.get("X-Echo-Default") == "pool-tls-token", (
            f"Expected pool default header, got {resp.headers.get('X-Echo-Default')}"
        )

        # Verify helper function
        sample_resp = send_https_post_request(pool, "/echo", body=payload)
        assert sample_resp.status == 201
        assert sample_resp.data == b"Echo: secure payload data"

        # Verify context manager cleanup
        pool_ref = None
        with HTTPSConnectionPool("127.0.0.1", port=port, ca_certs=cert_path, assert_hostname="localhost") as ctx_pool:
            pool_ref = ctx_pool
            assert ctx_pool.pool is not None, "Pool queue should be active inside context manager"
            ctx_resp = ctx_pool.request("GET", "/hello")
            assert ctx_resp.status == 200
        assert pool_ref.pool is None, "Pool queue should be closed after exiting context manager"
    finally:
        pool.close()


def test_tls_certificate_validation_failure(port: int) -> None:
    """Assertion 3: urllib3.connectionpool.HTTPSConnectionPool enforces TLS certificate validation and raises SSLError on untrusted certificates."""
    pool_untrusted = HTTPSConnectionPool(
        "127.0.0.1",
        port=port,
        cert_reqs="CERT_REQUIRED",
    )
    raised = False
    try:
        pool_untrusted.request("GET", "/hello", retries=False)
    except urllib3.exceptions.SSLError:
        raised = True
    except urllib3.exceptions.MaxRetryError as e:
        if isinstance(e.reason, urllib3.exceptions.SSLError):
            raised = True
        else:
            raise
    finally:
        pool_untrusted.close()

    assert raised, "Expected SSLError when connecting without trusted CA certificates"


def test_is_same_host(port: int) -> None:
    """Assertion 4: urllib3.connectionpool.HTTPSConnectionPool.is_same_host matches https URLs for the configured host and port while rejecting mismatched schemes, ports, and hosts."""
    pool = HTTPSConnectionPool("127.0.0.1", port=port)
    try:
        assert pool.scheme == "https"
        assert pool.is_same_host(f"https://127.0.0.1:{port}/hello") is True
        assert pool.is_same_host(f"http://127.0.0.1:{port}/hello") is False
        assert pool.is_same_host(f"https://127.0.0.1:{port + 1}/hello") is False
        assert pool.is_same_host(f"https://example.com:{port}/hello") is False

        # Verify helper function
        assert check_https_same_host(pool, f"https://127.0.0.1:{port}/hello") is True
        assert check_https_same_host(pool, f"http://127.0.0.1:{port}/hello") is False
    finally:
        pool.close()


def main() -> None:
    with tempfile.TemporaryDirectory() as tmpdir:
        cert_path, key_path = _generate_test_certificates(tmpdir)

        server = http.server.HTTPServer(("127.0.0.1", 0), MockHTTPSHandler)
        server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        server_ctx.load_cert_chain(cert_path, key_path)
        server.socket = server_ctx.wrap_socket(server.socket, server_side=True)
        port = server.server_address[1]

        server_thread = threading.Thread(target=server.serve_forever, daemon=True)
        server_thread.start()

        try:
            test_get_requests(port, cert_path)
            test_post_requests_and_pool_headers(port, cert_path)
            test_tls_certificate_validation_failure(port)
            test_is_same_host(port)
            print("All contract assertions passed successfully.")
        finally:
            server.shutdown()
            server.server_close()


if __name__ == "__main__":
    main()