fix(spp_api_v2): add payload masking and URL sanitization to outgoing API log

- Add groups restriction to url field and XML view to limit access to auditors
- Add _sanitize_url method to strip sensitive query parameters before logging
- Call _sanitize_url in log_call so stored URLs never contain tokens or keys
- Add tests covering _sanitize_url and end-to-end URL sanitization in log_call

Author: jeremi

spp_api_v2/models/api_outgoing_log.py

@@ -27,7 +27,8 @@ class ApiOutgoingLog(models.Model):
     url = fields.Char(
         required=True,
         index=True,
-        help="Full URL called",
+        groups="spp_api_v2.group_api_v2_auditor",
+        help="Full URL called (may contain sensitive query parameters)",
     )
 
     endpoint = fields.Char(

spp_api_v2/services/outgoing_api_log_service.py

@@ -3,6 +3,7 @@
 
 import json
 import logging
+from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
 
 import psycopg2
 
@@ -55,6 +56,26 @@ def __init__(
         self.service_code = service_code
         self.user_id = user_id or env.uid
 
+    # Sensitive key patterns matched case-insensitively
+    SENSITIVE_KEYS = frozenset(
+        {
+            "authorization",
+            "password",
+            "token",
+            "access_token",
+            "refresh_token",
+            "api_key",
+            "apikey",
+            "secret",
+            "client_secret",
+            "credential",
+            "credentials",
+            "private_key",
+        }
+    )
+
+    MASK_VALUE = "***MASKED***"
+
     def log_call(
         self,
         url: str,
@@ -76,15 +97,21 @@ def log_call(
         Logging failures never raise exceptions.
         """
         try:
-            # Truncate large payloads
-            truncated_request = self._truncate_payload(request_summary)
-            truncated_response = self._truncate_payload(response_summary)
-
+            # Sanitize URL, mask sensitive keys, then truncate large payloads
+            safe_url = self._sanitize_url(url)
+            masked_request = self._mask_sensitive_keys(request_summary)
+            masked_response = self._mask_sensitive_keys(response_summary)
+            truncated_request = self._truncate_payload(masked_request)
+            truncated_response = self._truncate_payload(masked_response)
+
+            # sudo() is intentional: log records must be created regardless of
+            # the calling user's permissions on spp.api.outgoing.log. The
+            # service is an internal component, not a user-facing API.
             return (
                 self.env["spp.api.outgoing.log"]
                 .sudo()
                 .log_call(
-                    url=url,
+                    url=safe_url,
                     endpoint=endpoint,
                     http_method=http_method,
                     request_summary=truncated_request,
@@ -107,6 +134,66 @@ def log_call(
             _logger.exception("Failed to log outgoing API call")
             return None
 
+    def _mask_sensitive_keys(self, payload):
+        """Recursively mask values of known sensitive keys in a payload.
+
+        Args:
+            payload: Dict payload to mask, or None.
+
+        Returns:
+            A new dict with sensitive values replaced by MASK_VALUE,
+            or None if input is None.
+        """
+        if payload is None:
+            return None
+
+        return self._mask_recursive(payload)
+
+    def _mask_recursive(self, obj):
+        """Walk a structure and mask sensitive dict keys."""
+        if isinstance(obj, dict):
+            return {
+                key: (self.MASK_VALUE if key.lower() in self.SENSITIVE_KEYS else self._mask_recursive(value))
+                for key, value in obj.items()
+            }
+        if isinstance(obj, list):
+            return [self._mask_recursive(item) for item in obj]
+        return obj
+
+    def _sanitize_url(self, url):
+        """Remove sensitive query parameters from a URL before logging.
+
+        Query parameters whose names match SENSITIVE_KEYS (case-insensitively)
+        are replaced with MASK_VALUE so that API keys or tokens embedded in
+        query strings are never persisted.
+
+        Args:
+            url: URL string to sanitize.
+
+        Returns:
+            Sanitized URL string with sensitive query parameters masked,
+            or the original value if it cannot be parsed as a URL.
+        """
+        if not url:
+            return url
+
+        try:
+            parsed = urlparse(url)
+        except Exception:
+            return url
+
+        if not parsed.query:
+            return url
+
+        params = parse_qs(parsed.query, keep_blank_values=True)
+        sanitized_params = {
+            key: ([self.MASK_VALUE] if key.lower() in self.SENSITIVE_KEYS else values) for key, values in params.items()
+        }
+
+        sanitized_query = urlencode(sanitized_params, doseq=True)
+        sanitized = parsed._replace(query=sanitized_query)
+        return urlunparse(sanitized)
+
     def _truncate_payload(self, payload, max_length=10000):
         """Truncate large payloads for DB storage.
 

spp_api_v2/tests/test_api_outgoing_log.py

@@ -266,8 +266,9 @@ def setUpClass(cls):
         )
 
     def test_auditor_can_read_sensitive_fields(self):
-        """User with auditor group can read request_summary, response_summary, error_detail"""
+        """User with auditor group can read url, request_summary, response_summary, error_detail"""
         log = self.log_record.with_user(self.auditor_user)
+        self.assertEqual(log.url, "https://crvs.example.org/api/registry/sync/search")
         self.assertTrue(log.request_summary)
         self.assertEqual(log.request_summary["message"]["national_id"], "123456")
         self.assertTrue(log.response_summary)
@@ -278,6 +279,8 @@ def test_auditor_can_read_sensitive_fields(self):
     def test_non_auditor_cannot_read_sensitive_fields(self):
         """User without auditor group gets AccessError for sensitive fields"""
         log = self.log_record.with_user(self.viewer_user)
+        with self.assertRaises(AccessError):
+            _ = log.url
         with self.assertRaises(AccessError):
             _ = log.request_summary
         with self.assertRaises(AccessError):
@@ -288,7 +291,6 @@ def test_non_auditor_cannot_read_sensitive_fields(self):
     def test_non_auditor_can_read_metadata_fields(self):
         """User without auditor group can still read non-sensitive metadata"""
         log = self.log_record.with_user(self.viewer_user)
-        self.assertEqual(log.url, "https://crvs.example.org/api/registry/sync/search")
         self.assertEqual(log.endpoint, "/registry/sync/search")
         self.assertEqual(log.http_method, "POST")
         self.assertEqual(log.status, "http_error")
@@ -299,8 +301,9 @@ def test_sensitive_fields_hidden_in_fields_get(self):
         fields_info = (
             self.env["spp.api.outgoing.log"]
             .with_user(self.viewer_user)
-            .fields_get(["request_summary", "response_summary", "error_detail"])
+            .fields_get(["url", "request_summary", "response_summary", "error_detail"])
         )
+        self.assertNotIn("url", fields_info)
         self.assertNotIn("request_summary", fields_info)
         self.assertNotIn("response_summary", fields_info)
         self.assertNotIn("error_detail", fields_info)

spp_api_v2/tests/test_outgoing_api_log_service.py

@@ -1,6 +1,8 @@
 # Part of OpenSPP. See LICENSE file for full copyright and licensing details.
 """Tests for OutgoingApiLogService"""
 
+import json
+
 from odoo.tests.common import TransactionCase
 
 from ..services.outgoing_api_log_service import OutgoingApiLogService
@@ -150,8 +152,6 @@ def test_truncate_payload_exact_boundary(self):
         )
 
         # Build a payload whose JSON serialization is exactly max_length
-        import json
-
         max_length = 50
         # {"k": "..."} — adjust value to hit exact length
         base = json.dumps({"k": ""})  # '{"k": ""}' = 10 chars
@@ -173,8 +173,6 @@ def test_truncate_payload_one_over_boundary(self):
             service_code="test",
         )
 
-        import json
-
         max_length = 50
         base = json.dumps({"k": ""})
         filler = "x" * (max_length - len(base) + 1)
@@ -186,3 +184,194 @@ def test_truncate_payload_one_over_boundary(self):
         self.assertTrue(result["_truncated"])
         self.assertEqual(result["_original_length"], max_length + 1)
         self.assertEqual(len(result["_preview"]), max_length)
+
+    def test_mask_sensitive_keys_top_level(self):
+        """_mask_sensitive_keys masks known sensitive keys at top level"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        payload = {
+            "Authorization": "Bearer secret-token-123",
+            "Content-Type": "application/json",
+            "data": "visible",
+        }
+
+        result = service._mask_sensitive_keys(payload)
+        self.assertEqual(result["Authorization"], "***MASKED***")
+        self.assertEqual(result["Content-Type"], "application/json")
+        self.assertEqual(result["data"], "visible")
+
+    def test_mask_sensitive_keys_nested(self):
+        """_mask_sensitive_keys masks sensitive keys in nested dicts"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        payload = {
+            "header": {
+                "authorization": "Bearer xyz",
+                "action": "search",
+            },
+            "body": {"password": "s3cret", "username": "admin"},
+        }
+
+        result = service._mask_sensitive_keys(payload)
+        self.assertEqual(result["header"]["authorization"], "***MASKED***")
+        self.assertEqual(result["header"]["action"], "search")
+        self.assertEqual(result["body"]["password"], "***MASKED***")
+        self.assertEqual(result["body"]["username"], "admin")
+
+    def test_mask_sensitive_keys_case_insensitive(self):
+        """_mask_sensitive_keys matches key names case-insensitively"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        payload = {
+            "API_KEY": "key123",
+            "api_key": "key456",
+            "Api_Key": "key789",
+        }
+
+        result = service._mask_sensitive_keys(payload)
+        for key in payload:
+            self.assertEqual(result[key], "***MASKED***")
+
+    def test_mask_sensitive_keys_various_sensitive_names(self):
+        """_mask_sensitive_keys masks all common sensitive key names"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        sensitive_keys = [
+            "authorization",
+            "password",
+            "token",
+            "access_token",
+            "refresh_token",
+            "api_key",
+            "apikey",
+            "secret",
+            "client_secret",
+            "credential",
+            "private_key",
+        ]
+
+        payload = {key: f"value_{key}" for key in sensitive_keys}
+        result = service._mask_sensitive_keys(payload)
+
+        for key in sensitive_keys:
+            self.assertEqual(
+                result[key],
+                "***MASKED***",
+                f"Key '{key}' should be masked",
+            )
+
+    def test_mask_sensitive_keys_preserves_non_dict_values(self):
+        """_mask_sensitive_keys handles lists and non-dict nested structures"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        payload = {
+            "items": [
+                {"password": "secret", "name": "item1"},
+                {"token": "abc", "name": "item2"},
+            ],
+            "count": 2,
+        }
+
+        result = service._mask_sensitive_keys(payload)
+        self.assertEqual(result["items"][0]["password"], "***MASKED***")
+        self.assertEqual(result["items"][0]["name"], "item1")
+        self.assertEqual(result["items"][1]["token"], "***MASKED***")
+        self.assertEqual(result["items"][1]["name"], "item2")
+        self.assertEqual(result["count"], 2)
+
+    def test_mask_sensitive_keys_none_input(self):
+        """_mask_sensitive_keys returns None for None input"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        self.assertIsNone(service._mask_sensitive_keys(None))
+
+    def test_mask_sensitive_keys_does_not_mutate_original(self):
+        """_mask_sensitive_keys returns a new dict without mutating the input"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        payload = {"password": "secret", "data": "visible"}
+        result = service._mask_sensitive_keys(payload)
+
+        self.assertEqual(payload["password"], "secret")
+        self.assertEqual(result["password"], "***MASKED***")
+
+    def test_log_call_masks_sensitive_keys_in_payloads(self):
+        """log_call applies masking to request and response payloads"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        result = service.log_call(
+            url="https://example.org/api/test",
+            request_summary={"authorization": "Bearer xyz", "action": "search"},
+            response_summary={"token": "resp-token", "status": "ok"},
+        )
+
+        self.assertTrue(result)
+        self.assertEqual(result.request_summary["authorization"], "***MASKED***")
+        self.assertEqual(result.request_summary["action"], "search")
+        self.assertEqual(result.response_summary["token"], "***MASKED***")
+        self.assertEqual(result.response_summary["status"], "ok")
+
+    def test_sanitize_url_no_sensitive_params(self):
+        """_sanitize_url leaves URLs without sensitive params unchanged"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        url = "https://example.org/api/search?q=hello&limit=10"
+        result = service._sanitize_url(url)
+        self.assertIn("q=hello", result)
+        self.assertIn("limit=10", result)
+
+    def test_sanitize_url_masks_sensitive_params(self):
+        """_sanitize_url replaces sensitive query parameter values with MASKED"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        url = "https://example.org/api/search?api_key=secret123&q=hello"
+        result = service._sanitize_url(url)
+        self.assertNotIn("secret123", result)
+        self.assertIn("***MASKED***", result)
+        self.assertIn("q=hello", result)
+
+    def test_sanitize_url_masks_multiple_sensitive_params(self):
+        """_sanitize_url masks all sensitive params in a single URL"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        url = "https://example.org/api?token=abc&api_key=xyz&page=1"
+        result = service._sanitize_url(url)
+        self.assertNotIn("abc", result)
+        self.assertNotIn("xyz", result)
+        self.assertIn("page=1", result)
+
+    def test_sanitize_url_case_insensitive_params(self):
+        """_sanitize_url matches sensitive param names case-insensitively"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        url = "https://example.org/api?API_KEY=secret&Access_Token=tok"
+        result = service._sanitize_url(url)
+        self.assertNotIn("secret", result)
+        self.assertNotIn("tok", result)
+
+    def test_sanitize_url_no_query_string(self):
+        """_sanitize_url returns URL unchanged when there is no query string"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        url = "https://example.org/api/search"
+        result = service._sanitize_url(url)
+        self.assertEqual(result, url)
+
+    def test_sanitize_url_none_input(self):
+        """_sanitize_url returns None for None input"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        self.assertIsNone(service._sanitize_url(None))
+
+    def test_log_call_sanitizes_url(self):
+        """log_call strips sensitive query parameters from URL before storing"""
+        service = OutgoingApiLogService(self.env, service_name="Test", service_code="test")
+
+        result = service.log_call(
+            url="https://example.org/api/search?api_key=supersecret&q=hello",
+        )
+
+        self.assertTrue(result)
+        self.assertNotIn("supersecret", result.url)
+        self.assertIn("***MASKED***", result.url)
+        self.assertIn("q=hello", result.url)