onconova.core.history.middleware

SENSITIVE_KEYS module-attribute

audit_logger module-attribute

ASGIRequest

Bases: DjangoRequest, ASGIRequest

ASGIRequest is a request class that inherits from both DjangoRequest and DjangoASGIRequest, combining their functionality to handle HTTP requests in ASGI-compatible Django applications.

This class serves as a unified interface for processing requests in environments where ASGI support is required, such as asynchronous web servers.

Inheritance

DjangoRequest: Standard Django HTTP request handling. DjangoASGIRequest: ASGI-specific request handling for asynchronous support.

AuditLogMiddleware(get_response)

Middleware that logs detailed audit information for each HTTP request and response.

This middleware captures and logs the following information: - User identification (ID, username, access level), or marks as anonymous if unauthenticated. - Request metadata such as IP address, HTTP method, endpoint path, user agent, and processing duration. - Request and response data, with sensitive fields redacted and optionally compressed and base64-encoded. - HTTP status code of the response.

Sensitive fields in request and response data are redacted based on a predefined list of keys. Handles both JSON and non-JSON responses gracefully, and ensures that unreadable or non-JSON data is marked accordingly.

Parameters:

Name	Type	Description	Default
get_response	callable	The next middleware or view in the Django request/response chain.	required

Source code in onconova/core/history/middleware.py

def __init__(self, get_response):
    self.get_response = get_response

get_response instance-attribute

__call__(request)

Handles incoming HTTP requests, measures processing time, and logs audit information.

This method records the start time, processes the request, and calculates the duration. It extracts user information (ID, username, access level), request metadata (IP, method, path, user agent), and both request and response data (optionally compressed and base64-encoded, except for 'openapi.json' endpoints). All relevant details are logged using the audit_logger for auditing purposes.

Parameters:

Name	Type	Description	Default
request	HttpRequest	The incoming HTTP request object.	required

Returns:

Name	Type	Description
response	Any	The HTTP response object generated by processing the request.

Source code in onconova/core/history/middleware.py

def __call__(self, request: HttpRequest):
    """
    Handles incoming HTTP requests, measures processing time, and logs audit information.

    This method records the start time, processes the request, and calculates the duration.
    It extracts user information (ID, username, access level), request metadata (IP, method, path, user agent),
    and both request and response data (optionally compressed and base64-encoded, except for 'openapi.json' endpoints).
    All relevant details are logged using the audit_logger for auditing purposes.

    Args:
        request (HttpRequest): The incoming HTTP request object.

    Returns:
        response (Any): The HTTP response object generated by processing the request.
    """
    start_time = time.time()
    response = self.get_response(request)
    processing_time = round((time.time() - start_time) * 1000)

    user = getattr(request, "user", None)
    user_id = str(user.id) if user and user.is_authenticated else "anonymous"
    user_access_level = (
        int(user.access_level) if user and user.is_authenticated else -1
    )
    username = str(user.username) if user and user.is_authenticated else "anonymous"
    endpoint = request.get_full_path()
    audit_logger.info(
        "",
        extra={
            "user_id": user_id,
            "username": username,
            "access_level": user_access_level,
            "ip": self.get_client_ip(request),
            "method": request.method,
            "duration": processing_time,
            "path": endpoint,
            "status_code": response.status_code,
            "user_agent": request.META.get("HTTP_USER_AGENT", "")[:100],
            "request_data": (
                self.compress_b64(self.get_request_data(request))
                if "openapi.json" not in endpoint
                else "[openapi.json]"
            ),
            "response_data": (
                self.compress_b64(self.get_response_data(response))
                if "openapi.json" not in endpoint
                else "[openapi.json]"
            ),
        },
    )
    return response

compress_b64(data) staticmethod

Compresses a string using gzip and encodes the result in base64.

Parameters:

Name	Type	Description	Default
data	str	The input string to be compressed and encoded.	required

Returns:

Name	Type	Description
str	str	The base64-encoded compressed string, or "[compress-error]" if compression fails.

Source code in onconova/core/history/middleware.py

@staticmethod
def compress_b64(data: str) -> str:
    """
    Compresses a string using gzip and encodes the result in base64.

    Args:
        data (str): The input string to be compressed and encoded.

    Returns:
        str: The base64-encoded compressed string, or "[compress-error]" if compression fails.
    """
    try:
        compressed = gzip.compress(bytes(data, "utf-8"))
        return base64.b64encode(compressed).decode("utf-8")
    except Exception:
        return "[compress-error]"

