20383610525

Committed 19 Dec 2025 09:59PM UTC coverage: 34.865% (-62.6%) from 97.512%

Build # 20383610525

Build Type

Pull #603

github

Committed by

terjekv

Commit Message

Test disabling parallel testing

Pull Request Pull Request #603: Parallel testing

Run Details

3293 of 9445 relevant lines covered (34.87%)

1.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

24.05

/mreg/middleware/logging_http.py

"""Middleware to handle logging of HTTP requests and responses."""
import http
import logging
import time
import uuid
from typing import Callable, cast

import structlog
import sentry_sdk
import traceback
from django.conf import settings
from django.http import HttpRequest, HttpResponse

mreg_logger = structlog.getLogger("mreg.http")

LOGMAP = {
    "ERROR": logging.ERROR,
    "CRITICAL": logging.CRITICAL,
    "WARNING": logging.WARNING,
    "INFO": logging.INFO,
    "DEBUG": logging.DEBUG,
}


class LoggingMiddleware:
    """Middleware to log HTTP requests and responses.

    This middleware checks the status code of the response and logs a message
    based on the response code range (success, redirection, client error, or server error).
    The time it took to process the response is also logged.
    """

    def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
        """Initialize the middleware.

        :param get_response: A reference to the next middleware or view in the chain.
        """
        self.get_response = get_response

    def __call__(self, request: HttpRequest) -> HttpResponse:
        """Process the request and log the response.

        :param request: The incoming request.
        :return: A response object
        """
        start_time = int(time.time())

        self.log_request(request)
 
        try:
            response = self.get_response(request)
        except Exception as e: # pragma: no cover (this is somewhat tricky to properly test)
            self.log_exception(request, e, start_time)
            raise

        self.log_response(request, response, start_time)
        return response

    def _get_body(self, request: HttpRequest) -> str:
        """Get the request body as a string, or '<Binary Data>' if it's binary.

        We currently do not support multipart/form-data requests.
        """
        if request.POST:
            return request.POST.dict()

        try:
            body = request.body.decode("utf-8")
        except UnicodeDecodeError:
            return "<Binary Data>"

        # Try to remove the content-type line and leading line breaks
        body = body.split("\n", 1)[-1]  # Removes the first line
        body = body.lstrip()  # Removes leading line breaks

        # Limit the size of the body logged
        return body[: settings.LOGGING_MAX_BODY_LENGTH]

    def _get_request_header(
        self, request: HttpRequest, header_key: str, meta_key: str
    ) -> str:
        """Get the value of a header from the request, either via headers or META."""
        if hasattr(request, "headers"): # pragma: no cover
            return request.headers.get(header_key)

        return request.META.get(meta_key)

    def log_request(self, request: HttpRequest) -> None:
        """Log the request."""
        request_id = self._get_request_header(
            request, "x-request-id", "HTTP_X_REQUEST_ID"
        ) or str(uuid.uuid4())
        correlation_id = self._get_request_header(
            request, "x-correlation-id", "HTTP_X_CORRELATION_ID"
        )
        structlog.contextvars.bind_contextvars(request_id=request_id)
        if correlation_id:
            structlog.contextvars.bind_contextvars(correlation_id=correlation_id)

        remote_ip = request.META.get("REMOTE_ADDR")

        # Check for a proxy address
        x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
        if x_forwarded_for:
            proxy_ip = x_forwarded_for.split(",")[0]
        else:
            proxy_ip = ""

        user_agent = self._get_request_header(request, "user-agent", "HTTP_USER_AGENT")

        # Size of request
        request_size = len(request.body)

        mreg_logger.bind(
            method=request.method,
            remote_ip=remote_ip,
            proxy_ip=proxy_ip,
            user_agent=user_agent,
            path=request.path_info,
            query_string=request.META.get("QUERY_STRING"),
            request_size=request_size,
            content=self._get_body(request),
        ).info("request")

    def log_response(
        self, request: HttpRequest, response: HttpResponse, start_time: int
    ) -> HttpResponse:
        """Log the response."""
        end_time = time.time()
        status_code = response.status_code
        run_time_ms = (end_time - start_time) * 1000

        status_label = cast(str, http.client.responses[status_code])

        if status_code in range(200, 399):
            log_level = logging.INFO
        elif status_code in range(400, 499):
            log_level = logging.WARNING
        else:
            log_level = logging.ERROR

        extra_data = {}

        if run_time_ms >= settings.REQUESTS_THRESHOLD_VERY_SLOW:
            extra_data["original_log_level"] = log_level
            extra_data["very_slow_response"] = True
            log_level = LOGMAP[settings.REQUESTS_LOG_LEVEL_VERY_SLOW.upper()]
        elif run_time_ms >= settings.REQUESTS_THRESHOLD_SLOW:
            extra_data["original_log_level"] = log_level
            extra_data["slow_response"] = True
            log_level = LOGMAP[settings.REQUESTS_LOG_LEVEL_SLOW.upper()]

        content = ""
        if "application/json" in response.headers.get("Content-Type", ""):
            content = response.content.decode("utf-8")

        username = request.user.username

        user_agent = self._get_request_header(request, "user-agent", "HTTP_USER_AGENT")

        mreg_logger.bind(
            user=username,
            method=request.method,
            user_agent=user_agent,
            status_code=status_code,
            status_label=status_label,
            path=request.path_info,
            query_string=request.META.get("QUERY_STRING"),  
            content=content,
            **extra_data,
            run_time_ms=round(run_time_ms, 2),
        ).log(log_level, "response")

        contextvars = structlog.contextvars.get_contextvars()
        response["X-Request-ID"] = contextvars["request_id"]
        if "correlation_id" in contextvars:
            response["X-Correlation-ID"] = contextvars["correlation_id"]

        structlog.contextvars.clear_contextvars()

        return response

    def log_exception(self, request: HttpRequest, exception: Exception, start_time: float) -> None: # pragma: no cover
        """Log an exception that occurred during request processing."""
        end_time = time.time()
        run_time_ms = (end_time - start_time) * 1000

        stack_trace = traceback.format_exc()

        username = getattr(request.user, 'username', 'AnonymousUser')
        user_agent = self._get_request_header(request, "user-agent", "HTTP_USER_AGENT")

        # Log the exception with stack trace
        mreg_logger.bind(
            user=username,
            method=request.method,
            user_agent=user_agent,
            path=request.path_info,
            query_string=request.META.get("QUERY_STRING"),
            run_time_ms=round(run_time_ms, 2),
        ).error(
            "Unhandled exception occurred",
            exception_string=str(exception),
            exception_type=type(exception).__name__,
            stack_trace=stack_trace,
        )

        # Capture the exception with Sentry and add context
        with sentry_sdk.push_scope() as scope:
            scope.set_user({"username": username})
            scope.set_extra("method", request.method)
            scope.set_extra("user_agent", user_agent)
            scope.set_extra("path", request.path_info)
            scope.set_extra("query_string", request.META.get("QUERY_STRING"))
            scope.set_extra("run_time_ms", round(run_time_ms, 2))
            scope.set_extra("exception_string", str(exception))
            scope.set_extra("stack_trace", stack_trace)

            scope.set_extra("request_body", self._get_body(request))

            # Capture the exception
            sentry_sdk.capture_exception(exception)

