urllib3 2.7.0: util.ssl_wrap_socket

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.util.ssl_wrap_socket 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.util.ssl_wrap_socket

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:d43b2235ab184a7ad423905134f074be1d62e300c3afe9b0b81f6ffd0029f035","contract":["urllib3.util.ssl_wrap_socket is exported as a callable function from urllib3.util","ssl_wrap_socket wraps connected socket with explicit ssl_context and server_hostname","ssl_wrap_socket performs TLS handshake allowing secure bidirectional communication","ssl_wrap_socket loads verification locations from ca_cert_data when cert_reqs is specified","ssl_wrap_socket returns an SSLTransport instance when tls_in_tls is True","ssl_wrap_socket raises urllib3.exceptions.SSLError when invalid ca_certs path is provided","ssl_wrap_socket raises urllib3.exceptions.SSLError when encrypted private key is provided without key_password"],"goal":"verify urllib3.util.ssl_wrap_socket in pkg:pypi/urllib3@2.7.0","kind":"HOW","packages":["pkg:pypi/urllib3@2.7.0"],"schemaVersion":1,"symbols":["urllib3.util.ssl_wrap_socket"]},"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.util.ssl_wrap_socket"],"verifierAdapter":"python@1"}

urllib3==2.7.0

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

"""SSL socket utilities package."""

"""Module demonstrating socket wrapping with urllib3.util.ssl_wrap_socket."""

import socket
import ssl
from typing import Optional, Union

import urllib3.exceptions
import urllib3.util
from urllib3.util.ssltransport import SSLTransport


def wrap_client_socket(
    sock: socket.socket,
    server_hostname: Optional[str] = None,
    ssl_context: Optional[ssl.SSLContext] = None,
    ca_cert_data: Optional[Union[str, bytes]] = None,
    cert_reqs: Optional[int] = None,
    tls_in_tls: bool = False,
) -> Union[ssl.SSLSocket, SSLTransport]:
    """Wrap a connected TCP socket using urllib3.util.ssl_wrap_socket.

    Args:
        sock: The raw socket to wrap.
        server_hostname: Expected SNI hostname for TLS verification.
        ssl_context: Optional pre-configured SSLContext.
        ca_cert_data: Optional CA certificates in PEM format.
        cert_reqs: Optional certificate verification requirements (e.g. ssl.CERT_REQUIRED).
        tls_in_tls: Whether to wrap using SSLTransport.

    Returns:
        The wrapped SSLSocket or SSLTransport.

    Raises:
        urllib3.exceptions.SSLError: If certificate verification or TLS wrapping fails.
    """
    return urllib3.util.ssl_wrap_socket(
        sock=sock,
        server_hostname=server_hostname,
        ssl_context=ssl_context,
        ca_cert_data=ca_cert_data,
        cert_reqs=cert_reqs,
        tls_in_tls=tls_in_tls,
    )

"""Test package."""

"""Contract verification tests for urllib3.util.ssl_wrap_socket."""

import os
import socket
import ssl
import subprocess
import sys
import tempfile
import threading

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

import urllib3.exceptions
import urllib3.util
import urllib3.util.ssltransport
from src.tls_wrapper import wrap_client_socket


def _generate_test_certificates(tmpdir: str):
    """Generate self-signed test certificate and keys using OpenSSL."""
    cert_path = os.path.join(tmpdir, "cert.pem")
    key_path = os.path.join(tmpdir, "key.pem")
    enc_key_path = os.path.join(tmpdir, "enc_key.pem")

    # Generate self-signed cert with subjectAltName for localhost
    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,
    )

    with open(cert_path, "r", encoding="utf-8") as f:
        cert_pem = f.read()

    # Generate password-protected encrypted key
    subprocess.run(
        [
            "openssl",
            "rsa",
            "-in",
            key_path,
            "-des3",
            "-out",
            enc_key_path,
            "-passout",
            "pass:testpassword123",
        ],
        check=True,
        capture_output=True,
    )

    return cert_path, key_path, enc_key_path, cert_pem


def _start_echo_server(server_ctx: ssl.SSLContext):
    """Start a local TLS echo server thread."""
    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.bind(("127.0.0.1", 0))
    srv.listen(1)
    port = srv.getsockname()[1]

    def handle():
        conn, _ = srv.accept()
        sconn = server_ctx.wrap_socket(conn, server_side=True)
        try:
            msg = sconn.recv(1024)
            sconn.sendall(b"ACK:" + msg)
        finally:
            sconn.close()
            srv.close()

    thread = threading.Thread(target=handle)
    thread.daemon = True
    thread.start()
    return port, thread