get_client_ip(request) staticmethod

Retrieve the client's IP address from the given Django request object.

This function checks for the 'HTTP_X_FORWARDED_FOR' header, which is set by proxies to indicate the original IP address of the client. If present, it returns the first IP address in the list. If not present, it falls back to the 'REMOTE_ADDR' value, which contains the direct IP address of the client.

Parameters:

Name	Type	Description	Default
request	HttpRequest	The Django request object.	required

Returns:

Type	Description
str | None	The client's IP address as a string, or None if not found.

Source code in onconova/core/history/middleware.py

@staticmethod
def get_client_ip(request):
    """
    Retrieve the client's IP address from the given Django request object.

    This function checks for the 'HTTP_X_FORWARDED_FOR' header, which is set by proxies to indicate the original IP address of the client. If present, it returns the first IP address in the list. If not present, it falls back to the 'REMOTE_ADDR' value, which contains the direct IP address of the client.

    Args:
        request (HttpRequest): The Django request object.

    Returns:
        (str | None): The client's IP address as a string, or None if not found.
    """
    x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
    return (
        x_forwarded_for.split(",")[0]
        if x_forwarded_for
        else request.META.get("REMOTE_ADDR")
    )

get_request_data(request)

Extracts and redacts request data for auditing purposes.

For POST, PUT, and PATCH requests, attempts to decode and parse the request body as JSON, then redacts sensitive information and returns the result as a compact JSON string. For other request methods, redacts sensitive information from query parameters and returns the result as a compact JSON string.

In case of any exception during processing, logs the exception and returns "[unreadable]".

Parameters:

Name	Type	Description	Default
request	HttpRequest	The HTTP request object.	required

Returns:

Type	Description
str	A compact JSON string of the redacted request data, or "[unreadable]" if an error occurs.

Source code in onconova/core/history/middleware.py

def get_request_data(self, request: HttpRequest):
    """
    Extracts and redacts request data for auditing purposes.

    For POST, PUT, and PATCH requests, attempts to decode and parse the request body as JSON,
    then redacts sensitive information and returns the result as a compact JSON string.
    For other request methods, redacts sensitive information from query parameters and returns
    the result as a compact JSON string.

    In case of any exception during processing, logs the exception and returns "[unreadable]".

    Args:
        request (HttpRequest): The HTTP request object.

    Returns:
        (str): A compact JSON string of the redacted request data, or "[unreadable]" if an error occurs.
    """
    try:
        if request.method in ["POST", "PUT", "PATCH"]:
            if request.body:
                body = request.body.decode("utf-8")
                data = json.loads(body)
                return json.dumps(self.redact(data), separators=(",", ":"))
        return json.dumps(self.redact(request.GET.dict()), separators=(",", ":"))
    except Exception as e:
        audit_logger.exception(e)
        return "[unreadable]"

get_response_data(response)

Extracts and returns the JSON-encoded data from a response object.

If the response has a 'data' attribute, it redacts sensitive information and serializes it to a compact JSON string. If the response is a JSON HTTP response, it decodes and returns the content as a string. For non-JSON responses, returns a placeholder string. In case of any exception, logs the error and returns an unreadable placeholder.

Parameters:

Name	Type	Description	Default
response	Any	The response object to extract data from.	required

Returns:

Type	Description
str	The JSON string representation of the response data, a placeholder for non-JSON responses, or an unreadable placeholder in case of errors.

Source code in onconova/core/history/middleware.py

def get_response_data(self, response):
    """
    Extracts and returns the JSON-encoded data from a response object.

    If the response has a 'data' attribute, it redacts sensitive information and serializes it to a compact JSON string.
    If the response is a JSON HTTP response, it decodes and returns the content as a string.
    For non-JSON responses, returns a placeholder string.
    In case of any exception, logs the error and returns an unreadable placeholder.

    Args:
        response (Any): The response object to extract data from.

    Returns:
        (str): The JSON string representation of the response data, a placeholder for non-JSON responses,
             or an unreadable placeholder in case of errors.
    """
    try:
        if hasattr(response, "data"):
            return json.dumps(self.redact(response.data), separators=(",", ":"))
        content_type = response.get("Content-Type", "")
        if "application/json" in content_type:
            return response.content.decode("utf-8")
        return "[non-json-response]"
    except Exception as e:
        audit_logger.exception(e)
        return "[unreadable]"