1	"""Middleware to handle logging of HTTP requests and responses."""
2	import http	5✔
3	import logging	5✔
4	import time	5✔
5	import uuid	5✔
6	from typing import Callable, cast	5✔
7
8	import structlog	5✔
9	import sentry_sdk	5✔
10	import traceback	5✔
11	from django.conf import settings	5✔
12	from django.http import HttpRequest, HttpResponse	5✔
13
14	mreg_logger = structlog.getLogger("mreg.http")	5✔
15
16	LOGMAP = {	5✔
17	"ERROR": logging.ERROR,
18	"CRITICAL": logging.CRITICAL,
19	"WARNING": logging.WARNING,
20	"INFO": logging.INFO,
21	"DEBUG": logging.DEBUG,
22	}
23
24
25	class LoggingMiddleware:	5✔
26	"""Middleware to log HTTP requests and responses.
27
28	This middleware checks the status code of the response and logs a message
29	based on the response code range (success, redirection, client error, or server error).
30	The time it took to process the response is also logged.
31	"""
32
33	def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:	5✔
34	"""Initialize the middleware.
35
36	:param get_response: A reference to the next middleware or view in the chain.
37	"""
38	self.get_response = get_response	×
39
40	def __call__(self, request: HttpRequest) -> HttpResponse:	5✔
41	"""Process the request and log the response.
42
43	:param request: The incoming request.
44	:return: A response object
45	"""
46	start_time = int(time.time())	×
47
48	self.log_request(request)	×
49
50	try:	×
51	response = self.get_response(request)	×
52	except Exception as e: # pragma: no cover (this is somewhat tricky to properly test)
53	self.log_exception(request, e, start_time)
54	raise
55
56	self.log_response(request, response, start_time)	×
57	return response	×
58
59	def _get_body(self, request: HttpRequest) -> str:	5✔
60	"""Get the request body as a string, or '<Binary Data>' if it's binary.
61
62	We currently do not support multipart/form-data requests.
63	"""
64	if request.POST:	×
65	return request.POST.dict()	×
66
67	try:	×
68	body = request.body.decode("utf-8")	×
69	except UnicodeDecodeError:	×
70	return "<Binary Data>"	×
71
72	# Try to remove the content-type line and leading line breaks
73	body = body.split("\n", 1)[-1] # Removes the first line	×
74	body = body.lstrip() # Removes leading line breaks	×
75
76	# Limit the size of the body logged
77	return body[: settings.LOGGING_MAX_BODY_LENGTH]	×
78
79	def _get_request_header(	5✔
80	self, request: HttpRequest, header_key: str, meta_key: str
81	) -> str:
82	"""Get the value of a header from the request, either via headers or META."""
83	if hasattr(request, "headers"): # pragma: no cover
84	return request.headers.get(header_key)
85
86	return request.META.get(meta_key)	×
87
88	def log_request(self, request: HttpRequest) -> None:	5✔
89	"""Log the request."""
90	request_id = self._get_request_header(	×
91	request, "x-request-id", "HTTP_X_REQUEST_ID"
92	) or str(uuid.uuid4())
93	correlation_id = self._get_request_header(	×
94	request, "x-correlation-id", "HTTP_X_CORRELATION_ID"
95	)
96	structlog.contextvars.bind_contextvars(request_id=request_id)	×
97	if correlation_id:	×
98	structlog.contextvars.bind_contextvars(correlation_id=correlation_id)	×
99
100	remote_ip = request.META.get("REMOTE_ADDR")	×
101
102	# Check for a proxy address
103	x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")	×
104	if x_forwarded_for:	×
105	proxy_ip = x_forwarded_for.split(",")[0]	×
106	else:
107	proxy_ip = ""	×
108
109	user_agent = self._get_request_header(request, "user-agent", "HTTP_USER_AGENT")	×
110
111	# Size of request
112	request_size = len(request.body)	×
113
114	mreg_logger.bind(	×
115	method=request.method,
116	remote_ip=remote_ip,
117	proxy_ip=proxy_ip,
118	user_agent=user_agent,
119	path=request.path_info,
120	query_string=request.META.get("QUERY_STRING"),
121	request_size=request_size,
122	content=self._get_body(request),
123	).info("request")
124
125	def log_response(	5✔
126	self, request: HttpRequest, response: HttpResponse, start_time: int
127	) -> HttpResponse:
128	"""Log the response."""
129	end_time = time.time()	×
130	status_code = response.status_code	×
131	run_time_ms = (end_time - start_time) * 1000	×
132
133	status_label = cast(str, http.client.responses[status_code])	×
134
135	if status_code in range(200, 399):	×
136	log_level = logging.INFO	×
137	elif status_code in range(400, 499):	×
138	log_level = logging.WARNING	×
139	else:
140	log_level = logging.ERROR	×
141
142	extra_data = {}	×
143
144	if run_time_ms >= settings.REQUESTS_THRESHOLD_VERY_SLOW:	×
145	extra_data["original_log_level"] = log_level	×
146	extra_data["very_slow_response"] = True	×
147	log_level = LOGMAP[settings.REQUESTS_LOG_LEVEL_VERY_SLOW.upper()]	×
148	elif run_time_ms >= settings.REQUESTS_THRESHOLD_SLOW:	×
149	extra_data["original_log_level"] = log_level	×
150	extra_data["slow_response"] = True	×
151	log_level = LOGMAP[settings.REQUESTS_LOG_LEVEL_SLOW.upper()]	×
152
153	content = ""	×
154	if "application/json" in response.headers.get("Content-Type", ""):	×
155	content = response.content.decode("utf-8")	×
156
157	username = request.user.username	×
158
159	user_agent = self._get_request_header(request, "user-agent", "HTTP_USER_AGENT")	×
160
161	mreg_logger.bind(	×
162	user=username,
163	method=request.method,
164	user_agent=user_agent,
165	status_code=status_code,
166	status_label=status_label,
167	path=request.path_info,
168	query_string=request.META.get("QUERY_STRING"),
169	content=content,
170	**extra_data,
171	run_time_ms=round(run_time_ms, 2),
172	).log(log_level, "response")
173
174	contextvars = structlog.contextvars.get_contextvars()	×
175	response["X-Request-ID"] = contextvars["request_id"]	×
176	if "correlation_id" in contextvars:	×
177	response["X-Correlation-ID"] = contextvars["correlation_id"]	×
178
179	structlog.contextvars.clear_contextvars()	×
180
181	return response	×
182
183	def log_exception(self, request: HttpRequest, exception: Exception, start_time: float) -> None: # pragma: no cover
184	"""Log an exception that occurred during request processing."""
185	end_time = time.time()
186	run_time_ms = (end_time - start_time) * 1000
187
188	stack_trace = traceback.format_exc()
189
190	username = getattr(request.user, 'username', 'AnonymousUser')
191	user_agent = self._get_request_header(request, "user-agent", "HTTP_USER_AGENT")
192
193	# Log the exception with stack trace
194	mreg_logger.bind(
195	user=username,
196	method=request.method,
197	user_agent=user_agent,
198	path=request.path_info,
199	query_string=request.META.get("QUERY_STRING"),
200	run_time_ms=round(run_time_ms, 2),
201	).error(
202	"Unhandled exception occurred",
203	exception_string=str(exception),
204	exception_type=type(exception).__name__,
205	stack_trace=stack_trace,
206	)
207
208	# Capture the exception with Sentry and add context
209	with sentry_sdk.push_scope() as scope:
210	scope.set_user({"username": username})
211	scope.set_extra("method", request.method)
212	scope.set_extra("user_agent", user_agent)
213	scope.set_extra("path", request.path_info)
214	scope.set_extra("query_string", request.META.get("QUERY_STRING"))
215	scope.set_extra("run_time_ms", round(run_time_ms, 2))
216	scope.set_extra("exception_string", str(exception))
217	scope.set_extra("stack_trace", stack_trace)
218
219	scope.set_extra("request_body", self._get_body(request))
220
221	# Capture the exception
222	sentry_sdk.capture_exception(exception)

unioslo / mreg / 20383610525

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous