CVE-2019-12387: Improper Neutralization of Special Elements in Output Used by a Downstream Component ('Injection') (CWE-74)
value: In Twisted before 19.2.1, twisted.web did not validate or sanitize URIs or HTTP methods, allowing an attacker to inject invalid characters such as CRLF.
CWE description: The product constructs all or part of a command, data structure, or record using externally-influenced input from an upstream component, but it does not neutralize or incorrectly neutralizes special elements that could modify how it is parsed or interpreted when it is sent to a downstream component.
CVSS v3 base score: 6.1
Published: 2019-06-10T12:29Z
Repository: https://github.com/twisted/twisted
Language: Python
Files changed: src/twisted/web/_newclient.py, src/twisted/web/client.py, src/twisted/web/newsfragments/9647.bugfix, src/twisted/web/test/injectionhelpers.py, src/twisted/web/test/test_agent.py, src/twisted/web/test/test_webclient.py
Commit date: 2019-06-05 00:03:37 -0700
Commit message: Prevent CRLF injections described in CVE-2019-12387

Author: markrwilliams

Reviewers: glyph

Fixes: ticket:9647

Twisted's HTTP client APIs were vulnerable to maliciously constructed
HTTP methods, hosts, and/or paths, URI components such as paths and
query parameters.  These vulnerabilities were beyond the header name
and value injection vulnerabilities addressed in:

https://twistedmatrix.com/trac/ticket/9420
https://github.com/twisted/twisted/pull/999/

The following client APIs will raise a ValueError if given a method,
host, or URI that includes newlines or other disallowed characters:

- twisted.web.client.Agent.request
- twisted.web.client.ProxyAgent.request
- twisted.web.client.Request.__init__
- twisted.web.client.Request.writeTo

ProxyAgent is patched separately from Agent because unlike other
agents (e.g. CookieAgent) it is not implemented as an Agent wrapper.

Request.__init__ checks its method and URI so that errors occur closer
to their originating input.  Request.method and Request.uri are both
public APIs, however, so Request.writeTo (via Request._writeHeaders)
also checks the validity of both before writing anything to the wire.

Additionally, the following deprecated client APIs have also been
patched:

- twisted.web.client.HTTPPageGetter.__init__
- twisted.web.client.HTTPPageDownloader.__init__
- twisted.web.client.HTTPClientFactory.__init__
- twisted.web.client.HTTPClientFactory.setURL
- twisted.web.client.HTTPDownloader.__init__
- twisted.web.client.HTTPDownloader.setURL
- twisted.web.client.getPage
- twisted.web.client.downloadPage

These have been patched prior to their removal so that they won't be
vulnerable in the last Twisted release that includes them.  They
represent a best effort, because testing every combination of these
public APIs would require more code than deprecated APIs warrant.

In all cases URI components, including hostnames, are restricted to
the characters allowed in path components.  This mirrors the CPython
patch (for bpo-30458) that addresses equivalent vulnerabilities:

https://github.com/python/cpython/commit/bb8071a4cae5ab3fe321481dd3d73662ffb26052

