ctms/backend/tests/test_audit_access_context.py

import uuid

import pytest

from app.core.request_context import (
    RequestAuditContext,
    build_request_audit_context,
    reset_request_audit_context,
    set_request_audit_context,
)
from app.crud.audit import log_action


class FakeAsyncSession:
    def __init__(self) -> None:
        self.added = None
        self.flushed = False

    def add(self, value) -> None:
        self.added = value

    async def flush(self) -> None:
        self.flushed = True


class FakeClient:
    def __init__(self, host: str = "10.0.0.8") -> None:
        self.host = host
        self.port = 54321


class FakeUrl:
    scheme = "https"
    path = "/api/v1/audit"
    query = "token=secret-token&view=all&subject_name=Alice"


class FakeQueryParams:
    def __init__(self) -> None:
        self._items = [
            ("token", "secret-token"),
            ("view", "all"),
            ("subject_name", "Alice"),
        ]

    def __bool__(self) -> bool:
        return True

    def multi_items(self):
        return list(self._items)


class FakeRequest:
    def __init__(self, headers: dict[str, str], *, client_host: str = "10.0.0.8") -> None:
        self.headers = headers
        self.client = FakeClient(client_host)
        self.method = "GET"
        self.url = FakeUrl()
        self.query_params = FakeQueryParams()
        self.scope = {
            "http_version": "1.1",
            "scheme": "https",
            "path": "/api/v1/audit",
            "method": "GET",
        }


@pytest.mark.asyncio
async def test_audit_log_action_captures_request_access_context():
    context_token = set_request_audit_context(
        RequestAuditContext(
            client_ip="203.0.113.10",
            user_agent="curl/8.1.2",
            client_type="desktop",
            client_version="1.2.3",
            client_platform="macos",
            build_channel="release",
            build_commit="abcdef1",
        )
    )
    db = FakeAsyncSession()
    try:
        log = await log_action(
            db,
            study_id=uuid.uuid4(),
            entity_type="subject",
            entity_id=uuid.uuid4(),
            action="UPDATE_SUBJECT",
            detail=None,
            operator_id=uuid.uuid4(),
            operator_role="PM",
            auto_commit=False,
        )
    finally:
        reset_request_audit_context(context_token)

    assert db.added is log
    assert db.flushed is True
    assert log.client_ip == "203.0.113.10"
    assert log.user_agent == "curl/8.1.2"
    assert log.client_type == "desktop"
    assert log.client_version == "1.2.3"
    assert log.client_platform == "macos"
    assert log.build_channel == "release"
    assert log.build_commit == "abcdef1"


def test_request_audit_context_captures_sanitized_headers():
    context = build_request_audit_context(
        FakeRequest(
            {
                "user-agent": "Chrome",
                "referer": "https://ctms.local/audit?token=secret-token",
                "authorization": "Bearer secret-token",
                "cookie": "sid=secret",
                "x-ctms-client-source": "ctms-web",
                "x-ctms-client-type": "web",
                "x-ctms-client-version": "0.1.0",
                "x-request-id": "req-123",
                "x-custom-secret": "hidden",
                "x-random-debug": "skip-me",
            }
        )
    )

    assert context.user_agent == "Chrome"
    assert context.client_type == "web"
    assert context.client_ip == "10.0.0.8"
    assert uuid.UUID(context.request_id)
    assert context.request_headers == {
        "authorization": "[redacted]",
        "cookie": "[redacted]",
        "user-agent": "Chrome",
        "x-ctms-client-source": "ctms-web",
        "x-ctms-client-type": "web",
        "x-ctms-client-version": "0.1.0",
        "x-custom-secret": "[redacted]",
        "x-request-id": "req-123",
    }
    assert context.request_snapshot is not None
    assert context.request_snapshot["method"] == "GET"
    assert context.request_snapshot["request_id"] == context.request_id
    assert context.request_snapshot["path"] == "/api/v1/audit"
    assert context.request_snapshot["query_string"] == (
        "token=[redacted]&view=all&subject_name=[redacted]"
    )
    assert context.request_snapshot["query_params"] == [
        {"name": "token", "value": "[redacted]"},
        {"name": "view", "value": "all"},
        {"name": "subject_name", "value": "[redacted]"},
    ]
    assert context.request_snapshot["headers"]["authorization"] == "[redacted]"
    assert context.request_snapshot["headers"]["cookie"] == "[redacted]"
    assert context.request_snapshot["headers"]["x-ctms-client-source"] == "ctms-web"
    assert "referer" not in context.request_snapshot["headers"]
    assert "x-random-debug" not in context.request_snapshot["headers"]
    assert context.request_snapshot["client"] == {"host": "10.0.0.8", "port": 54321}
    assert context.request_snapshot["body"] == {
        "captured": False,
        "reason": "body_not_captured_by_audit_policy",
    }


def test_request_audit_context_prefers_explicit_ctms_client_source():
    context = build_request_audit_context(
        FakeRequest(
            {
                "x-ctms-client-source": "ctms-desktop",
                "x-ctms-client-type": "web",
            }
        )
    )

    assert context.client_type == "desktop"
    assert context.request_headers == {
        "x-ctms-client-source": "ctms-desktop",
        "x-ctms-client-type": "web",
    }


def test_request_audit_context_uses_proxy_observed_ip_instead_of_spoofed_forwarded_prefix(monkeypatch):
    monkeypatch.setattr("app.core.request_context.settings.TRUSTED_PROXY_CIDRS", "10.0.0.0/8")
    context = build_request_audit_context(
        FakeRequest(
            {
                "x-real-ip": "203.0.113.20",
                "x-forwarded-for": "1.2.3.4, 203.0.113.20",
            }
        )
    )

    assert context.client_ip == "203.0.113.20"


def test_request_audit_context_ignores_forwarded_headers_from_untrusted_peer(monkeypatch):
    monkeypatch.setattr("app.core.request_context.settings.TRUSTED_PROXY_CIDRS", "10.0.0.0/8")
    context = build_request_audit_context(
        FakeRequest(
            {
                "x-real-ip": "203.0.113.20",
                "x-forwarded-for": "1.2.3.4",
            },
            client_host="198.51.100.8",
        )
    )

    assert context.client_ip == "198.51.100.8"