redact(data) staticmethod

Recursively redacts sensitive information from dictionaries and lists.

Parameters:

Name	Type	Description	Default
data	object	The input data to be redacted. Can be a dictionary, list, or other object.	required

Returns:

Name	Type	Description
object	object	The redacted data with sensitive values replaced by "[REDACTED]".

Notes

• Keys in dictionaries that match any string in SENSITIVE_KEYS (case-insensitive) will have their values replaced.
• Lists are processed recursively.
• Non-dict and non-list objects are returned unchanged.

Source code in onconova/core/history/middleware.py

@staticmethod
def redact(data) -> object:
    """
    Recursively redacts sensitive information from dictionaries and lists.

    Args:
        data (object): The input data to be redacted. Can be a dictionary, list, or other object.

    Returns:
        object: The redacted data with sensitive values replaced by "[REDACTED]".

    Notes:
        - Keys in dictionaries that match any string in SENSITIVE_KEYS (case-insensitive) will have their values replaced.
        - Lists are processed recursively.
        - Non-dict and non-list objects are returned unchanged.
    """
    if isinstance(data, dict):
        redacted = {}
        for k, v in data.items():
            if any(s in k.lower() for s in SENSITIVE_KEYS):
                redacted[k] = "[REDACTED]"
            else:
                redacted[k] = AuditLogMiddleware.redact(v)
        return redacted
    elif isinstance(data, list):
        return [AuditLogMiddleware.redact(item) for item in data]
    else:
        return data

DjangoRequest

Although Django's auth middleware sets the user in middleware, apps like django-rest-framework set the user in the view layer. This creates issues for pghistory tracking since the context needs to be set before DB operations happen.

This special WSGIRequest updates pghistory context when the request.user attribute is updated.

__setattr__(attr, value)

Source code in onconova/core/history/middleware.py

def __setattr__(self, attr, value):
    if attr == "user":
        user = (
            str(value._meta.pk.get_db_prep_value(value.pk, connection))
            if value and hasattr(value, "_meta")
            else None
        )
        username = value.username if hasattr(value, "username") else None
        pghistory.context(username=username, user=user)
    return super().__setattr__(attr, value)

HistoryMiddleware

Bases: HistoryMiddleware

Custom middleware for tracking request history with additional context.

This middleware extends pghistory.middleware.HistoryMiddleware to enrich the history context with the requesting user's username and IP address. It overrides the get_context method to add these details to the context used for history tracking.

__call__(request)

Source code in onconova/core/history/middleware.py

def __call__(self, request):
    if request.method in config.middleware_methods():
        with pghistory.context(**self.get_context(request)):
            if isinstance(request, DjangoWSGIRequest):  # pragma: no branch
                request.__class__ = WSGIRequest
            elif isinstance(request, DjangoASGIRequest):  # pragma: no cover
                request.__class__ = ASGIRequest

            return self.get_response(request)
    else:
        return self.get_response(request)

get_context(request)

Returns a context dictionary for the given request, extending the parent context with additional information: - 'username': The username of the authenticated user, or None if unavailable. - 'ip_address': The IP address of the request, or 'unknown' if not found.

Parameters:

Name	Type	Description	Default
request	HttpRequest	The HTTP request object.	required

Returns:

Type	Description
dict	The context dictionary containing user and request metadata.

Source code in onconova/core/history/middleware.py

def get_context(self, request: HttpRequest):
    """
    Returns a context dictionary for the given request, extending the parent context
    with additional information:
        - 'username': The username of the authenticated user, or None if unavailable.
        - 'ip_address': The IP address of the request, or 'unknown' if not found.

    Args:
        request (HttpRequest): The HTTP request object.

    Returns:
        (dict): The context dictionary containing user and request metadata.
    """
    return super().get_context(request) | {
        "username": (
            request.user.username if hasattr(request.user, "username") else None
        ),
        "ip_address": request.META.get("REMOTE_ADDR", "unknown"),
    }

WSGIRequest

Bases: DjangoRequest, WSGIRequest

WSGIRequest is a subclass that combines functionality from both DjangoRequest and DjangoWSGIRequest.

This class is intended to represent an HTTP request in a WSGI-compliant Django application, inheriting all attributes and methods from its parent classes.

Inheritance

DjangoRequest: Provides core request handling features. DjangoWSGIRequest: Adds WSGI-specific request capabilities.

No additional attributes or methods are defined in this subclass.

2025-10-172025-10-17runner