bottleneckcrm/backend/common/tests/test_audit_log.py
Ashwin 5b6bc01687 feat(audit): enhance visitor tracking with user agent and IP address
- Updated the audit log to record the user's IP address and user agent from relayed requests.
- Introduced a new `user_agent` function in `request_meta.py` to handle user agent extraction.
- Modified the `AuditLogger` to utilize the new `user_agent` function.
- Added tests to ensure correct logging of relayed visitor information in `test_audit_log.py`.
- Updated various views and tests to accommodate the new user agent handling, including `public_views.py`, `test_invoices_api.py`, and frontend hooks.
- Enhanced relay headers to include user agent information for better tracking.
- Updated documentation and comments to reflect changes in user agent handling.
2026-10-02 08:32:51 +05:30

127 lines
5.1 KiB
Python

"""
Tests for common/audit_log.py - AuditLogger methods not yet covered.
Covers:
- _get_request_info with X-Forwarded-For header
- logout
- cross_org_attempt
- api_key_used
- api_key_invalid
- membership_revoked
- suspicious_activity
Run with: pytest common/tests/test_audit_log.py -v
"""
import pytest
from django.test import RequestFactory
from common.audit_log import AuditLogger, SecurityAuditLog
@pytest.mark.django_db
class TestAuditLoggerMethods:
"""Test AuditLogger methods that lack coverage."""
def setup_method(self):
self.logger = AuditLogger()
self.factory = RequestFactory()
def test_get_request_info_with_forwarded_for(self):
"""The first X-Forwarded-For entry is the caller's to write, so it is
not recorded. With no proxy configured the socket peer is."""
request = self.factory.get(
"/api/test/",
HTTP_X_FORWARDED_FOR="203.0.113.50, 70.41.3.18, 150.172.238.178",
HTTP_USER_AGENT="TestAgent/1.0",
REMOTE_ADDR="192.0.2.10",
)
info = self.logger._get_request_info(request)
assert info["ip_address"] == "192.0.2.10"
assert info["user_agent"] == "TestAgent/1.0"
assert info["request_path"] == "/api/test/"
assert info["request_method"] == "GET"
def test_get_request_info_records_the_relayed_visitor(self, settings):
"""A sign-in relayed by the web app records the visitor's address and
browser, not the web server's address and HTTP client."""
secret = "r" * 48
settings.RELAY_SECRET = secret
request = self.factory.post(
"/api/auth/magic-link/verify/",
HTTP_USER_AGENT="axios/1.20.0",
HTTP_X_BOTTLECRM_RELAY_SECRET=secret,
HTTP_X_BOTTLECRM_CLIENT_IP="203.0.113.50",
HTTP_X_BOTTLECRM_USER_AGENT="Mozilla/5.0 (relayed)",
REMOTE_ADDR="13.235.35.197",
)
info = self.logger._get_request_info(request)
assert info["ip_address"] == "203.0.113.50"
assert info["user_agent"] == "Mozilla/5.0 (relayed)"
def test_get_request_info_none_request(self):
"""None request should return empty dict."""
info = self.logger._get_request_info(None)
assert info == {}
def test_logout(self, admin_user, org_a):
request = self.factory.post("/api/auth/logout/")
self.logger.logout(admin_user, org_a, request)
log = SecurityAuditLog.objects.filter(event_type="LOGOUT").first()
assert log is not None
assert log.user == admin_user
assert log.org == org_a
assert "logged out" in log.description
def test_cross_org_attempt(self, admin_user, org_a, org_b):
request = self.factory.get("/api/accounts/")
self.logger.cross_org_attempt(admin_user, org_a, org_b, "Account:123", request)
log = SecurityAuditLog.objects.filter(event_type="CROSS_ORG_ATTEMPT").first()
assert log is not None
assert log.success is False
assert str(org_b.id) in log.metadata["target_org_id"]
def test_api_key_used(self, org_a):
request = self.factory.get("/api/leads/")
self.logger.api_key_used(org_a, "/api/leads/", request)
log = SecurityAuditLog.objects.filter(event_type="API_KEY_USED").first()
assert log is not None
assert log.org == org_a
assert log.metadata["endpoint"] == "/api/leads/"
def test_api_key_invalid(self):
request = self.factory.get("/api/leads/")
self.logger.api_key_invalid("abc123", request)
log = SecurityAuditLog.objects.filter(event_type="API_KEY_INVALID").first()
assert log is not None
assert log.success is False
assert "abc123" in log.metadata["api_key_prefix"]
def test_membership_revoked(self, admin_user, org_a, regular_user):
request = self.factory.post("/api/profiles/revoke/")
self.logger.membership_revoked(
admin_user, org_a, revoked_by=regular_user, request=request
)
log = SecurityAuditLog.objects.filter(event_type="MEMBERSHIP_REVOKED").first()
assert log is not None
assert log.user == admin_user
assert log.org == org_a
assert str(regular_user.id) in log.metadata["revoked_by"]
def test_membership_revoked_by_system(self, admin_user, org_a):
"""When revoked_by is None, description should say 'system'."""
self.logger.membership_revoked(admin_user, org_a)
log = SecurityAuditLog.objects.filter(event_type="MEMBERSHIP_REVOKED").first()
assert log is not None
assert "system" in log.description
def test_suspicious_activity(self, admin_user, org_a):
request = self.factory.get("/api/accounts/")
self.logger.suspicious_activity(
admin_user, org_a, "BRUTE_FORCE", "Multiple failed logins", request
)
log = SecurityAuditLog.objects.filter(event_type="SUSPICIOUS_ACTIVITY").first()
assert log is not None
assert log.success is False
assert log.metadata["activity_type"] == "BRUTE_FORCE"
assert log.metadata["details"] == "Multiple failed logins"