Patched code sample

import html
import re
import tornado.web
import tornado.ioloop
import tornado.httpserver


class VulnerableHandler(tornado.web.RequestHandler):
    def get(self):
        """
        This handler demonstrates the vulnerability.
        A malicious 'reason' phrase is passed directly to set_status.
        In vulnerable versions, this is not escaped.
        """
        # Payload for XSS injection into the error page
        xss_payload = "Not Found<script>alert('XSS vulnerability demonstrated')</script>"
        
        # Payload for HTTP Header Injection (CRLF injection)
        header_injection_payload = "OK\r\nInjected-Header: Malicious-Value"

        payload_type = self.get_argument("payload", "xss")

        if payload_type == "xss":
            # This would trigger an XSS vulnerability on the default error page
            # in unpatched versions of Tornado.
            self.set_status(404, reason=xss_payload)
        elif payload_type == "header":
            # This would trigger an HTTP header injection in unpatched versions.
            # We use a 200 status code to show the reason phrase in the status line.
            self.set_status(200, reason=header_injection_payload)
            self.write("Check the raw HTTP response for an injected header.")
        else:
            self.set_status(400, "Invalid payload type specified.")


class FixedHandler(tornado.web.RequestHandler):
    def _sanitize_reason(self, reason):
        """
        Represents the fix applied in patched Tornado versions.
        It removes control characters like carriage returns and newlines
        to prevent HTTP header injection (CRLF injection).
        The default error page template in Tornado also performs HTML escaping,
        which prevents XSS. This function focuses on the header part of the fix.
        """
        return re.sub(r"[\r\n]", " ", reason)

    def set_status(self, status_code, reason=None):
        """
        Overrides the default set_status to demonstrate the fix.
        The 'reason' is sanitized before being passed to the original method.
        """
        if reason:
            sanitized_reason = self._sanitize_reason(reason)
            # In the actual Tornado fix, this sanitization happens internally.
            # We are calling the parent method with the sanitized reason.
            super().set_status(status_code, reason=sanitized_reason)
        else:
            super().set_status(status_code, reason=reason)

    def write_error(self, status_code, **kwargs):
        """
        Overrides write_error to demonstrate the XSS fix.
        The reason phrase is fetched and explicitly HTML-escaped before rendering.
        Patched Tornado versions ensure this escaping happens in the default template.
        """
        reason = kwargs.get('reason', 'An error occurred')
        
        # Manually apply the HTML escaping fix
        escaped_reason = html.escape(reason)
        
        self.set_header('Content-Type', 'text/html')
        self.finish(f"""
        <html>
            <title>{status_code}: {escaped_reason}</title>
            <body>
                <h2>HTTP Error {status_code}: {escaped_reason}</h2>
                <p>The 'reason' has been HTML-escaped, preventing XSS.</p>
            </body>
        </html>
        """)

    def get(self):
        """
        This handler demonstrates the fix.
        The same malicious payloads are used, but the overridden methods
        apply sanitization and escaping.
        """
        # Payload for XSS injection
        xss_payload = "Not Found<script>alert('This should not execute')</script>"
        
        # Payload for HTTP Header Injection
        header_injection_payload = "OK\r\nInjected-Header: Malicious-Value"

        payload_type = self.get_argument("payload", "xss")

        try:
            if payload_type == "xss":
                # The custom write_error will be triggered by raising an HTTPError.
                raise tornado.web.HTTPError(404, reason=xss_payload)
            elif payload_type == "header":
                # The overridden set_status will sanitize the reason.
                self.set_status(200, reason=header_injection_payload)
                self.write("Check the raw HTTP response. The header injection is prevented.")
            else:
                self.set_status(400, "Invalid payload type specified.")
        except tornado.web.HTTPError as e:
            self.write_error(e.status_code, reason=e.reason)


def make_app():
    return tornado.web.Application([
        (r"/vulnerable", VulnerableHandler),
        (r"/fixed", FixedHandler),
    ])


if __name__ == "__main__":
    app = make_app()
    http_server = tornado.httpserver.HTTPServer(app)
    port = 8888
    http_server.listen(port)
    print(f"Server running on http://localhost:{port}")
    print("Demonstrate vulnerability: http://localhost:8888/vulnerable?payload=xss")
    print("Demonstrate fix: http://localhost:8888/fixed?payload=xss")
    print("\nTo test header injection, use curl:")
    print("Vulnerable: curl -v 'http://localhost:8888/vulnerable?payload=header'")
    print("Fixed:      curl -v 'http://localhost:8888/fixed?payload=header'")
    tornado.ioloop.IOLoop.current().start()