796 lines
26 KiB
Python
796 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import json
|
|
import logging
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from billiard.einfo import ExceptionInfo
|
|
from billiard.exceptions import WorkerLostError
|
|
from django.conf import settings
|
|
from django.http import HttpResponse
|
|
from django.test import RequestFactory
|
|
from django.test.utils import override_settings
|
|
|
|
from api.auth.authentication import SessionAwareJWTAuthentication
|
|
from eusrr_backend.celery import (
|
|
_normalize_failure_traceback,
|
|
inject_task_log_context,
|
|
log_task_completed,
|
|
log_task_failed,
|
|
log_task_started,
|
|
)
|
|
from eusrr_backend.channels_jwt import JWTAuthMiddleware
|
|
from eusrr_backend.logging_utils import (
|
|
ConsoleFormatter,
|
|
ContextFilter,
|
|
JsonFormatter,
|
|
WebSocketLoggingMiddleware,
|
|
bind_log_context,
|
|
get_log_context,
|
|
redact,
|
|
reset_log_context,
|
|
resolve_client_ip,
|
|
)
|
|
from eusrr_backend.middleware import RequestLoggingMiddleware
|
|
|
|
BACKEND_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _record(message="operation completed", **extra):
|
|
record = logging.LogRecord(
|
|
"test.logging",
|
|
logging.INFO,
|
|
__file__,
|
|
10,
|
|
message,
|
|
(),
|
|
None,
|
|
)
|
|
for key, value in extra.items():
|
|
setattr(record, key, value)
|
|
return record
|
|
|
|
|
|
def test_json_formatter_emits_context_and_structured_fields():
|
|
token = bind_log_context(request_id="req-123", user_id=42)
|
|
try:
|
|
output = JsonFormatter().format(
|
|
_record(
|
|
event="document.created",
|
|
document_id=7,
|
|
duration_ms=12.5,
|
|
)
|
|
)
|
|
finally:
|
|
reset_log_context(token)
|
|
|
|
payload = json.loads(output)
|
|
assert payload["event"] == "document.created"
|
|
assert payload["request_id"] == "req-123"
|
|
assert payload["user_id"] == 42
|
|
assert payload["document_id"] == 7
|
|
assert payload["duration_ms"] == 12.5
|
|
assert payload["source"].endswith(":10")
|
|
|
|
|
|
def test_logging_settings_use_one_context_aware_console_pipeline():
|
|
config = settings.LOGGING
|
|
|
|
assert config["root"]["handlers"] == ["console"]
|
|
assert config["handlers"]["console"]["filters"] == ["context"]
|
|
assert config["handlers"]["console"]["formatter"] in {"console", "json"}
|
|
assert "eusrr_backend.auth_backends" not in config["loggers"]
|
|
assert settings.CELERY_WORKER_HIJACK_ROOT_LOGGER is False
|
|
assert isinstance(settings.LOG_TRUST_X_FORWARDED_FOR, bool)
|
|
assert config["loggers"]["daphne"]["level"] == "WARNING"
|
|
assert config["loggers"]["daphne"]["propagate"] is False
|
|
assert config["loggers"]["django.channels.server"]["level"] == "WARNING"
|
|
assert config["loggers"]["django.channels.server"]["propagate"] is False
|
|
assert config["loggers"]["httpcore"]["level"] == "WARNING"
|
|
assert config["loggers"]["httpcore"]["propagate"] is False
|
|
assert config["loggers"]["httpx"]["level"] == "WARNING"
|
|
assert config["loggers"]["httpx"]["propagate"] is False
|
|
for logger_name in (
|
|
"uvicorn",
|
|
"uvicorn.error",
|
|
"uvicorn.access",
|
|
"gunicorn.access",
|
|
):
|
|
assert config["loggers"][logger_name]["handlers"] == ["console"]
|
|
assert config["loggers"][logger_name]["propagate"] is False
|
|
assert config["loggers"]["uvicorn.access"]["level"] == "WARNING"
|
|
assert config["loggers"]["gunicorn.access"]["level"] == "WARNING"
|
|
|
|
|
|
def test_context_filter_assigns_stable_event_to_framework_logs():
|
|
record = _record("transport warning")
|
|
record.name = "daphne.ws_protocol"
|
|
record.levelno = logging.WARNING
|
|
record.levelname = "WARNING"
|
|
|
|
assert ContextFilter().filter(record) is True
|
|
assert record.event == "framework.daphne_ws_protocol.warning"
|
|
|
|
|
|
def test_formatter_removes_terminal_ansi_sequences():
|
|
record = _record("\x1b[mHTTP GET /health/ 200\x1b[0m", event="server.access")
|
|
|
|
output = ConsoleFormatter().format(record)
|
|
|
|
assert "\\u001b" not in output
|
|
assert "\x1b" not in output
|
|
assert "HTTP GET /health/ 200" in output
|
|
|
|
|
|
def test_console_formatter_groups_http_context_and_event_data():
|
|
output = ConsoleFormatter().format(
|
|
_record(
|
|
"HTTP request completed",
|
|
event="http.request.completed",
|
|
http_method="GET",
|
|
path="/api/v1/tasks/boards/1/",
|
|
status_code=200,
|
|
duration_ms=27.5,
|
|
request_id="req-123",
|
|
user_id=42,
|
|
client_ip="127.0.0.1",
|
|
cache_hit=True,
|
|
)
|
|
)
|
|
|
|
assert "INFO [HTTP] http.request.completed @test.logging" in output
|
|
assert '[GET "/api/v1/tasks/boards/1/" → 200 27.5ms]' in output
|
|
assert '[req="req-123" user=42 ip="127.0.0.1"]' in output
|
|
assert "[cache_hit=true]" in output
|
|
assert '"HTTP request completed"' in output
|
|
assert "level=" not in output
|
|
assert "logger=" not in output
|
|
assert "event=" not in output
|
|
assert "status_code=" not in output
|
|
assert "duration_ms=" not in output
|
|
assert "message=" not in output
|
|
assert "\x1b" not in output
|
|
assert output.index("[GET ") < output.index("[req=")
|
|
assert output.index("[req=") < output.index("[cache_hit=")
|
|
|
|
|
|
def test_console_formatter_does_not_repeat_websocket_correlation_id():
|
|
connection_id = "connection-123"
|
|
output = ConsoleFormatter().format(
|
|
_record(
|
|
"WebSocket connection accepted",
|
|
event="websocket.connection.accepted",
|
|
connection_id=connection_id,
|
|
request_id=connection_id,
|
|
path="/ws/",
|
|
user_id=42,
|
|
)
|
|
)
|
|
|
|
assert output.count(connection_id) == 1
|
|
assert '[conn="connection-123" path="/ws/"]' in output
|
|
assert "[user=42]" in output
|
|
|
|
|
|
@pytest.mark.parametrize("formatter", [ConsoleFormatter(), JsonFormatter()])
|
|
def test_formatters_redact_secrets_and_escape_newlines(formatter):
|
|
output = formatter.format(
|
|
_record(
|
|
"token=top-secret Authorization: Bearer jwt-value\nsecond line",
|
|
event="security.test",
|
|
password="plain-password",
|
|
cookie="session=abc",
|
|
)
|
|
)
|
|
|
|
assert "top-secret" not in output
|
|
assert "jwt-value" not in output
|
|
assert "plain-password" not in output
|
|
assert "session=abc" not in output
|
|
assert "\nsecond line" not in output
|
|
assert "\\nsecond line" in output
|
|
|
|
|
|
@pytest.mark.parametrize("formatter", [ConsoleFormatter(), JsonFormatter()])
|
|
def test_formatters_redact_personal_directory_data_and_bound_values(formatter):
|
|
output = formatter.format(
|
|
_record(
|
|
"employee=CN=Private Person,OU=Users,DC=example,DC=test "
|
|
"phone=+7 (999) 123-45-67 " + "x" * 3000,
|
|
event="security.directory_data",
|
|
)
|
|
)
|
|
|
|
assert "Private Person" not in output
|
|
assert "999" not in output
|
|
assert "[REDACTED_DN]" in output
|
|
assert "[REDACTED_PHONE]" in output
|
|
assert "[TRUNCATED]" in output
|
|
|
|
|
|
@pytest.mark.parametrize("formatter", [ConsoleFormatter(), JsonFormatter()])
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
[
|
|
"/ws/?ticket=websocket-secret&other=keep",
|
|
"/callback/?code=oauth-code-secret&state=oauth-state-secret&other=keep",
|
|
"/ws/?%74%69cket=websocket-secret&access_token=access-secret&other=keep",
|
|
"/ws/?ticket=websocket-secret&ticket=second-secret&other=keep",
|
|
],
|
|
)
|
|
def test_access_messages_redact_authentication_query_parameters(formatter, url):
|
|
record = _record('127.0.0.1 - "GET %s HTTP/1.1" 403', event="server.access")
|
|
record.args = (url,)
|
|
output = formatter.format(record)
|
|
assert "secret" not in output
|
|
assert "other=keep" in output
|
|
assert "[REDACTED]" in output
|
|
|
|
|
|
def test_redaction_covers_structured_and_quoted_tickets():
|
|
assert redact({"ticket": "private-ticket", "status_code": 403}) == {
|
|
"ticket": "[REDACTED]",
|
|
"status_code": 403,
|
|
}
|
|
assert "private-ticket" not in redact('{"ticket": "private-ticket"}')
|
|
assert "multi word secret" not in redact("password='multi word secret'")
|
|
|
|
|
|
def test_redaction_preserves_correlation_ids_containing_phone_shaped_digits():
|
|
request_id = "d24e55ebc077455b89991234567e7e"
|
|
payload = json.loads(
|
|
JsonFormatter().format(_record(f"request={request_id}", request_id=request_id))
|
|
)
|
|
assert payload["request_id"] == request_id
|
|
assert request_id in payload["message"]
|
|
assert redact("89991234567", key="request_id") == "89991234567"
|
|
assert redact("phone 89991234567") == "phone [REDACTED_PHONE]"
|
|
|
|
|
|
def test_long_exception_keeps_the_terminal_cause_and_redacts_the_whole_traceback():
|
|
try:
|
|
raise RuntimeError(
|
|
"ticket=hidden-start " + "x" * 24000 + " token=hidden-end terminal-cause"
|
|
)
|
|
except RuntimeError:
|
|
record = _record(event="operation.failed")
|
|
record.exc_info = sys.exc_info()
|
|
exception = json.loads(JsonFormatter().format(record))["exception"]
|
|
assert len(exception) <= 16384
|
|
assert "Traceback" in exception
|
|
assert "[TRUNCATED]" in exception
|
|
assert exception.endswith("terminal-cause")
|
|
assert "hidden-start" not in exception
|
|
assert "hidden-end" not in exception
|
|
|
|
|
|
def test_normal_exception_is_not_cut_at_the_old_2048_character_limit():
|
|
try:
|
|
raise RuntimeError("x" * 3000 + " final-exception-cause")
|
|
except RuntimeError:
|
|
record = _record(event="operation.failed")
|
|
record.exc_info = sys.exc_info()
|
|
exception = json.loads(JsonFormatter().format(record))["exception"]
|
|
assert len(exception) > 3000
|
|
assert "[TRUNCATED]" not in exception
|
|
assert exception.endswith("final-exception-cause")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"peer,forwarded,expected",
|
|
[
|
|
("10.0.0.5", "198.51.100.99, 203.0.113.10, 10.0.0.6", "203.0.113.10"),
|
|
("203.0.113.10", "198.51.100.99", "203.0.113.10"),
|
|
("10.0.0.5", "bad-value, 203.0.113.10", "203.0.113.10"),
|
|
("10.0.0.5", "203.0.113.10, bad-value", "10.0.0.5"),
|
|
("10.0.0.5", "", "10.0.0.5"),
|
|
("invalid", "203.0.113.10", None),
|
|
],
|
|
)
|
|
def test_proxy_chain_stops_at_the_first_untrusted_hop(peer, forwarded, expected):
|
|
assert (
|
|
resolve_client_ip(
|
|
peer,
|
|
forwarded_for=forwarded,
|
|
trust_proxy=True,
|
|
trusted_proxies=("10.0.0.0/24",),
|
|
)
|
|
== expected
|
|
)
|
|
|
|
|
|
def test_proxy_trust_requires_both_opt_in_and_an_allowlisted_peer():
|
|
assert (
|
|
resolve_client_ip("10.0.0.5", forwarded_for="203.0.113.10", trust_proxy=True)
|
|
== "10.0.0.5"
|
|
)
|
|
assert (
|
|
resolve_client_ip(
|
|
"10.0.0.5", forwarded_for="203.0.113.10", trusted_proxies=("10.0.0.0/24",)
|
|
)
|
|
== "10.0.0.5"
|
|
)
|
|
|
|
|
|
def test_formatter_serializes_exception_without_losing_event():
|
|
try:
|
|
raise ValueError("broken")
|
|
except ValueError:
|
|
record = _record(event="operation.failed")
|
|
record.exc_info = __import__("sys").exc_info()
|
|
|
|
payload = json.loads(JsonFormatter().format(record))
|
|
assert payload["event"] == "operation.failed"
|
|
assert "ValueError" in payload["exception"]
|
|
|
|
|
|
def test_worker_lost_string_traceback_does_not_break_celery_logging(monkeypatch):
|
|
captured = {}
|
|
|
|
def capture_error(message, **kwargs):
|
|
captured.update(message=message, **kwargs)
|
|
|
|
monkeypatch.setattr("eusrr_backend.celery.logger.error", capture_error)
|
|
failure = WorkerLostError("worker exited with SIGTERM")
|
|
|
|
log_task_failed(
|
|
sender=SimpleNamespace(name="commerce.sync_connection_run"),
|
|
task_id="lost-task",
|
|
exception=failure,
|
|
traceback="serialized billiard traceback",
|
|
)
|
|
|
|
assert captured["exc_info"] is None
|
|
assert captured["extra"]["failure_traceback"] == ("serialized billiard traceback")
|
|
record = _record(**captured["extra"])
|
|
record.exc_info = captured["exc_info"]
|
|
assert "WorkerLostError" in JsonFormatter().format(record)
|
|
|
|
|
|
def test_billiard_exception_info_is_normalized_to_real_traceback():
|
|
try:
|
|
raise RuntimeError("original worker failure")
|
|
except RuntimeError as error:
|
|
billiard_info = ExceptionInfo()
|
|
exc_info, fallback = _normalize_failure_traceback(error, billiard_info)
|
|
|
|
assert exc_info is not None
|
|
assert exc_info[2].tb_frame is not None
|
|
assert fallback
|
|
record = _record(event="celery.task.failed")
|
|
record.exc_info = exc_info
|
|
payload = json.loads(JsonFormatter().format(record))
|
|
assert "original worker failure" in payload["exception"]
|
|
|
|
|
|
@pytest.mark.parametrize("formatter", [ConsoleFormatter(), JsonFormatter()])
|
|
def test_formatter_survives_malformed_string_exc_info(formatter):
|
|
failure = WorkerLostError("worker disappeared")
|
|
record = _record(event="celery.task.failed")
|
|
record.exc_info = (WorkerLostError, failure, "not-a-traceback-object")
|
|
|
|
output = formatter.format(record)
|
|
|
|
assert "worker disappeared" in output
|
|
|
|
|
|
def test_formatter_sanitizes_django_request_extra():
|
|
request = RequestFactory().get(
|
|
"/api/v1/documents/?token=query-secret&search=private"
|
|
)
|
|
record = _record(event="django.request.failed")
|
|
record.request = request
|
|
|
|
payload = json.loads(JsonFormatter().format(record))
|
|
|
|
assert payload["request"] == {
|
|
"method": "GET",
|
|
"path": "/api/v1/documents/",
|
|
}
|
|
assert "query-secret" not in json.dumps(payload)
|
|
assert "private" not in json.dumps(payload)
|
|
|
|
|
|
def test_http_middleware_accepts_safe_request_id_and_cleans_context(monkeypatch):
|
|
contexts = []
|
|
log_calls = []
|
|
|
|
def get_response(request):
|
|
contexts.append(get_log_context())
|
|
return HttpResponse("ok", status=200)
|
|
|
|
monkeypatch.setattr(
|
|
"eusrr_backend.middleware.logger.info",
|
|
lambda message, **kwargs: log_calls.append((message, kwargs)),
|
|
)
|
|
request = RequestFactory().get(
|
|
"/api/v1/documents/?token=must-not-leak",
|
|
HTTP_X_REQUEST_ID="trace-123",
|
|
REMOTE_ADDR="127.0.0.7",
|
|
)
|
|
request.user = SimpleNamespace(is_authenticated=True, pk=42)
|
|
|
|
response = RequestLoggingMiddleware(get_response)(request)
|
|
|
|
assert response["X-Request-ID"] == "trace-123"
|
|
assert contexts == [
|
|
{
|
|
"request_id": "trace-123",
|
|
"user_id": 42,
|
|
"http_method": "GET",
|
|
"path": "/api/v1/documents/",
|
|
"client_ip": "127.0.0.7",
|
|
}
|
|
]
|
|
assert get_log_context() == {}
|
|
assert log_calls[0][1]["extra"]["event"] == "http.request.completed"
|
|
assert log_calls[0][1]["extra"]["status_code"] == 200
|
|
|
|
|
|
def test_http_middleware_replaces_unsafe_request_id():
|
|
request = RequestFactory().get(
|
|
"/health/", HTTP_X_REQUEST_ID="bad id\nAuthorization: secret"
|
|
)
|
|
request.user = SimpleNamespace(is_authenticated=False)
|
|
|
|
response = RequestLoggingMiddleware(lambda request: HttpResponse("ok"))(request)
|
|
|
|
assert response["X-Request-ID"] != request.headers["X-Request-ID"]
|
|
assert len(response["X-Request-ID"]) == 32
|
|
assert get_log_context() == {}
|
|
|
|
|
|
@override_settings(
|
|
LOG_TRUST_X_FORWARDED_FOR=True, LOG_TRUSTED_PROXY_CIDRS=("10.0.0.0/24",)
|
|
)
|
|
def test_http_middleware_uses_valid_trusted_proxy_ip():
|
|
contexts = []
|
|
request = RequestFactory().get(
|
|
"/health/",
|
|
HTTP_X_FORWARDED_FOR="203.0.113.10, 10.0.0.5",
|
|
REMOTE_ADDR="10.0.0.5",
|
|
)
|
|
request.user = SimpleNamespace(is_authenticated=False)
|
|
|
|
RequestLoggingMiddleware(
|
|
lambda request: contexts.append(get_log_context()) or HttpResponse("ok")
|
|
)(request)
|
|
|
|
assert contexts[0]["client_ip"] == "203.0.113.10"
|
|
assert get_log_context() == {}
|
|
|
|
|
|
@override_settings(
|
|
LOG_TRUST_X_FORWARDED_FOR=True, LOG_TRUSTED_PROXY_CIDRS=("192.0.2.20/32",)
|
|
)
|
|
def test_http_middleware_rejects_invalid_forwarded_ip():
|
|
contexts = []
|
|
request = RequestFactory().get(
|
|
"/health/",
|
|
HTTP_X_FORWARDED_FOR="attacker-controlled-value",
|
|
REMOTE_ADDR="192.0.2.20",
|
|
)
|
|
request.user = SimpleNamespace(is_authenticated=False)
|
|
|
|
RequestLoggingMiddleware(
|
|
lambda request: contexts.append(get_log_context()) or HttpResponse("ok")
|
|
)(request)
|
|
|
|
assert contexts[0]["client_ip"] == "192.0.2.20"
|
|
|
|
|
|
def test_http_jwt_authentication_binds_user_before_view_logging(monkeypatch):
|
|
user = SimpleNamespace(pk=73)
|
|
validated_token = object()
|
|
monkeypatch.setattr(
|
|
"rest_framework_simplejwt.authentication.JWTAuthentication.authenticate",
|
|
lambda authentication, request: (user, validated_token),
|
|
)
|
|
monkeypatch.setattr(
|
|
SessionAwareJWTAuthentication,
|
|
"validate_session",
|
|
lambda authentication, token, user, request=None: None,
|
|
)
|
|
token = bind_log_context(client_ip="203.0.113.11")
|
|
try:
|
|
result = SessionAwareJWTAuthentication().authenticate(SimpleNamespace())
|
|
assert result == (user, validated_token)
|
|
assert get_log_context() == {
|
|
"client_ip": "203.0.113.11",
|
|
"user_id": 73,
|
|
}
|
|
finally:
|
|
reset_log_context(token)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_websocket_middleware_logs_lifecycle_without_query_string(caplog):
|
|
sent = []
|
|
|
|
async def app(scope, receive, send):
|
|
assert get_log_context()["path"] == "/ws/notifications/"
|
|
assert "query_string" not in get_log_context()
|
|
await send({"type": "websocket.accept"})
|
|
await receive()
|
|
|
|
messages = iter([{"type": "websocket.disconnect", "code": 1000}])
|
|
|
|
async def receive():
|
|
return next(messages)
|
|
|
|
async def send(message):
|
|
sent.append(message)
|
|
|
|
scope = {
|
|
"type": "websocket",
|
|
"path": "/ws/notifications/",
|
|
"query_string": b"token=must-not-leak",
|
|
"headers": [(b"x-request-id", b"ws-123")],
|
|
"client": ("127.0.0.8", 10000),
|
|
}
|
|
with caplog.at_level(logging.INFO, logger="eusrr.websocket"):
|
|
await WebSocketLoggingMiddleware(app)(scope, receive, send)
|
|
|
|
assert scope["request_id"] == "ws-123"
|
|
assert [record.event for record in caplog.records] == [
|
|
"websocket.connection.accepted",
|
|
"websocket.connection.closed",
|
|
]
|
|
assert caplog.records[-1].close_code == 1000
|
|
assert get_log_context() == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@override_settings(
|
|
LOG_TRUST_X_FORWARDED_FOR=True, LOG_TRUSTED_PROXY_CIDRS=("10.0.0.0/24",)
|
|
)
|
|
async def test_websocket_middleware_uses_valid_trusted_proxy_ip():
|
|
async def app(scope, receive, send):
|
|
assert get_log_context()["client_ip"] == "2001:db8::7"
|
|
|
|
scope = {
|
|
"type": "websocket",
|
|
"path": "/ws/notifications/",
|
|
"headers": [(b"x-forwarded-for", b"2001:db8::7, 10.0.0.5")],
|
|
"client": ("10.0.0.5", 10000),
|
|
}
|
|
|
|
await WebSocketLoggingMiddleware(app)(scope, None, None)
|
|
assert get_log_context() == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_websocket_session_authentication_binds_user_context():
|
|
async def app(scope, receive, send):
|
|
assert get_log_context()["user_id"] == 91
|
|
|
|
token = bind_log_context(client_ip="203.0.113.12")
|
|
try:
|
|
await JWTAuthMiddleware(app)(
|
|
{
|
|
"type": "websocket",
|
|
"headers": [],
|
|
"query_string": b"",
|
|
"user": SimpleNamespace(is_authenticated=True, pk=91),
|
|
},
|
|
None,
|
|
None,
|
|
)
|
|
finally:
|
|
reset_log_context(token)
|
|
|
|
|
|
def test_celery_lifecycle_has_task_context_and_duration(caplog):
|
|
sender = SimpleNamespace(name="documents.tasks.index")
|
|
|
|
with caplog.at_level(logging.INFO, logger="eusrr.celery"):
|
|
log_task_started(sender=sender, task_id="task-123")
|
|
log_task_completed(
|
|
sender=sender,
|
|
task_id="task-123",
|
|
state="SUCCESS",
|
|
)
|
|
|
|
assert [record.event for record in caplog.records] == [
|
|
"celery.task.started",
|
|
"celery.task.completed",
|
|
]
|
|
assert caplog.records[0].task_id == "task-123"
|
|
assert caplog.records[-1].task_name == "documents.tasks.index"
|
|
assert caplog.records[-1].duration_ms >= 0
|
|
assert get_log_context() == {}
|
|
|
|
|
|
def test_celery_propagates_safe_request_context_through_headers():
|
|
headers = {}
|
|
token = bind_log_context(
|
|
request_id="request-123",
|
|
user_id=42,
|
|
client_ip="203.0.113.15",
|
|
path="/must-not-propagate/",
|
|
)
|
|
try:
|
|
inject_task_log_context(headers=headers)
|
|
finally:
|
|
reset_log_context(token)
|
|
|
|
assert headers == {
|
|
"eusrr_log_context": {
|
|
"request_id": "request-123",
|
|
"user_id": 42,
|
|
"client_ip": "203.0.113.15",
|
|
}
|
|
}
|
|
|
|
sender = SimpleNamespace(
|
|
name="notifications.tasks.send",
|
|
request=SimpleNamespace(headers=headers),
|
|
)
|
|
log_task_started(sender=sender, task_id="task-context-1")
|
|
assert get_log_context() == {
|
|
"request_id": "request-123",
|
|
"user_id": 42,
|
|
"client_ip": "203.0.113.15",
|
|
"task_id": "task-context-1",
|
|
"task_name": "notifications.tasks.send",
|
|
}
|
|
log_task_completed(sender=sender, task_id="task-context-1", state="SUCCESS")
|
|
assert get_log_context() == {}
|
|
|
|
|
|
def test_celery_derives_user_id_from_named_positional_task_argument():
|
|
def run(user_id, frequency="daily"):
|
|
return None
|
|
|
|
sender = SimpleNamespace(
|
|
name="notifications.send_digest",
|
|
request=SimpleNamespace(headers={}),
|
|
run=run,
|
|
)
|
|
log_task_started(
|
|
sender=sender,
|
|
task_id="task-context-2",
|
|
args=(84,),
|
|
kwargs={"frequency": "weekly"},
|
|
)
|
|
assert get_log_context()["user_id"] == 84
|
|
log_task_completed(sender=sender, task_id="task-context-2", state="SUCCESS")
|
|
assert get_log_context() == {}
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_failed_authentication_does_not_log_identifier_password_or_hash(
|
|
caplog, user_factory
|
|
):
|
|
from eusrr_backend.auth_backends import EmailOrPhoneBackend
|
|
|
|
user = user_factory(email="private-person@example.test")
|
|
password_hash = user.password
|
|
|
|
with caplog.at_level(logging.WARNING, logger="eusrr_backend.auth_backends"):
|
|
result = EmailOrPhoneBackend().authenticate(
|
|
None,
|
|
username=user.email,
|
|
password="plain-secret-password",
|
|
)
|
|
|
|
assert result is None
|
|
output = JsonFormatter().format(caplog.records[-1])
|
|
assert caplog.records[-1].event == "authentication.local.denied"
|
|
assert caplog.records[-1].reason == "invalid_password"
|
|
assert "private-person@example.test" not in output
|
|
assert "plain-secret-password" not in output
|
|
assert password_hash not in output
|
|
|
|
|
|
def test_production_python_does_not_use_print_for_runtime_logging():
|
|
excluded_parts = {".venv", "docs", "migrations", "scripts", "tests"}
|
|
violations = []
|
|
|
|
for path in BACKEND_ROOT.rglob("*.py"):
|
|
if any(part in excluded_parts for part in path.parts):
|
|
continue
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
for node in ast.walk(tree):
|
|
if (
|
|
isinstance(node, ast.Call)
|
|
and isinstance(node.func, ast.Name)
|
|
and node.func.id == "print"
|
|
):
|
|
violations.append(f"{path.relative_to(BACKEND_ROOT)}:{node.lineno}")
|
|
|
|
assert violations == []
|
|
|
|
|
|
def test_structured_log_extras_do_not_overwrite_log_record_fields():
|
|
excluded_parts = {".venv", "docs", "migrations", "scripts", "tests"}
|
|
reserved = set(logging.makeLogRecord({}).__dict__) | {"message", "asctime"}
|
|
violations = []
|
|
|
|
for path in BACKEND_ROOT.rglob("*.py"):
|
|
if any(part in excluded_parts for part in path.parts):
|
|
continue
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
for node in ast.walk(tree):
|
|
if not (
|
|
isinstance(node, ast.Call)
|
|
and isinstance(node.func, ast.Attribute)
|
|
and node.func.attr
|
|
in {"debug", "info", "warning", "error", "exception", "critical", "log"}
|
|
):
|
|
continue
|
|
for keyword in node.keywords:
|
|
if keyword.arg != "extra" or not isinstance(keyword.value, ast.Dict):
|
|
continue
|
|
for key in keyword.value.keys:
|
|
if isinstance(key, ast.Constant) and key.value in reserved:
|
|
violations.append(
|
|
f"{path.relative_to(BACKEND_ROOT)}:{node.lineno}:{key.value}"
|
|
)
|
|
|
|
assert violations == []
|
|
|
|
|
|
def test_production_logger_calls_have_explicit_well_formed_events():
|
|
excluded_parts = {".venv", "docs", "migrations", "scripts", "tests"}
|
|
levels = {"debug", "info", "warning", "error", "exception", "critical", "log"}
|
|
event_pattern = re.compile(r"[a-z][a-z0-9_]*(?:\.[a-z0-9_]+)+")
|
|
violations = []
|
|
logger_call_count = 0
|
|
|
|
for path in BACKEND_ROOT.rglob("*.py"):
|
|
if any(part in excluded_parts for part in path.parts):
|
|
continue
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
for node in ast.walk(tree):
|
|
if not (
|
|
isinstance(node, ast.Call)
|
|
and isinstance(node.func, ast.Attribute)
|
|
and node.func.attr in levels
|
|
):
|
|
continue
|
|
receiver = node.func.value
|
|
receiver_name = (
|
|
receiver.id
|
|
if isinstance(receiver, ast.Name)
|
|
else receiver.attr
|
|
if isinstance(receiver, ast.Attribute)
|
|
else ""
|
|
)
|
|
if receiver_name != "logger" and not receiver_name.endswith("_logger"):
|
|
continue
|
|
|
|
logger_call_count += 1
|
|
event_value = None
|
|
for keyword in node.keywords:
|
|
if keyword.arg != "extra" or not isinstance(keyword.value, ast.Dict):
|
|
continue
|
|
for key, value in zip(keyword.value.keys, keyword.value.values):
|
|
if isinstance(key, ast.Constant) and key.value == "event":
|
|
event_value = value
|
|
break
|
|
|
|
if event_value is None:
|
|
violations.append(
|
|
f"{path.relative_to(BACKEND_ROOT)}:{node.lineno}:missing"
|
|
)
|
|
elif isinstance(event_value, ast.Constant):
|
|
if not isinstance(
|
|
event_value.value, str
|
|
) or not event_pattern.fullmatch(event_value.value):
|
|
violations.append(
|
|
f"{path.relative_to(BACKEND_ROOT)}:{node.lineno}:invalid"
|
|
)
|
|
elif not (isinstance(event_value, ast.Name) and event_value.id == "event"):
|
|
violations.append(
|
|
f"{path.relative_to(BACKEND_ROOT)}:{node.lineno}:dynamic"
|
|
)
|
|
|
|
assert logger_call_count >= 400
|
|
assert violations == []
|