CVE-2019-12387 (Python, published 2019-06-10T12:29Z)
Description: 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.
Severity: MEDIUM (CVSS v3 base score: 6.1)
Weakness: CWE-74 \u2014 Improper Neutralization of Special Elements in Output Used by a Downstream Component ('Injection')
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.
Repository: https://github.com/twisted/twisted
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 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.

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)