HTTP methods, however, are checked against the set of characters
described in RFC-7230.
Diff: diff --git a/src/twisted/web/_newclient.py b/src/twisted/web/_newclient.py
index 370f47d57..74a8a6c2d 100644
--- a/src/twisted/web/_newclient.py
+++ b/src/twisted/web/_newclient.py
@@ -27,10 +27,12 @@ Various other classes in this module support this usage:
 """
 
 from __future__ import division, absolute_import
 __metaclass__ = type
 
+import re
+
 from zope.interface import implementer
 
 from twisted.python.compat import networkString
 from twisted.python.components import proxyForInterface
 from twisted.python.reflect import fullyQualifiedName
@@ -577,10 +579,78 @@ class HTTPClientParser(HTTPParser):
             self._responseDeferred.errback(Failure(exceptionClass([reason])))
             del self._responseDeferred
 
 
 
+_VALID_METHOD = re.compile(
+    br"\A[%s]+\Z" % (
+        bytes().join(
+            (
+                b"!", b"#", b"$", b"%", b"&", b"'", b"*",
+                b"+", b"-", b".", b"^", b"_", b"`", b"|", b"~",
+                b"\x30-\x39",
+                b"\x41-\x5a",
+                b"\x61-\x7A",
+            ),
+        ),
+    ),
+)
+
+
+
+def _ensureValidMethod(method):
+    """
+    An HTTP method is an HTTP token, which consists of any visible
+    ASCII character that is not a delimiter (i.e. one of
+    C{"(),/:;<=>?@[\\]{}}.)
+
+    @param method: the method to check
+    @type method: L{bytes}
+
+    @return: the method if it is valid
+    @rtype: L{bytes}
+
+    @raise ValueError: if the method is not valid
+
+    @see: U{https://tools.ietf.org/html/rfc7230#section-3.1.1},
+        U{https://tools.ietf.org/html/rfc7230#section-3.2.6},
+        U{https://tools.ietf.org/html/rfc5234#appendix-B.1}
+    """
+    if _VALID_METHOD.match(method):
+        return method
+    raise ValueError("Invalid method {!r}".format(method))
+
+
+
+_VALID_URI = re.compile(br'\A[\x21-\x7e]+\Z')
+
+
+
+def _ensureValidURI(uri):
+    """
+    A valid URI cannot contain control characters (i.e., characters
+    between 0-32, inclusive and 127) or non-ASCII characters (i.e.,
+    characters with values between 128-255, inclusive).
+
+    @param uri: the URI to check
+    @type uri: L{bytes}
+
+    @return: the URI if it is valid
+    @rtype: L{bytes}
+
+    @raise ValueError: if the URI is not valid
+
+    @see: U{https://tools.ietf.org/html/rfc3986#section-3.3},
+        U{https://tools.ietf.org/html/rfc3986#appendix-A},
+        U{https://tools.ietf.org/html/rfc5234#appendix-B.1}
+    """
+    if _VALID_URI.match(uri):
+        return uri
+    raise ValueError("Invalid URI {!r}".format(uri))
+
+
+
 @implementer(IClientRequest)
 class Request:
     """
     A L{Request} instance describes an HTTP request to be sent to an HTTP
     server.
@@ -616,12 +686,12 @@ class Request:
 
         @param persistent: Set to C{True} when you use HTTP persistent
             connection, defaults to C{False}.
         @type persistent: L{bool}
         """
-        self.method = method
-        self.uri = uri
+        self.method = _ensureValidMethod(method)
+        self.uri = _ensureValidURI(uri)
         self.headers = headers
         self.bodyProducer = bodyProducer
         self.persistent = persistent
         self._parsedURI = None
 
@@ -662,12 +732,19 @@ class Request:
 
         # In the future, having the protocol version be a parameter to this
         # method would probably be good.  It would be nice if this method
         # weren't limited to issuing HTTP/1.1 requests.
         requestLines = []
-        requestLines.append(b' '.join([self.method, self.uri,
-            b'HTTP/1.1\r\n']))
+        requestLines.append(
+            b' '.join(
+                [
+                    _ensureValidMethod(self.method),
+                    _ensureValidURI(self.uri),
+                    b'HTTP/1.1\r\n',
+                ]
+            ),
+        )
         if not self.persistent:
             requestLines.append(b'Connection: close\r\n')
         if TEorCL is not None:
             requestLines.append(TEorCL)
         for name, values in self.headers.getAllRawHeaders():
diff --git a/src/twisted/web/client.py b/src/twisted/web/client.py
index e1a92fef4..340906f62 100644
--- a/src/twisted/web/client.py
+++ b/src/twisted/web/client.py
@@ -45,10 +45,13 @@ from twisted.python.components import proxyForInterface
 from twisted.web import error
 from twisted.web.iweb import UNKNOWN_LENGTH, IAgent, IBodyProducer, IResponse
 from twisted.web.http_headers import Headers
 from twisted.logger import Logger
 
+from twisted.web._newclient import _ensureValidURI, _ensureValidMethod
+
+
 
 class PartialDownloadError(error.Error):
     """
     Page was only partially downloaded, we got disconnected in middle.
 
@@ -76,15 +79,17 @@ class HTTPPageGetter(http.HTTPClient):
     followRedirect = True
     failed = 0
 
     _completelyDone = True
 
-    _specialHeaders = set((b'host', b'user-agent', b'cookie', b'content-length'))
+    _specialHeaders = set(
+        (b'host', b'user-agent', b'cookie', b'content-length'),
+    )
 
     def connectionMade(self):
-        method = getattr(self.factory, 'method', b'GET')
-        self.sendCommand(method, self.factory.path)
+        method = _ensureValidMethod(getattr(self.factory, 'method', b'GET'))
+        self.sendCommand(method, _ensureValidURI(self.factory.path))
         if self.factory.scheme == b'http' and self.factory.port != 80:
             host = self.factory.host + b':' + intToBytes(self.factory.port)
         elif self.factory.scheme == b'https' and self.factory.port != 443:
             host = self.factory.host + b':' + intToBytes(self.factory.port)
         else:
@@ -360,11 +365,11 @@ class HTTPClientFactory(protocol.ClientFactory):
             self.headers.setdefault(b'Content-Length',
                                     intToBytes(len(postdata)))
             # just in case a broken http/1.1 decides to keep connection alive
             self.headers.setdefault(b"connection", b"close")
         self.postdata = postdata
-        self.method = method
+        self.method = _ensureValidMethod(method)
 
         self.setURL(url)
 
         self.waiting = 1
         self._disconnectedDeferred = defer.Deferred()
@@ -387,10 +392,11 @@ class HTTPClientFactory(protocol.ClientFactory):
 
     def __repr__(self):
         return "<%s: %s>" % (self.__class__.__name__, self.url)
 
     def setURL(self, url):
+        _ensureValidURI(url.strip())
         self.url = url
         uri = URI.fromBytes(url)
         if uri.scheme and uri.host:
             self.scheme = uri.scheme
             self.host = uri.host
@@ -731,11 +737,11 @@ def _makeGetterFactory(url, factoryFactory, contextFactory=None,
     @param contextFactory: Context factory to use when creating a secure
         connection, defaulting to L{None}
 
     @return: The factory created by C{factoryFactory}
     """
-    uri = URI.fromBytes(url)
+    uri = URI.fromBytes(_ensureValidURI(url.strip()))
     factory = factoryFactory(url, *args, **kwargs)
     if uri.scheme == b'https':
         from twisted.internet import ssl
         if contextFactory is None:
             contextFactory = ssl.ClientContextFactory()
@@ -1491,10 +1497,13 @@ class _AgentBase(object):
         the request.
         """
         if not isinstance(method, bytes):
             raise TypeError('method={!r} is {}, but must be bytes'.format(
                     method, type(method)))
+
+        method = _ensureValidMethod(method)
+
         # Create minimal headers, if necessary:
         if headers is None:
             headers = Headers()
         if not headers.hasHeader(b'host'):
             headers = headers.copy()
@@ -1715,10 +1724,11 @@ class Agent(_AgentBase):
 
         I{HTTP} and I{HTTPS} schemes are supported in C{uri}.
 
         @see: L{twisted.web.iweb.IAgent.request}
         """
+        uri = _ensureValidURI(uri.strip())
         parsedURI = URI.fromBytes(uri)
         try:
             endpoint = self._getEndpoint(parsedURI)
         except SchemeNotSupported:
             return defer.fail(Failure())
@@ -1748,10 +1758,12 @@ class ProxyAgent(_AgentBase):
 
     def request(self, method, uri, headers=None, bodyProducer=None):
         """
         Issue a new request via the configured proxy.
         """
+        uri = _ensureValidURI(uri.strip())
+
         # Cache *all* connections under the same key, since we are only
         # connecting to a single destination, the proxy:
         key = ("http-proxy", self._proxyEndpoint)
 
         # To support proxying HTTPS via CONNECT, we will use key
diff --git a/src/twisted/web/newsfragments/9647.bugfix b/src/twisted/web/newsfragments/9647.bugfix
new file mode 100644
index 000000000..b76916cdc
--- /dev/null
+++ b/src/twisted/web/newsfragments/9647.bugfix
@@ -0,0 +1 @@
+All HTTP clients in twisted.web.client now raise a ValueError when called with a method and/or URL that contain invalid characters.  This mitigates CVE-2019-12387.  Thanks to Alex Brasetvik for reporting this vulnerability.
\ No newline at end of file
diff --git a/src/twisted/web/test/injectionhelpers.py b/src/twisted/web/test/injectionhelpers.py
new file mode 100644
index 000000000..ffeb8629c
--- /dev/null
+++ b/src/twisted/web/test/injectionhelpers.py
@@ -0,0 +1,168 @@
+"""
+Helpers for URI and method injection tests.
+
+@see: U{CVE-2019-12387}
+"""
+
+import string
+
+
+UNPRINTABLE_ASCII = (
+    frozenset(range(0, 128)) -
+    frozenset(bytearray(string.printable, 'ascii'))
+)
+
+NONASCII = frozenset(range(128, 256))
+
+
+
+class MethodInjectionTestsMixin(object):
+    """
+    A mixin that runs HTTP method injection tests.  Define
+    L{MethodInjectionTestsMixin.attemptRequestWithMaliciousMethod} in
+    a L{twisted.trial.unittest.SynchronousTestCase} subclass to test
+    how HTTP client code behaves when presented with malicious HTTP
+    methods.
+
+    @see: U{CVE-2019-12387}
+    """
+
+    def attemptRequestWithMaliciousMethod(self, method):
+        """
+        Attempt to send a request with the given method.  This should
+        synchronously raise a L{ValueError} if either is invalid.
+
+        @param method: the method (e.g. C{GET\x00})
+
+        @param uri: the URI
+
+        @type method:
+        """
+        raise NotImplementedError()
+
+
+    def test_methodWithCLRFRejected(self):
+        """
+        Issuing a request with a method that contains a carriage
+        return and line feed fails with a L{ValueError}.
+        """
+        with self.assertRaises(ValueError) as cm:
+            method = b"GET\r\nX-Injected-Header: value"
+            self.attemptRequestWithMaliciousMethod(method)
+        self.assertRegex(str(cm.exception), "^Invalid method")
+
+
+    def test_methodWithUnprintableASCIIRejected(self):
+        """
+        Issuing a request with a method that contains unprintable
+        ASCII characters fails with a L{ValueError}.
+        """
+        for c in UNPRINTABLE_ASCII:
+            method = b"GET%s" % (bytearray([c]),)
+            with self.assertRaises(ValueError) as cm:
+                self.attemptRequestWithMaliciousMethod(method)
+            self.assertRegex(str(cm.exception), "^Invalid method")
+
+
+    def test_methodWithNonASCIIRejected(self):
+        """
+        Issuing a request with a method that contains non-ASCII
+        characters fails with a L{ValueError}.
+        """
+        for c in NONASCII:
+            method = b"GET%s" % (bytearray([c]),)
+            with self.assertRaises(ValueError) as cm:
+                self.attemptRequestWithMaliciousMethod(method)
+            self.assertRegex(str(cm.exception), "^Invalid method")
+
+
+
+class URIInjectionTestsMixin(object):
+    """
+    A mixin that runs HTTP URI injection tests.  Define
+    L{MethodInjectionTestsMixin.attemptRequestWithMaliciousURI} in a
+    L{twisted.trial.unittest.SynchronousTestCase} subclass to test how
+    HTTP client code behaves when presented with malicious HTTP
+    URIs.
+    """
+
+    def attemptRequestWithMaliciousURI(self, method):
+        """
+        Attempt to send a request with the given URI.  This should
+        synchronously raise a L{ValueError} if either is invalid.
+
+        @param uri: the URI.
+
+        @type method:
+        """
+        raise NotImplementedError()
+
+
+    def test_hostWithCRLFRejected(self):
+        """
+        Issuing a request with a URI whose host contains a carriage
+        return and line feed fails with a L{ValueError}.
+        """
+        with self.assertRaises(ValueError) as cm:
+            uri = b"http://twisted\r\n.invalid/path"
+            self.attemptRequestWithMaliciousURI(uri)
+        self.assertRegex(str(cm.exception), "^Invalid URI")
+
+
+    def test_hostWithWithUnprintableASCIIRejected(self):
+        """
+        Issuing a request with a URI whose host contains unprintable
+        ASCII characters fails with a L{ValueError}.
+        """
+        for c in UNPRINTABLE_ASCII:
+            uri = b"http://twisted%s.invalid/OK" % (bytearray([c]),)
+            with self.assertRaises(ValueError) as cm:
+                self.attemptRequestWithMaliciousURI(uri)
+            self.assertRegex(str(cm.exception), "^Invalid URI")
+
+
+    def test_hostWithNonASCIIRejected(self):
+        """
+        Issuing a request with a URI whose host contains non-ASCII
+        characters fails with a L{ValueError}.
+        """
+        for c in NONASCII:
+            uri = b"http://twisted%s.invalid/OK" % (bytearray([c]),)
+            with self.assertRaises(ValueError) as cm:
+                self.attemptRequestWithMaliciousURI(uri)
+            self.assertRegex(str(cm.exception), "^Invalid URI")
+
+
+    def test_pathWithCRLFRejected(self):
+        """
+        Issuing a request with a URI whose path contains a carriage
+        return and line feed fails with a L{ValueError}.
+        """
+        with self.assertRaises(ValueError) as cm:
+            uri = b"http://twisted.invalid/\r\npath"
+            self.attemptRequestWithMaliciousURI(uri)
+        self.assertRegex(str(cm.exception), "^Invalid URI")
+
+
+    def test_pathWithWithUnprintableASCIIRejected(self):
+        """
+        Issuing a request with a URI whose path contains unprintable
+        ASCII characters fails with a L{ValueError}.
+        """
+        for c in UNPRINTABLE_ASCII:
+            uri = b"http://twisted.invalid/OK%s" % (bytearray([c]),)
+            with self.assertRaises(ValueError) as cm:
+                self.attemptRequestWithMaliciousURI(uri)
+            self.assertRegex(str(cm.exception), "^Invalid URI")
+
+
+    def test_pathWithNonASCIIRejected(self):
+        """
+        Issuing a request with a URI whose path contains non-ASCII
+        characters fails with a L{ValueError}.
+        """
+        for c in NONASCII:
+            uri = b"http://twisted.invalid/OK%s" % (bytearray([c]),)
+            with self.assertRaises(ValueError) as cm:
+                self.attemptRequestWithMaliciousURI(uri)
+            self.assertRegex(str(cm.exception), "^Invalid URI")
diff --git a/src/twisted/web/test/test_agent.py b/src/twisted/web/test/test_agent.py
index fc67d1541..040dc9ef3 100644
--- a/src/twisted/web/test/test_agent.py
+++ b/src/twisted/web/test/test_agent.py
@@ -9,11 +9,11 @@ import zlib
 
 from io import BytesIO
 
 from zope.interface.verify import verifyObject
 
-from twisted.trial.unittest import TestCase
+from twisted.trial.unittest import TestCase, SynchronousTestCase
 from twisted.web import client, error, http_headers
 from twisted.web._newclient import RequestNotSent, RequestTransmissionFailed
 from twisted.web._newclient import ResponseNeverReceived, ResponseFailed
 from twisted.web._newclient import PotentialDataLoss
 from twisted.internet import defer, task
@@ -49,10 +49,14 @@ from twisted.web.client import (BrowserLikePolicyForHTTPS,
 from twisted.internet.test.test_endpoints import deterministicResolvingReactor
 from twisted.internet.endpoints import HostnameEndpoint
 from twisted.test.proto_helpers import AccumulatingProtocol
 from twisted.test.iosim import IOPump, FakeTransport
 from twisted.test.test_sslverify import certificatesForAuthorityAndServer
+from twisted.web.test.injectionhelpers import (
+    MethodInjectionTestsMixin,
+    URIInjectionTestsMixin,
+)
 from twisted.web.error import SchemeNotSupported
 from twisted.logger import globalLogPublisher
 
 try:
     from twisted.internet import ssl
@@ -895,10 +899,11 @@ class StubEndpointFactory(object):
 class AgentTests(TestCase, FakeReactorAndConnectMixin, AgentTestsMixin,
                  IntegrationTestingMixin):
     """
     Tests for the new HTTP client API provided by L{Agent}.
     """
+
     def makeAgent(self):
         """
         @return: a new L{twisted.web.client.Agent} instance
         """
         return client.Agent(self.reactor)
@@ -1325,10 +1330,52 @@ class AgentTests(TestCase, FakeReactorAndConnectMixin, AgentTestsMixin,
             self.reactor, StubEndpointFactory(), pool)
         self.assertIs(pool, agent._pool)
 
 
 
+class AgentMethodInjectionTests(
+        FakeReactorAndConnectMixin,
+        MethodInjectionTestsMixin,
+        SynchronousTestCase,
+):
+    """
+    Test L{client.Agent} against HTTP method injections.
+    """
+
+    def attemptRequestWithMaliciousMethod(self, method):
+        """
+        Attempt a request with the provided method.
+
+        @param method: see L{MethodInjectionTestsMixin}
+        """
+        agent = client.Agent(self.createReactor())
+        uri = b"http://twisted.invalid"
+        agent.request(method, uri, client.Headers(), None)
+
+
+
+class AgentURIInjectionTests(
+        FakeReactorAndConnectMixin,
+        URIInjectionTestsMixin,
+        SynchronousTestCase,
+):
+    """
+    Test L{client.Agent} against URI injections.
+    """
+
+    def attemptRequestWithMaliciousURI(self, uri):
+        """
+        Attempt a request with the provided method.
+
+        @param uri: see L{URIInjectionTestsMixin}
+        """
+        agent = client.Agent(self.createReactor())
+        method = b"GET"
+        agent.request(method, uri, client.Headers(), None)
+
+
+
 class AgentHTTPSTests(TestCase, FakeReactorAndConnectMixin,
                       IntegrationTestingMixin):
     """
     Tests for the new HTTP client API that depends on SSL.
     """
@@ -3200,5 +3247,103 @@ class HostnameCachingHTTPSPolicyTests(TestCase):
         policy.creatorForNetloc(hostn.encode("ascii"), 309)
         self.assertNotIn(first, policy._cache)
         self.assertEquals(5, len(policy._cache))
 
         self.assertIn(hostn, policy._cache)
+
+
+
+class RequestMethodInjectionTests(
+        MethodInjectionTestsMixin,
+        SynchronousTestCase,
+):
+    """
+    Test L{client.Request} against HTTP method injections.
+    """
+
+    def attemptRequestWithMaliciousMethod(self, method):
+        """
+        Attempt a request with the provided method.
+
+        @param method: see L{MethodInjectionTestsMixin}
+        """
+        client.Request(
+            method=method,
+            uri=b"http://twisted.invalid",
+            headers=http_headers.Headers(),
+            bodyProducer=None,
+        )
+
+
+
+class RequestWriteToMethodInjectionTests(
+        MethodInjectionTestsMixin,
+        SynchronousTestCase,
+):
+    """
+    Test L{client.Request.writeTo} against HTTP method injections.
+    """
+
+    def attemptRequestWithMaliciousMethod(self, method):
+        """
+        Attempt a request with the provided method.
+
+        @param method: see L{MethodInjectionTestsMixin}
+        """
+        headers = http_headers.Headers({b"Host": [b"twisted.invalid"]})
+        req = client.Request(
+            method=b"GET",
+            uri=b"http://twisted.invalid",
+            headers=headers,
+            bodyProducer=None,
+        )
+        req.method = method
+        req.writeTo(StringTransport())
+
+
+
+class RequestURIInjectionTests(
+        URIInjectionTestsMixin,
+        SynchronousTestCase,
+):
+    """
+    Test L{client.Request} against HTTP URI injections.
+    """
+
+    def attemptRequestWithMaliciousURI(self, uri):
+        """
+        Attempt a request with the provided URI.
+
+        @param method: see L{URIInjectionTestsMixin}
+        """
+        client.Request(
+            method=b"GET",
+            uri=uri,
+            headers=http_headers.Headers(),
+            bodyProducer=None,
+        )
+
+
+
+class RequestWriteToURIInjectionTests(
+        URIInjectionTestsMixin,
+        SynchronousTestCase,
+):
+    """
+    Test L{client.Request.writeTo} against HTTP method injections.
+    """
+
+    def attemptRequestWithMaliciousURI(self, uri):
+        """
+        Attempt a request with the provided method.
+
+        @param method: see L{URIInjectionTestsMixin}
+        """
+        headers = http_headers.Headers({b"Host": [b"twisted.invalid"]})
+        req = client.Request(
+            method=b"GET",
+            uri=b"http://twisted.invalid",
+            headers=headers,
+            bodyProducer=None,
+        )
+        req.uri = uri
+        req.writeTo(StringTransport())
diff --git a/src/twisted/web/test/test_webclient.py b/src/twisted/web/test/test_webclient.py
index 41cff5407..680e02780 100644
--- a/src/twisted/web/test/test_webclient.py
+++ b/src/twisted/web/test/test_webclient.py
@@ -5,10 +5,11 @@
 Tests for the old L{twisted.web.client} APIs, C{getPage} and friends.
 """
 
 from __future__ import division, absolute_import
 
+import io
 import os
 from errno import ENOSPC
 
 try:
     from urlparse import urlparse, urljoin
@@ -18,11 +19,12 @@ except ImportError:
 from twisted.python.compat import networkString, nativeString, intToBytes
 from twisted.trial import unittest, util
 from twisted.web import server, client, error, resource
 from twisted.web.static import Data
 from twisted.web.util import Redirect
-from twisted.internet import reactor, defer, interfaces
+from twisted.internet import address, reactor, defer, interfaces
+from twisted.internet.protocol import ClientFactory
 from twisted.python.filepath import FilePath
 from twisted.protocols.policies import WrappingFactory
 from twisted.test.proto_helpers import (
     StringTransport, waitUntilAllDisconnected, EventLoggingObserver)
 
@@ -33,10 +35,16 @@ except:
 
 from twisted import test
 from twisted.logger import (globalLogPublisher, FilteringLogObserver,
                             LogLevelFilterPredicate, LogLevel, Logger)
 
+from twisted.web.test.injectionhelpers import (
+    MethodInjectionTestsMixin,
+    URIInjectionTestsMixin,
+)
+
+
 
 serverPEM = FilePath(test.__file__).sibling('server.pem')
 serverPEMPath = serverPEM.asBytesMode().path
 
 
@@ -1517,5 +1525,308 @@ class DeprecationTests(unittest.TestCase):
     def test_httpDownloaderDeprecated(self):
         """
         L{client.HTTPDownloader} is deprecated.
         """
         self._testDeprecatedClass("HTTPDownloader")
+
+
+
+class GetPageMethodInjectionTests(
+        MethodInjectionTestsMixin,
+        unittest.SynchronousTestCase,
+):
+    """
+    Test L{client.getPage} against HTTP method injections.
+    """
+
+    def attemptRequestWithMaliciousMethod(self, method):
+        """
+        Attempt a request with the provided method.
+
+        @param method: see L{MethodInjectionTestsMixin}
+        """
+        uri = b'http://twisted.invalid'
+        client.getPage(uri, method=method)
+
+
+
+class GetPageURIInjectionTests(
+        URIInjectionTestsMixin,
+        unittest.SynchronousTestCase,
+):
+    """
+    Test L{client.getPage} against URI injections.
+    """
+
+    def attemptRequestWithMaliciousURI(self, uri):
+        """
+        Attempt a request with the provided URI.
+
+        @param uri: see L{URIInjectionTestsMixin}
+        """
+        client.getPage(uri)
+
+
+
+class DownloadPageMethodInjectionTests(
+        MethodInjectionTestsMixin,
+        unittest.SynchronousTestCase,
+):
+    """
+    Test L{client.getPage} against HTTP method injections.
+    """
+
+    def attemptRequestWithMaliciousMethod(self, method):
+        """
+        Attempt a request with the provided method.
+
+        @param method: see L{MethodInjectionTestsMixin}
+        """
+        uri = b'http://twisted.invalid'
+        client.downloadPage(uri, file=io.BytesIO(), method=method)
+
+
+
+class DownloadPageURIInjectionTests(
+        URIInjectionTestsMixin,
+        unittest.SynchronousTestCase,
+):
+    """
+    Test L{client.downloadPage} against URI injections.
+    """
+
+    def attemptRequestWithMaliciousURI(self, uri):
+        """
+        Attempt a request with the provided URI.
+
+        @param uri: see L{URIInjectionTestsMixin}
+        """
+        client.downloadPage(uri, file=io.BytesIO())
+
+
+
+def makeHTTPPageGetterFactory(protocolClass, method, host, path):
+    """
+    Make a L{ClientFactory} that can be used with
+    L{client.HTTPPageGetter} and its subclasses.
+
+    @param protocolClass: The protocol class
+    @type protocolClass: A subclass of L{client.HTTPPageGetter}
+
+    @param method: the HTTP method
+
+    @param host: the host
+
+    @param path: The URI path
+
+    @return: A L{ClientFactory}.
+    """
+    factory = ClientFactory.forProtocol(protocolClass)
+
+    factory.method = method
+    factory.host = host
+    factory.path = path
+
+    factory.scheme = b"http"
+    factory.port = 0
+    factory.headers = {}
+    factory.agent = b"User/Agent"
+    factory.cookies = {}
+
+    return factory
+
+
+
+class HTTPPageGetterMethodInjectionTests(
+        MethodInjectionTestsMixin,
+        unittest.SynchronousTestCase,
+):
+    """
+    Test L{client.HTTPPageGetter} against HTTP method injections.
+    """
+    protocolClass = client.HTTPPageGetter
+
+    def attemptRequestWithMaliciousMethod(self, method):
+        """
+        Attempt a request with the provided method.
+
+        @param method: L{MethodInjectionTestsMixin}
+        """
+        transport = StringTransport()
+        factory = makeHTTPPageGetterFactory(
+            self.protocolClass,
+            method=method,
+            host=b"twisted.invalid",
+            path=b"/",
+        )
+        getter = factory.buildProtocol(
+            address.IPv4Address("TCP", "127.0.0.1", 0),
+        )
+        getter.makeConnection(transport)
+
+
+
+class HTTPPageGetterURIInjectionTests(
+        URIInjectionTestsMixin,
+        unittest.SynchronousTestCase,
+):
+    """
+    Test L{client.HTTPPageGetter} against HTTP URI injections.
+    """
+    protocolClass = client.HTTPPageGetter
+
+    def attemptRequestWithMaliciousURI(self, uri):
+        """
+        Attempt a request with the provided URI.
+
+        @param uri: L{URIInjectionTestsMixin}
+        """
+        transport = StringTransport()
+        # Setting the host and path to the same value is imprecise but
+        # doesn't require parsing an invalid URI.
+        factory = makeHTTPPageGetterFactory(
+            self.protocolClass,
+            method=b"GET",
+            host=uri,
+            path=uri,
+        )
+        getter = factory.buildProtocol(
+            address.IPv4Address("TCP", "127.0.0.1", 0),
+        )
+        getter.makeConnection(transport)
+
+
+
+class HTTPPageDownloaderMethodInjectionTests(
+        HTTPPageGetterMethodInjectionTests
+):
+
+    """
+    Test L{client.HTTPPageDownloader} against HTTP method injections.
+    """
+    protocolClass = client.HTTPPageDownloader
+
+
+
+class HTTPPageDownloaderURIInjectionTests(
+        HTTPPageGetterURIInjectionTests
+):
+    """
+    Test L{client.HTTPPageDownloader} against HTTP URI injections.
+    """
+    protocolClass = client.HTTPPageDownloader
+
+
+
+class HTTPClientFactoryMethodInjectionTests(
+        MethodInjectionTestsMixin,
+        unittest.SynchronousTestCase,
+):
+    """
+    Tests L{client.HTTPClientFactory} against HTTP method injections.
+    """
+
+    def attemptRequestWithMaliciousMethod(self, method):
+        """
+        Attempt a request with the provided method.
+
+        @param method: L{MethodInjectionTestsMixin}
+        """
+        client.HTTPClientFactory(b"https://twisted.invalid", method)
+
+
+
+class HTTPClientFactoryURIInjectionTests(
+        URIInjectionTestsMixin,
+        unittest.SynchronousTestCase,
+):
+    """
+    Tests L{client.HTTPClientFactory} against HTTP URI injections.
+    """
+
+    def attemptRequestWithMaliciousURI(self, uri):
+        """
+        Attempt a request with the provided URI.
+
+        @param uri: L{URIInjectionTestsMixin}
+        """
+        client.HTTPClientFactory(uri)
+
+
+
+class HTTPClientFactorySetURLURIInjectionTests(
+        URIInjectionTestsMixin,
+        unittest.SynchronousTestCase,
+):
+    """
+    Tests L{client.HTTPClientFactory.setURL} against HTTP URI injections.
+    """
+
+    def attemptRequestWithMaliciousURI(self, uri):
+        """
+        Attempt a request with the provided URI.
+
+        @param uri: L{URIInjectionTestsMixin}
+        """
+        client.HTTPClientFactory(b"https://twisted.invalid").setURL(uri)
+
+
+
+class HTTPDownloaderMethodInjectionTests(
+        MethodInjectionTestsMixin,
+        unittest.SynchronousTestCase,
+):
+    """
+    Tests L{client.HTTPDownloader} against HTTP method injections.
+    """
+
+    def attemptRequestWithMaliciousMethod(self, method):
+        """
+        Attempt a request with the provided method.
+
+        @param method: L{MethodInjectionTestsMixin}
+        """
+        client.HTTPDownloader(
+            b"https://twisted.invalid",
+            io.BytesIO(),
+            method=method,
+        )
+
+
+
+class HTTPDownloaderURIInjectionTests(
+        URIInjectionTestsMixin,
+        unittest.SynchronousTestCase,
+):
+    """
+    Tests L{client.HTTPDownloader} against HTTP URI injections.
+    """
+
+    def attemptRequestWithMaliciousURI(self, uri):
+        """
+        Attempt a request with the provided URI.
+
+        @param uri: L{URIInjectionTestsMixin}
+        """
+        client.HTTPDownloader(uri, io.BytesIO())
+
+
+
+class HTTPDownloaderSetURLURIInjectionTests(
+        URIInjectionTestsMixin,
+        unittest.SynchronousTestCase,
+):
+    """
+    Tests L{client.HTTPDownloader.setURL} against HTTP URI injections.
+    """
+
+    def attemptRequestWithMaliciousURI(self, uri):
+        """
+        Attempt a request with the provided URI.
+
+        @param uri: L{URIInjectionTestsMixin}
+        """
+        downloader = client.HTTPDownloader(
+            b"https://twisted.invalid",
+            io.BytesIO(),
+        )
+        downloader.setURL(uri)

Vulnerable code:         self.method = method
        self.uri = uri
        requestLines.append(b' '.join([self.method, self.uri,
            b'HTTP/1.1\r\n']))
    _specialHeaders = set((b'host', b'user-agent', b'cookie', b'content-length'))
        method = getattr(self.factory, 'method', b'GET')
        self.sendCommand(method, self.factory.path)
        self.method = method
    uri = URI.fromBytes(url)
from twisted.trial.unittest import TestCase
from twisted.internet import reactor, defer, interfaces
Fixed code: import re

_VALID_METHOD = re.compile(
    br"\A[%s]+\Z" % (
        bytes().join(
            (
                b"!", b"#", b"$", b"%", b"&", b"'", b"*",
                b"+", b"-", b".", b"^", b"_", b"`", b"|", b"~",
                b"\x30-\x39",
                b"\x41-\x5a",
                b"\x61-\x7A",
            ),
        ),
    ),
)



def _ensureValidMethod(method):
    """
    An HTTP method is an HTTP token, which consists of any visible
    ASCII character that is not a delimiter (i.e. one of
    C{"(),/:;<=>?@[\\]{}}.)

    @param method: the method to check
    @type method: L{bytes}

    @return: the method if it is valid
    @rtype: L{bytes}

    @raise ValueError: if the method is not valid

    @see: U{https://tools.ietf.org/html/rfc7230#section-3.1.1},
        U{https://tools.ietf.org/html/rfc7230#section-3.2.6},
        U{https://tools.ietf.org/html/rfc5234#appendix-B.1}
    """
    if _VALID_METHOD.match(method):
        return method
    raise ValueError("Invalid method {!r}".format(method))



_VALID_URI = re.compile(br'\A[\x21-\x7e]+\Z')



def _ensureValidURI(uri):
    """
    A valid URI cannot contain control characters (i.e., characters
    between 0-32, inclusive and 127) or non-ASCII characters (i.e.,
    characters with values between 128-255, inclusive).

    @param uri: the URI to check
    @type uri: L{bytes}

    @return: the URI if it is valid
    @rtype: L{bytes}

    @raise ValueError: if the URI is not valid

    @see: U{https://tools.ietf.org/html/rfc3986#section-3.3},
        U{https://tools.ietf.org/html/rfc3986#appendix-A},
        U{https://tools.ietf.org/html/rfc5234#appendix-B.1}
    """
    if _VALID_URI.match(uri):
        return uri
    raise ValueError("Invalid URI {!r}".format(uri))



        self.method = _ensureValidMethod(method)
        self.uri = _ensureValidURI(uri)
        requestLines.append(
            b' '.join(
                [
                    _ensureValidMethod(self.method),
                    _ensureValidURI(self.uri),
                    b'HTTP/1.1\r\n',
                ]
            ),
        )
from twisted.web._newclient import _ensureValidURI, _ensureValidMethod


    _specialHeaders = set(
        (b'host', b'user-agent', b'cookie', b'content-length'),
    )
        method = _ensureValidMethod(getattr(self.factory, 'method', b'GET'))
        self.sendCommand(method, _ensureValidURI(self.factory.path))
        self.method = _ensureValidMethod(method)
        _ensureValidURI(url.strip())
    uri = URI.fromBytes(_ensureValidURI(url.strip()))

        method = _ensureValidMethod(method)

        uri = _ensureValidURI(uri.strip())
        uri = _ensureValidURI(uri.strip())

All HTTP clients in twisted.web.client now raise a ValueError when called with a method and/or URL that contain invalid characters.  This mitigates CVE-2019-12387.  Thanks to Alex Brasetvik for reporting this vulnerability.
"""
Helpers for URI and method injection tests.

@see: U{CVE-2019-12387}
"""

import string


UNPRINTABLE_ASCII = (
    frozenset(range(0, 128)) -
    frozenset(bytearray(string.printable, 'ascii'))
)

NONASCII = frozenset(range(128, 256))



class MethodInjectionTestsMixin(object):
    """
    A mixin that runs HTTP method injection tests.  Define
    L{MethodInjectionTestsMixin.attemptRequestWithMaliciousMethod} in
    a L{twisted.trial.unittest.SynchronousTestCase} subclass to test
    how HTTP client code behaves when presented with malicious HTTP
    methods.

    @see: U{CVE-2019-12387}
    """

    def attemptRequestWithMaliciousMethod(self, method):
        """
        Attempt to send a request with the given method.  This should
        synchronously raise a L{ValueError} if either is invalid.

        @param method: the method (e.g. C{GET\x00})

        @param uri: the URI

        @type method:
        """
        raise NotImplementedError()


    def test_methodWithCLRFRejected(self):
        """
        Issuing a request with a method that contains a carriage
        return and line feed fails with a L{ValueError}.
        """
        with self.assertRaises(ValueError) as cm:
            method = b"GET\r\nX-Injected-Header: value"
            self.attemptRequestWithMaliciousMethod(method)
        self.assertRegex(str(cm.exception), "^Invalid method")


    def test_methodWithUnprintableASCIIRejected(self):
        """
        Issuing a request with a method that contains unprintable
        ASCII characters fails with a L{ValueError}.
        """
        for c in UNPRINTABLE_ASCII:
            method = b"GET%s" % (bytearray([c]),)
            with self.assertRaises(ValueError) as cm:
                self.attemptRequestWithMaliciousMethod(method)
            self.assertRegex(str(cm.exception), "^Invalid method")


    def test_methodWithNonASCIIRejected(self):
        """
        Issuing a request with a method that contains non-ASCII
        characters fails with a L{ValueError}.
        """
        for c in NONASCII:
            method = b"GET%s" % (bytearray([c]),)
            with self.assertRaises(ValueError) as cm:
                self.attemptRequestWithMaliciousMethod(method)
            self.assertRegex(str(cm.exception), "^Invalid method")



class URIInjectionTestsMixin(object):
    """
    A mixin that runs HTTP URI injection tests.  Define
    L{MethodInjectionTestsMixin.attemptRequestWithMaliciousURI} in a
    L{twisted.trial.unittest.SynchronousTestCase} subclass to test how
    HTTP client code behaves when presented with malicious HTTP
    URIs.
    """

    def attemptRequestWithMaliciousURI(self, method):
        """
        Attempt to send a request with the given URI.  This should
        synchronously raise a L{ValueError} if either is invalid.

        @param uri: the URI.

        @type method:
        """
        raise NotImplementedError()


    def test_hostWithCRLFRejected(self):
        """
        Issuing a request with a URI whose host contains a carriage
        return and line feed fails with a L{ValueError}.
        """
        with self.assertRaises(ValueError) as cm:
            uri = b"http://twisted\r\n.invalid/path"
            self.attemptRequestWithMaliciousURI(uri)
        self.assertRegex(str(cm.exception), "^Invalid URI")


    def test_hostWithWithUnprintableASCIIRejected(self):
        """
        Issuing a request with a URI whose host contains unprintable
        ASCII characters fails with a L{ValueError}.
        """
        for c in UNPRINTABLE_ASCII:
            uri = b"http://twisted%s.invalid/OK" % (bytearray([c]),)
            with self.assertRaises(ValueError) as cm:
                self.attemptRequestWithMaliciousURI(uri)
            self.assertRegex(str(cm.exception), "^Invalid URI")


    def test_hostWithNonASCIIRejected(self):
        """
        Issuing a request with a URI whose host contains non-ASCII
        characters fails with a L{ValueError}.
        """
        for c in NONASCII:
            uri = b"http://twisted%s.invalid/OK" % (bytearray([c]),)
            with self.assertRaises(ValueError) as cm:
                self.attemptRequestWithMaliciousURI(uri)
            self.assertRegex(str(cm.exception), "^Invalid URI")


    def test_pathWithCRLFRejected(self):
        """
        Issuing a request with a URI whose path contains a carriage
        return and line feed fails with a L{ValueError}.
        """
        with self.assertRaises(ValueError) as cm:
            uri = b"http://twisted.invalid/\r\npath"
            self.attemptRequestWithMaliciousURI(uri)
        self.assertRegex(str(cm.exception), "^Invalid URI")


    def test_pathWithWithUnprintableASCIIRejected(self):
        """
        Issuing a request with a URI whose path contains unprintable
        ASCII characters fails with a L{ValueError}.
        """
        for c in UNPRINTABLE_ASCII:
            uri = b"http://twisted.invalid/OK%s" % (bytearray([c]),)
            with self.assertRaises(ValueError) as cm:
                self.attemptRequestWithMaliciousURI(uri)
            self.assertRegex(str(cm.exception), "^Invalid URI")


    def test_pathWithNonASCIIRejected(self):
        """
        Issuing a request with a URI whose path contains non-ASCII
        characters fails with a L{ValueError}.
        """
        for c in NONASCII:
            uri = b"http://twisted.invalid/OK%s" % (bytearray([c]),)
            with self.assertRaises(ValueError) as cm:
                self.attemptRequestWithMaliciousURI(uri)
            self.assertRegex(str(cm.exception), "^Invalid URI")
from twisted.trial.unittest import TestCase, SynchronousTestCase
from twisted.web.test.injectionhelpers import (
    MethodInjectionTestsMixin,
    URIInjectionTestsMixin,
)

class AgentMethodInjectionTests(
        FakeReactorAndConnectMixin,
        MethodInjectionTestsMixin,
        SynchronousTestCase,
):
    """
    Test L{client.Agent} against HTTP method injections.
    """

    def attemptRequestWithMaliciousMethod(self, method):
        """
        Attempt a request with the provided method.

        @param method: see L{MethodInjectionTestsMixin}
        """
        agent = client.Agent(self.createReactor())
        uri = b"http://twisted.invalid"
        agent.request(method, uri, client.Headers(), None)



class AgentURIInjectionTests(
        FakeReactorAndConnectMixin,
        URIInjectionTestsMixin,
        SynchronousTestCase,
):
    """
    Test L{client.Agent} against URI injections.
    """

    def attemptRequestWithMaliciousURI(self, uri):
        """
        Attempt a request with the provided method.

        @param uri: see L{URIInjectionTestsMixin}
        """
        agent = client.Agent(self.createReactor())
        method = b"GET"
        agent.request(method, uri, client.Headers(), None)






class RequestMethodInjectionTests(
        MethodInjectionTestsMixin,
        SynchronousTestCase,
):
    """
    Test L{client.Request} against HTTP method injections.
    """

    def attemptRequestWithMaliciousMethod(self, method):
        """
        Attempt a request with the provided method.

        @param method: see L{MethodInjectionTestsMixin}
        """
        client.Request(
            method=method,
            uri=b"http://twisted.invalid",
            headers=http_headers.Headers(),
            bodyProducer=None,
        )



class RequestWriteToMethodInjectionTests(
        MethodInjectionTestsMixin,
        SynchronousTestCase,
):
    """
    Test L{client.Request.writeTo} against HTTP method injections.
    """

    def attemptRequestWithMaliciousMethod(self, method):
        """
        Attempt a request with the provided method.

        @param method: see L{MethodInjectionTestsMixin}
        """
        headers = http_headers.Headers({b"Host": [b"twisted.invalid"]})
        req = client.Request(
            method=b"GET",
            uri=b"http://twisted.invalid",
            headers=headers,
            bodyProducer=None,
        )
        req.method = method
        req.writeTo(StringTransport())



class RequestURIInjectionTests(
        URIInjectionTestsMixin,
        SynchronousTestCase,
):
    """
    Test L{client.Request} against HTTP URI injections.
    """

    def attemptRequestWithMaliciousURI(self, uri):
        """
        Attempt a request with the provided URI.

        @param method: see L{URIInjectionTestsMixin}
        """
        client.Request(
            method=b"GET",
            uri=uri,
            headers=http_headers.Headers(),
            bodyProducer=None,
        )



class RequestWriteToURIInjectionTests(
        URIInjectionTestsMixin,
        SynchronousTestCase,
):
    """
    Test L{client.Request.writeTo} against HTTP method injections.
    """

    def attemptRequestWithMaliciousURI(self, uri):
        """
        Attempt a request with the provided method.

        @param method: see L{URIInjectionTestsMixin}
        """
        headers = http_headers.Headers({b"Host": [b"twisted.invalid"]})
        req = client.Request(
            method=b"GET",
            uri=b"http://twisted.invalid",
            headers=headers,
            bodyProducer=None,
        )
        req.uri = uri
        req.writeTo(StringTransport())
import io
from twisted.internet import address, reactor, defer, interfaces
from twisted.internet.protocol import ClientFactory
from twisted.web.test.injectionhelpers import (
    MethodInjectionTestsMixin,
    URIInjectionTestsMixin,
)





class GetPageMethodInjectionTests(
        MethodInjectionTestsMixin,
        unittest.SynchronousTestCase,
):
    """
    Test L{client.getPage} against HTTP method injections.
    """

    def attemptRequestWithMaliciousMethod(self, method):
        """
        Attempt a request with the provided method.

        @param method: see L{MethodInjectionTestsMixin}
        """
        uri = b'http://twisted.invalid'
        client.getPage(uri, method=method)



class GetPageURIInjectionTests(
        URIInjectionTestsMixin,
        unittest.SynchronousTestCase,
):
    """
    Test L{client.getPage} against URI injections.
    """

    def attemptRequestWithMaliciousURI(self, uri):
        """
        Attempt a request with the provided URI.

        @param uri: see L{URIInjectionTestsMixin}
        """
        client.getPage(uri)



class DownloadPageMethodInjectionTests(
        MethodInjectionTestsMixin,
        unittest.SynchronousTestCase,
):
    """
    Test L{client.getPage} against HTTP method injections.
    """

    def attemptRequestWithMaliciousMethod(self, method):
        """
        Attempt a request with the provided method.

        @param method: see L{MethodInjectionTestsMixin}
        """
        uri = b'http://twisted.invalid'
        client.downloadPage(uri, file=io.BytesIO(), method=method)



class DownloadPageURIInjectionTests(
        URIInjectionTestsMixin,
        unittest.SynchronousTestCase,
):
    """
    Test L{client.downloadPage} against URI injections.
    """

    def attemptRequestWithMaliciousURI(self, uri):
        """
        Attempt a request with the provided URI.

        @param uri: see L{URIInjectionTestsMixin}
        """
        client.downloadPage(uri, file=io.BytesIO())



def makeHTTPPageGetterFactory(protocolClass, method, host, path):
    """
    Make a L{ClientFactory} that can be used with
    L{client.HTTPPageGetter} and its subclasses.

    @param protocolClass: The protocol class
    @type protocolClass: A subclass of L{client.HTTPPageGetter}

    @param method: the HTTP method

    @param host: the host

    @param path: The URI path

    @return: A L{ClientFactory}.
    """
    factory = ClientFactory.forProtocol(protocolClass)

    factory.method = method
    factory.host = host
    factory.path = path

    factory.scheme = b"http"
    factory.port = 0
    factory.headers = {}
    factory.agent = b"User/Agent"
    factory.cookies = {}

    return factory



class HTTPPageGetterMethodInjectionTests(
        MethodInjectionTestsMixin,
        unittest.SynchronousTestCase,
):
    """
    Test L{client.HTTPPageGetter} against HTTP method injections.
    """
    protocolClass = client.HTTPPageGetter

    def attemptRequestWithMaliciousMethod(self, method):
        """
        Attempt a request with the provided method.

        @param method: L{MethodInjectionTestsMixin}
        """
        transport = StringTransport()
        factory = makeHTTPPageGetterFactory(
            self.protocolClass,
            method=method,
            host=b"twisted.invalid",
            path=b"/",
        )
        getter = factory.buildProtocol(
            address.IPv4Address("TCP", "127.0.0.1", 0),
        )
        getter.makeConnection(transport)



class HTTPPageGetterURIInjectionTests(
        URIInjectionTestsMixin,
        unittest.SynchronousTestCase,
):
    """
    Test L{client.HTTPPageGetter} against HTTP URI injections.
    """
    protocolClass = client.HTTPPageGetter

    def attemptRequestWithMaliciousURI(self, uri):
        """
        Attempt a request with the provided URI.

        @param uri: L{URIInjectionTestsMixin}
        """
        transport = StringTransport()
        # Setting the host and path to the same value is imprecise but
        # doesn't require parsing an invalid URI.
        factory = makeHTTPPageGetterFactory(
            self.protocolClass,
            method=b"GET",
            host=uri,
            path=uri,
        )
        getter = factory.buildProtocol(
            address.IPv4Address("TCP", "127.0.0.1", 0),
        )
        getter.makeConnection(transport)



class HTTPPageDownloaderMethodInjectionTests(
        HTTPPageGetterMethodInjectionTests
):

    """
    Test L{client.HTTPPageDownloader} against HTTP method injections.
    """
    protocolClass = client.HTTPPageDownloader



class HTTPPageDownloaderURIInjectionTests(
        HTTPPageGetterURIInjectionTests
):
    """
    Test L{client.HTTPPageDownloader} against HTTP URI injections.
    """
    protocolClass = client.HTTPPageDownloader



class HTTPClientFactoryMethodInjectionTests(
        MethodInjectionTestsMixin,
        unittest.SynchronousTestCase,
):
    """
    Tests L{client.HTTPClientFactory} against HTTP method injections.
    """

    def attemptRequestWithMaliciousMethod(self, method):
        """
        Attempt a request with the provided method.

        @param method: L{MethodInjectionTestsMixin}
        """
        client.HTTPClientFactory(b"https://twisted.invalid", method)



class HTTPClientFactoryURIInjectionTests(
        URIInjectionTestsMixin,
        unittest.SynchronousTestCase,
):
    """
    Tests L{client.HTTPClientFactory} against HTTP URI injections.
    """

    def attemptRequestWithMaliciousURI(self, uri):
        """
        Attempt a request with the provided URI.

        @param uri: L{URIInjectionTestsMixin}
        """
        client.HTTPClientFactory(uri)



class HTTPClientFactorySetURLURIInjectionTests(
        URIInjectionTestsMixin,
        unittest.SynchronousTestCase,
):
    """
    Tests L{client.HTTPClientFactory.setURL} against HTTP URI injections.
    """

    def attemptRequestWithMaliciousURI(self, uri):
        """
        Attempt a request with the provided URI.

        @param uri: L{URIInjectionTestsMixin}
        """
        client.HTTPClientFactory(b"https://twisted.invalid").setURL(uri)



class HTTPDownloaderMethodInjectionTests(
        MethodInjectionTestsMixin,
        unittest.SynchronousTestCase,
):
    """
    Tests L{client.HTTPDownloader} against HTTP method injections.
    """

    def attemptRequestWithMaliciousMethod(self, method):
        """
        Attempt a request with the provided method.

        @param method: L{MethodInjectionTestsMixin}
        """
        client.HTTPDownloader(
            b"https://twisted.invalid",
            io.BytesIO(),
            method=method,
        )



class HTTPDownloaderURIInjectionTests(
        URIInjectionTestsMixin,
        unittest.SynchronousTestCase,
):
    """
    Tests L{client.HTTPDownloader} against HTTP URI injections.
    """

    def attemptRequestWithMaliciousURI(self, uri):
        """
        Attempt a request with the provided URI.

        @param uri: L{URIInjectionTestsMixin}
        """
        client.HTTPDownloader(uri, io.BytesIO())



class HTTPDownloaderSetURLURIInjectionTests(
        URIInjectionTestsMixin,
        unittest.SynchronousTestCase,
):
    """
    Tests L{client.HTTPDownloader.setURL} against HTTP URI injections.
    """

    def attemptRequestWithMaliciousURI(self, uri):
        """
        Attempt a request with the provided URI.

        @param uri: L{URIInjectionTestsMixin}
        """
        downloader = client.HTTPDownloader(
            b"https://twisted.invalid",
            io.BytesIO(),
        )
        downloader.setURL(uri)