Log Injection

PoC (controlled-repro in a test environment): Idea: Before the fix, log lines could be polluted by control characters or CRLF injection from user-controlled request data, potentially producing fake lines or altering log readability. The patch shows escaping of such input when logging. 1) Reproduce the injection (pre-fix behavior): - Start a Django runserver using codebase before the patch. - Send a request whose path carries CR/LF or ANSI control codes, for example encoded as raw characters in the path: "/normal\r\nINJECT:1" (or URL-encode: "/normal%0D%0AINJECT%3A1") - Expected (unfixed) log line would include the raw CR/LF, producing a second log line or mangled log content, e.g.: INFO:django.server:GET /normal INJECT:1 404 2) Reproduce the fix (post-fix behavior): - Use the patched code (as in commit 9bb83925d6c231e964f8b54efbc982fb1333da27). - Send the same request path as above. - The log line will escape control characters, e.g. CR/LF is rendered as \r and \n, preventing the injection from creating new log lines: INFO:django.server:GET /normal\r\nINJECT:1 404 3) Minimal in-repo demonstration (standalone, without Django runtime): - This Python snippet demonstrates the escaping behavior implemented by log_message in the patch. import logging import io # Simulated pre-fix behavior (no escaping) def simulate_without_fix(path, status): logger = logging.getLogger("server_without_fix") logger.setLevel(logging.INFO) stream = io.StringIO() handler = logging.StreamHandler(stream) handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s")) logger.addHandler(handler) # This would log the raw path, potentially injecting newlines logger.info("GET %s %s", path, status) logger.removeHandler(handler) return stream.getvalue() # Simulated post-fix behavior (escaping) def simulate_with_fix(path, status): logger = logging.getLogger("server_with_fix") logger.setLevel(logging.INFO) stream = io.StringIO() handler = logging.StreamHandler(stream) handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s")) logger.addHandler(handler) escaped_path = path.encode("unicode_escape").decode("ascii") # Escaped path is logged, preventing newline injection logger.info("GET %s %s", escaped_path, status) logger.removeHandler(handler) return stream.getvalue() if __name__ == "__main__": path = "/normal\r\nINJECT:1" # example crafted input print("Without fix:\n" + simulate_without_fix(path, "404")) print("\nWith fix:\n" + simulate_with_fix(path, "404")) Expected outcome: - simulate_without_fix shows a log containing an actual newline due to the CRLF in the path. - simulate_with_fix shows the CR/LF escaped as \r and \n, preventing log-line injection.

diff --git a/django/core/servers/basehttp.py b/django/core/servers/basehttp.py index 41719034fb72..d62b88d28651 100644 --- a/django/core/servers/basehttp.py +++ b/django/core/servers/basehttp.py @@ -18,6 +18,7 @@ from django.core.handlers.wsgi import LimitedStream from django.core.wsgi import get_wsgi_application from django.db import connections +from django.utils.log import log_message from django.utils.module_loading import import_string __all__ = ("WSGIServer", "WSGIRequestHandler") @@ -182,35 +183,27 @@ def address_string(self): return self.client_address[0] def log_message(self, format, *args): - extra = { - "request": self.request, - "server_time": self.log_date_time_string(), - } - if args[1][0] == "4": + if args[1][0] == "4" and args[0].startswith("\x16\x03"): # 0x16 = Handshake, 0x03 = SSL 3.0 or TLS 1.x - if args[0].startswith("\x16\x03"): - extra["status_code"] = 500 - logger.error( - "You're accessing the development server over HTTPS, but " - "it only supports HTTP.", - extra=extra, - ) - return - - if args[1].isdigit() and len(args[1]) == 3: + format = ( + "You're accessing the development server over HTTPS, but it only " + "supports HTTP." + ) + status_code = 500 + args = () + elif args[1].isdigit() and len(args[1]) == 3: status_code = int(args[1]) - extra["status_code"] = status_code - - if status_code >= 500: - level = logger.error - elif status_code >= 400: - level = logger.warning - else: - level = logger.info else: - level = logger.info - - level(format, *args, extra=extra) + status_code = None + + log_message( + logger, + format, + *args, + request=self.request, + status_code=status_code, + server_time=self.log_date_time_string(), + ) def get_environ(self): # Strip all headers with underscores in the name before constructing diff --git a/django/utils/log.py b/django/utils/log.py index 67a40270f076..d4e96a981679 100644 --- a/django/utils/log.py +++ b/django/utils/log.py @@ -214,6 +214,46 @@ def uses_server_time(self): return self._fmt.find("{server_time}") >= 0 +def log_message( + logger, + message, + *args, + level=None, + status_code=None, + request=None, + exception=None, + **extra, +): + """Log `message` using `logger` based on `status_code` and logger `level`. + + Pass `request`, `status_code` (if defined) and any provided `extra` as such + to the logging method, + + Arguments from `args` will be escaped to avoid potential log injections. + + """ + extra = {"request": request, **extra} + if status_code is not None: + extra["status_code"] = status_code + if level is None: + if status_code >= 500: + level = "error" + elif status_code >= 400: + level = "warning" + + escaped_args = tuple( + a.encode("unicode_escape").decode("ascii") if isinstance(a, str) else a + for a in args + ) + + getattr(logger, level or "info")( + message, + *escaped_args, + extra=extra, + exc_info=exception, + ) + + def log_response( message, *args, @@ -237,26 +277,13 @@ def log_response( if getattr(response, "_has_been_logged", False): return - if level is None: - if response.status_code >= 500: - level = "error" - elif response.status_code >= 400: - level = "warning" - else: - level = "info" - - escaped_args = tuple( - a.encode("unicode_escape").decode("ascii") if isinstance(a, str) else a - for a in args - ) - - getattr(logger, level)( + log_message( + logger, message, - *escaped_args, - extra={ - "status_code": response.status_code, - "request": request, - }, - exc_info=exception, + *args, + level=level, + status_code=response.status_code, + request=request, + exception=exception, ) response._has_been_logged = True diff --git a/tests/servers/test_basehttp.py b/tests/servers/test_basehttp.py index cc4701114a78..9190fc8a204d 100644 --- a/tests/servers/test_basehttp.py +++ b/tests/servers/test_basehttp.py @@ -50,6 +50,21 @@ def test_log_message(self): cm.records[0].levelname, wrong_level.upper() ) + def test_log_message_escapes_control_sequences(self): + request = WSGIRequest(self.request_factory.get("/").environ) + request.makefile = lambda *args, **kwargs: BytesIO() + handler = WSGIRequestHandler(request, "192.168.0.2", None) + + malicious_path = "\x1b[31mALERT\x1b[0m" + + with self.assertLogs("django.server", "WARNING") as cm: + handler.log_message("GET %s %s", malicious_path, "404") + + log = cm.output[0] + + self.assertNotIn("\x1b[31m", log) + self.assertIn("\\x1b[31mALERT\\x1b[0m", log) + def test_https(self): request = WSGIRequest(self.request_factory.get("/").environ) request.makefile = lambda *args, **kwargs: BytesIO()