Twisted Python How To Create a twisted.web.client.

Question

I am trying to create a t.w.c.BrowserLikePolicyForHTTPS to use as the ContextFactory for a t.w.c.Agent. I am using an internal CA for all the servers I want the Agent to communicate with, so I'd like to be able to tell to load the CA cert (PEM format) and use it as the trustRoot argument to BrowserLikePolicyForHTTPS. I have read the docs and looked at the source, but I have no idea what I am supposed to supply as arguments. I tried providing a PyOPenSSL x509 object, but I get an error:

exceptions.TypeError: ('Could not adapt', <OpenSSL.crypto.X509 object at 0x280b290>, <InterfaceClass twisted.internet._sslverify.IOpenSSLTrustRoot>)

I can see in the code in t.i._sslverify that OpenSSLCertificateAuthorities somehow gets adapted to IOpenSSLTrustRoot, but it is not really clear to me how this happens.

I know the stock agent doesn't do any cert-checking. I am working with a fork of treq and am experimenting with adding an option to provide a custom Agent.

Any help with the trustRoot argument would be appreciated. If I am going about this the hard way, please let me know that, too.

Answer 1

IOpenSSLTrustRoot is a bit of a nonsense API.

It is not a public interface itself - so you can't implement your own. If it were, it has no public methods so it's not clear how you might customize its behavior anyway.

Considering the security-sensitive nature of this interface, I bet Twisted will go to great lengths to make sure whatever code you write continues to work - even though the interface is private and there are only private methods on the interface.

If you write some code that depends on this interface always working like it works now then you can probably at least post to the Twisted mailing list and say so and the folks there will probably try not to break your application.

That said, as I pointed out above, every part of this is private. Twisted's stated policy is there are no guarantees of backwards compatibility here. So proceed at your own risk.

In any case, you can write something like this:

from zope.interface import implementer

from characteristic import attributes

from twisted.internet._sslverify import IOpenSSLTrustRoot

@implementer(IOpenSSLTrustRoot)
@attributes(["root_certificate_path"])
class MyCATrustRoot(object):
    def _addCACertsToContext(self, context):
        context.load_verify_locations(self.root_certificate_path)

Then use a MyCATrustRoot instance as the trustRoot argument to BrowserLikePolicyForHTTPS. Make sure you use at least Twisted 14.0.2 because earlier versions actually ignored the trustRoot you passed to BrowserLikePolicyForHTTPS.

For example, if your trusted "CA" certificate is at /foo/ca.pem:

from twisted.web.client import BrowserLikePolicyForHTTPS, Agent
from twisted.internet import reactor

agent = Agent(reactor, BrowserLikePolicyForHTTPS(
    MyCATrustRoot(root_certificate_path="/foo/ca.pem")))

Answer 2

Your question here highlights a terrible oversight in the documentation; both in the API documentation, and in the narrative documentation for. If Jean-Paul can't figure out the "right way" to do this, then there is clearly no hope for a regular user. I have filed a bug to correct this oversight.

In the meanwhile, please avoid Jean-Paul's solution. While it is functional, it involves techniques which will almost certainly break without warning in future releases (as he clearly notes). Luckily there are supported ways to do this. If you have a single alternate trust root, Certificate is usable as a value to the trustRoot parameter. You can use it like so (I have tested the following example with Twisted 14.0.2):

from __future__ import print_function
from twisted.web.client import Agent, BrowserLikePolicyForHTTPS
from twisted.internet.task import react
from twisted.internet.ssl import Certificate
from twisted.internet.protocol import Protocol
from twisted.python.filepath import FilePath
from twisted.internet.defer import inlineCallbacks, Deferred

@inlineCallbacks
def main(reactor):
    customPolicy = BrowserLikePolicyForHTTPS(
        Certificate.loadPEM(FilePath("your-trust-root.pem").getContent())
    )
    agent = Agent(reactor, customPolicy)
    response = yield agent.request(
        "GET", "https://your-web-site.example.com/"
    )
    done = Deferred()
    class CaptureString(Protocol):
        def dataReceived(self, data):
            print("Received:", data)
        def connectionLost(self, reason):
            done.callback(None)
    response.deliverBody(CaptureString())
    yield done

react(main)