test_fbp_err.py

import datetime
import io

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, asymmetric
from cryptography.hazmat.primitives.serialization import (Encoding,
                                                          PrivateFormat,
                                                          NoEncryption)
from cryptography import x509
from twisted.web import client
from twisted.internet import reactor
from twisted.web.server import Site
from twisted.web.static import Data
from twisted.web.resource import Resource
from twisted.internet.ssl import (Certificate, KeyPair, PrivateCertificate,
                                  trustRootFromCertificates)

from twisted.internet.interfaces import IHostnameResolver, IHostResolution
from twisted.internet.address import IPv4Address

rootKey, intermediateKey, serverKey = tuple(
    asymmetric.rsa.generate_private_key(public_exponent=65537,
                                        key_size=2048,
                                        backend=default_backend())
    for i in range(3)
)

def createCert(issuer, subject, privateKey, canSign, signingKey):
    issuer = x509.Name([
        x509.NameAttribute(x509.NameOID.COMMON_NAME, issuer)])

    subject = x509.Name([
        x509.NameAttribute(x509.NameOID.COMMON_NAME, subject)])

    builder = x509.CertificateBuilder().subject_name(
        subject
    ).issuer_name(
        issuer
    ).public_key(
        privateKey.public_key()
    ).serial_number(
        x509.random_serial_number()
    ).not_valid_before(
        datetime.datetime.utcnow()
    ).not_valid_after(
        datetime.datetime.utcnow() + datetime.timedelta(days=1)
    ).add_extension(
        x509.SubjectAlternativeName([x509.DNSName(u"localhost")]),
        critical=False
    )

    if canSign:
        builder = builder.add_extension(
            x509.BasicConstraints(True, None),
            critical=True
        )

    return builder.sign(signingKey, hashes.SHA256(), default_backend())

rootCert = createCert(u"root", u"root", rootKey, True, rootKey)

intermediateCert = createCert(
    u"root",
    u"intermediate",
    intermediateKey,
    True,
    rootKey
)

serverCert = createCert(
    u"intermediate",
    u"server",
    serverKey,
    False,
    intermediateKey
)

serverPrivate = serverKey.private_bytes(
    Encoding.DER,
    PrivateFormat.TraditionalOpenSSL,
    NoEncryption()
)

trustRoot = trustRootFromCertificates(
    [Certificate.loadPEM(rootCert.public_bytes(Encoding.PEM)),
     Certificate.loadPEM(intermediateCert.public_bytes(Encoding.PEM))]
)

privCert = PrivateCertificate.fromCertificateAndKeyPair(
    Certificate.loadPEM(serverCert.public_bytes(Encoding.PEM)),
    KeyPair.load(serverPrivate)
)

root = Resource()


class TestResource(Resource):

    def render_POST(self, request):
        print("here we go")
        return b'Hello, World'


test = TestResource()
root.putChild(b'', TestResource())

port = reactor.listenSSL(
    8080,
    Site(root),
    privCert.options(),
    backlog=128,
    interface='127.0.0.1'
)


def err(failure):
    print(failure.getErrorMessage())
    d = port.stopListening()
    reactor.stop()


def cleanup(response):
    print(response)
    d = port.stopListening()
    d.addCallback(lambda ignored: passthrough)
    reactor.stop()


def headers(response):
    d = client.readBody(response)
    d.addCallbacks(cleanup, err)


host = b'https://localhost:8080/'
# Uncommenting this will ensure the POST succeeds
#cf = client.BrowserLikePolicyForHTTPS(trustRoot=trustRoot)
#agent = client.Agent(reactor, #contextFactory=cf)
agent = client.Agent(reactor)

d = agent.request(
    b'POST',
    host,
    bodyProducer=client.FileBodyProducer(io.BytesIO(b'testdata'))
)
d.addCallbacks(headers, err)
reactor.run()

Try adding the following:

diff --git a/test_fbp_err.py b/test_fbp_err.py
index 8be2249..dc39849 100644
--- a/test_fbp_err.py
+++ b/test_fbp_err.py
@@ -18,6 +18,9 @@ from twisted.internet.ssl import (Certificate, KeyPair, PrivateCertificate,
 from twisted.internet.interfaces import IHostnameResolver, IHostResolution
 from twisted.internet.address import IPv4Address
 
+import sys
+from twisted.logger import globalLogPublisher, textFileLogObserver
+
 rootKey, intermediateKey, serverKey = tuple(
     asymmetric.rsa.generate_private_key(public_exponent=65537,
                                         key_size=2048,
@@ -137,6 +140,8 @@ host = b'https://localhost:8080/'
 #agent = client.Agent(reactor, #contextFactory=cf)
 agent = client.Agent(reactor)
 
+globalLogPublisher.addObserver(textFileLogObserver(sys.stdout))
+
 d = agent.request(
     b'POST',
     host,