def test_exported():
    assert callable(urllib3.util.ssl_wrap_socket), "ssl_wrap_socket should be callable"


def test_wrap_with_explicit_context(cert_path, key_path):
    server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    server_ctx.load_cert_chain(cert_path, key_path)
    port, server_thread = _start_echo_server(server_ctx)

    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect(("127.0.0.1", port))

    client_ctx = urllib3.util.create_urllib3_context(cert_reqs=ssl.CERT_REQUIRED)
    client_ctx.load_verify_locations(cert_path)

    wrapped = urllib3.util.ssl_wrap_socket(
        sock=client,
        ssl_context=client_ctx,
        server_hostname="localhost",
    )
    assert isinstance(wrapped, ssl.SSLSocket), f"Expected SSLSocket, got {type(wrapped)}"

    wrapped.sendall(b"test_payload_1")
    response = wrapped.recv(1024)
    assert response == b"ACK:test_payload_1", f"Unexpected response: {response}"

    wrapped.close()
    server_thread.join(timeout=5.0)


def test_wrap_with_ca_cert_data(cert_path, key_path, cert_pem):
    server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    server_ctx.load_cert_chain(cert_path, key_path)
    port, server_thread = _start_echo_server(server_ctx)

    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect(("127.0.0.1", port))

    wrapped = wrap_client_socket(
        sock=client,
        cert_reqs=ssl.CERT_REQUIRED,
        ca_cert_data=cert_pem,
        server_hostname="localhost",
    )
    assert isinstance(wrapped, ssl.SSLSocket), f"Expected SSLSocket, got {type(wrapped)}"

    wrapped.sendall(b"test_payload_2")
    response = wrapped.recv(1024)
    assert response == b"ACK:test_payload_2", f"Unexpected response: {response}"

    wrapped.close()
    server_thread.join(timeout=5.0)


def test_invalid_ca_certs_raises():
    dummy = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        urllib3.util.ssl_wrap_socket(sock=dummy, ca_certs="nonexistent_ca_cert_file.pem")
        assert False, "Expected SSLError for invalid ca_certs path"
    except urllib3.exceptions.SSLError:
        pass
    finally:
        dummy.close()


def test_encrypted_key_without_password_raises(cert_path, enc_key_path):
    dummy = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        urllib3.util.ssl_wrap_socket(
            sock=dummy,
            certfile=cert_path,
            keyfile=enc_key_path,
            key_password=None,
        )
        assert False, "Expected SSLError when key is encrypted without password"
    except urllib3.exceptions.SSLError as e:
        assert "encrypted" in str(e).lower(), f"Expected 'encrypted' in error message, got: {e}"
    finally:
        dummy.close()


def test_tls_in_tls(cert_path, key_path):
    server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    server_ctx.load_cert_chain(cert_path, key_path)
    port, server_thread = _start_echo_server(server_ctx)

    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect(("127.0.0.1", port))

    client_ctx = urllib3.util.create_urllib3_context(cert_reqs=ssl.CERT_REQUIRED)
    client_ctx.load_verify_locations(cert_path)

    transport = wrap_client_socket(
        sock=client,
        ssl_context=client_ctx,
        server_hostname="localhost",
        tls_in_tls=True,
    )
    assert isinstance(
        transport, urllib3.util.ssltransport.SSLTransport
    ), f"Expected SSLTransport, got {type(transport)}"

    transport.sendall(b"test_payload_tls_in_tls")
    response = transport.recv(1024)
    assert response == b"ACK:test_payload_tls_in_tls", f"Unexpected response: {response}"

    transport.close()
    server_thread.join(timeout=5.0)


def main():
    with tempfile.TemporaryDirectory() as tmpdir:
        cert_path, key_path, enc_key_path, cert_pem = _generate_test_certificates(tmpdir)

        test_exported()
        test_wrap_with_explicit_context(cert_path, key_path)
        test_wrap_with_ca_cert_data(cert_path, key_path, cert_pem)
        test_invalid_ca_certs_raises()
        test_encrypted_key_without_password_raises(cert_path, enc_key_path)
        test_tls_in_tls(cert_path, key_path)

    print("All contract assertions passed for urllib3.util.ssl_wrap_socket.")
    return 0


if __name__ == "__main__":
    sys.exit(main())