88 lines
3.8 KiB
Python
88 lines
3.8 KiB
Python
"""Isolated test routing; never imported by the installed node package."""
|
|
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timedelta, timezone
|
|
import ipaddress
|
|
from pathlib import Path
|
|
import socket
|
|
import ssl
|
|
import tempfile
|
|
from unittest.mock import patch
|
|
|
|
from cryptography import x509
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from cryptography.x509.oid import NameOID
|
|
from urllib3.util import connection
|
|
|
|
|
|
class TLSFixture:
|
|
"""Fresh test-only certificate; production verification stays enabled."""
|
|
|
|
def __init__(self, directory=None):
|
|
self.temp = tempfile.TemporaryDirectory() if directory is None else None
|
|
root = Path(self.temp.name if self.temp else directory)
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
self.cert = root / "test-cert.pem"
|
|
key_path = root / "test-key.pem"
|
|
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "NotifyMe isolated test")])
|
|
now = datetime.now(timezone.utc)
|
|
certificate = (x509.CertificateBuilder().subject_name(name).issuer_name(name)
|
|
.public_key(key.public_key()).serial_number(x509.random_serial_number())
|
|
.not_valid_before(now - timedelta(minutes=5)).not_valid_after(now + timedelta(days=1))
|
|
.add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
|
|
.add_extension(x509.SubjectAlternativeName([
|
|
x509.DNSName("notify.bro3.net"), x509.DNSName("callback.example"),
|
|
x509.DNSName("localhost"), x509.IPAddress(ipaddress.ip_address("127.0.0.1")),
|
|
]), critical=False).sign(key, hashes.SHA256()))
|
|
self.cert.write_bytes(certificate.public_bytes(serialization.Encoding.PEM))
|
|
key_path.write_bytes(key.private_bytes(serialization.Encoding.PEM,
|
|
serialization.PrivateFormat.PKCS8, serialization.NoEncryption()))
|
|
self.server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
self.server_context.load_cert_chain(str(self.cert), str(key_path))
|
|
self.client_context = ssl.create_default_context(cafile=str(self.cert))
|
|
|
|
def close(self):
|
|
if self.temp:
|
|
self.temp.cleanup()
|
|
|
|
def trust(self):
|
|
return trust_test_certificate(self.cert)
|
|
|
|
|
|
@contextmanager
|
|
def trust_test_certificate(cert):
|
|
with patch("requests.certs.where", return_value=str(cert)), patch(
|
|
"requests.adapters.DEFAULT_CA_BUNDLE_PATH", str(cert)
|
|
):
|
|
yield
|
|
|
|
|
|
@contextmanager
|
|
def routed_callback(port, cert=None):
|
|
"""Use synthetic DNS and a loopback socket with production validation intact."""
|
|
original_resolve = socket.getaddrinfo
|
|
original_connect = connection.create_connection
|
|
public_address = "93.184.216.34"
|
|
|
|
def resolve(host, query_port, *args, **kwargs):
|
|
if host == "notify.bro3.net":
|
|
if int(query_port) != port:
|
|
raise OSError("Unexpected test callback port")
|
|
return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (public_address, port))]
|
|
return original_resolve(host, query_port, *args, **kwargs)
|
|
|
|
def connect(address, *args, **kwargs):
|
|
if address[0] == public_address:
|
|
if int(address[1]) != port:
|
|
raise OSError("Unexpected test callback port")
|
|
address = ("127.0.0.1", port)
|
|
return original_connect(address, *args, **kwargs)
|
|
|
|
from contextlib import nullcontext
|
|
with trust_test_certificate(cert) if cert else nullcontext(), patch("socket.getaddrinfo", side_effect=resolve), patch(
|
|
"urllib3.util.connection.create_connection", side_effect=connect
|
|
):
|
|
yield
|