From 028e4e71f83727f2f934daf48ae1c6c7cd5e2c37 Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Sat, 7 Feb 2026 15:49:53 +0700 Subject: [PATCH 1/5] fix(dci): allow non-admin users to use DCI verification - Use sudo() when accessing OAuth2 credentials in data source - Use sudo() when caching OAuth2 token (write to restricted model) - Add _after_submit() hook to auto-approve CR on submit if DCI verified This enables users like demo_officer to use DCI birth verification without requiring administrator privileges. --- spp_dci_client/models/data_source.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/spp_dci_client/models/data_source.py b/spp_dci_client/models/data_source.py index a58913c5..08a84def 100644 --- a/spp_dci_client/models/data_source.py +++ b/spp_dci_client/models/data_source.py @@ -326,10 +326,12 @@ def get_oauth2_token(self, force_refresh=False): _logger.info("Requesting new OAuth2 token for data source: %s", self.code) try: + # Use sudo() to access OAuth2 credentials which are restricted to administrators + sudo_self = self.sudo() token_data = { "grant_type": "client_credentials", - "client_id": self.oauth2_client_id, - "client_secret": self.oauth2_client_secret, + "client_id": sudo_self.oauth2_client_id, + "client_secret": sudo_self.oauth2_client_secret, } if self.oauth2_scope: @@ -358,9 +360,9 @@ def get_oauth2_token(self, force_refresh=False): if not access_token: raise UserError(_("OAuth2 token response did not contain access_token")) - # Cache token with expiry + # Cache token with expiry (use sudo to write to restricted model) expires_at = now + timedelta(seconds=expires_in) - self.write( + sudo_self.write( { "_oauth2_access_token": access_token, "_oauth2_token_expires_at": expires_at, From 7bcc8e4ada859709a17242d35de02e9b21705738 Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Sat, 7 Feb 2026 14:14:00 +0700 Subject: [PATCH 2/5] feat(spp_approval): add action_approve_system() for automated approvals Add a dedicated method for system-initiated approvals that bypasses user permission checks. This enables automated approval workflows triggered by system events (e.g., DCI verification match) where there is no human approver. Also adds security control to invalidate verification when verified fields (name, DOB, gender, BRN) are edited after verification. --- spp_approval/models/approval_mixin.py | 29 +++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/spp_approval/models/approval_mixin.py b/spp_approval/models/approval_mixin.py index 92816eaf..a2f469bb 100644 --- a/spp_approval/models/approval_mixin.py +++ b/spp_approval/models/approval_mixin.py @@ -332,6 +332,35 @@ def action_approve(self, comment=None): record._check_can_approve() record._do_approve(comment=comment) + def action_approve_system(self, comment=None): + """System-initiated approval bypassing user permission checks. + + Use this for automated approvals triggered by system events (e.g., DCI + verification match, scheduled jobs) where there is no human approver. + + This method uses sudo() to bypass access controls and skips the + _check_can_approve() permission validation. + + Args: + comment: Optional approval comment for audit trail + """ + for record in self: + if record.approval_state != "pending": + _logger.warning( + "Skipping system approval for %s %s: state is %s, not pending", + record._name, + record.id, + record.approval_state, + ) + continue + record.sudo()._do_approve(comment=comment) + _logger.info( + "System auto-approved %s %s: %s", + record._name, + record.id, + comment or "(no comment)", + ) + def _do_approve(self, comment=None, auto=False): """Internal method to perform approval.""" self.ensure_one() From 6507a5a632ae0789468defa639cd0f663e87db82 Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Fri, 13 Feb 2026 22:13:41 +0100 Subject: [PATCH 3/5] feat(spp_dci_client): log outgoing DCI requests to audit trail Instrument _make_request to log all outgoing calls via spp.api.outgoing.log (soft dependency). Logs persist in a separate cursor so they survive transaction rollback on UserError. Captures timing, status codes, request/response payloads, and error details for success, HTTP errors, connection errors, timeouts, and 401 retries. --- spp_dci_client/services/client.py | 122 +++++ spp_dci_client/tests/__init__.py | 1 + .../tests/test_outgoing_log_integration.py | 442 ++++++++++++++++++ 3 files changed, 565 insertions(+) create mode 100644 spp_dci_client/tests/test_outgoing_log_integration.py diff --git a/spp_dci_client/services/client.py b/spp_dci_client/services/client.py index 187dacc0..5ca3a5b2 100644 --- a/spp_dci_client/services/client.py +++ b/spp_dci_client/services/client.py @@ -1,6 +1,7 @@ """DCI Client Service for making signed API requests.""" import logging +import time import uuid from datetime import UTC, datetime from typing import Any @@ -904,6 +905,13 @@ def _make_request(self, endpoint: str, envelope: dict, _retry_auth: bool = True) # Get headers from data source (includes auth) headers = self.data_source.get_headers() + # Track timing and result for outgoing log + start_time = time.monotonic() + log_status = "success" + log_status_code = None + log_response_data = None + log_error_detail = None + try: _logger.info( "Making DCI request to %s (action: %s)", @@ -930,6 +938,13 @@ def _make_request(self, endpoint: str, envelope: dict, _retry_auth: bool = True) # Handle 401 Unauthorized - try refreshing token once if response.status_code == 401 and _retry_auth and self.data_source.auth_type == "oauth2": _logger.warning("Got 401 Unauthorized, clearing OAuth2 token cache and retrying with fresh token") + log_status = "http_error" + log_status_code = 401 + log_error_detail = "401 Unauthorized - retrying with fresh token" + try: + log_response_data = response.json() + except Exception: + log_response_data = None self.data_source.clear_oauth2_token_cache() return self._make_request(endpoint, envelope, _retry_auth=False) @@ -938,6 +953,8 @@ def _make_request(self, endpoint: str, envelope: dict, _retry_auth: bool = True) # Parse response response_data = response.json() + log_status_code = response.status_code + log_response_data = response_data _logger.info( "DCI request successful (status: %s, message_id: %s)", @@ -949,11 +966,15 @@ def _make_request(self, endpoint: str, envelope: dict, _retry_auth: bool = True) return response_data except httpx.HTTPStatusError as e: + log_status = "http_error" + log_status_code = e.response.status_code + # Log technical details for troubleshooting technical_detail = f"DCI request failed with status {e.response.status_code}" response_text = e.response.text try: error_data = e.response.json() + log_response_data = error_data if "header" in error_data and "status_reason_message" in error_data["header"]: technical_detail += f": {error_data['header']['status_reason_message']}" else: @@ -961,6 +982,7 @@ def _make_request(self, endpoint: str, envelope: dict, _retry_auth: bool = True) except Exception: technical_detail += f": {response_text}" + log_error_detail = technical_detail _logger.error(technical_detail) _logger.error("Full response body: %s", response_text) _logger.error("Request envelope was: %s", envelope) @@ -978,26 +1000,126 @@ def _make_request(self, endpoint: str, envelope: dict, _retry_auth: bool = True) error_str = str(e).lower() if "timeout" in error_str or "timed out" in error_str: connection_type = "timeout" + log_status = "timeout" elif "ssl" in error_str or "certificate" in error_str: connection_type = "ssl" + log_status = "connection_error" elif "name or service not known" in error_str or "nodename nor servname" in error_str: connection_type = "dns" + log_status = "connection_error" else: connection_type = "connection" + log_status = "connection_error" + + log_error_detail = technical_detail # Show user-friendly message user_msg = format_connection_error(connection_type, technical_detail) raise UserError(user_msg) from e except Exception as e: + log_status = "error" + # Log technical details for troubleshooting technical_detail = f"Unexpected error during DCI request: {str(e)}" + log_error_detail = technical_detail _logger.error(technical_detail, exc_info=True) # Show generic user-friendly message user_msg = _("An unexpected error occurred. Please contact your administrator.") raise UserError(user_msg) from e + finally: + duration_ms = int((time.monotonic() - start_time) * 1000) + self._log_outgoing_call( + url=url, + endpoint=endpoint, + envelope=envelope, + response_data=log_response_data, + status_code=log_status_code, + duration_ms=duration_ms, + status=log_status, + error_detail=log_error_detail, + ) + + # ========================================================================= + # OUTGOING LOG + # ========================================================================= + + def _log_outgoing_call( + self, + url: str, + endpoint: str, + envelope: dict, + response_data: dict | None, + status_code: int | None, + duration_ms: int, + status: str, + error_detail: str | None, + ): + """Log an outgoing API call to spp.api.outgoing.log (soft dependency). + + Uses runtime check to avoid hard manifest dependency on spp_api_v2. + Uses a separate database cursor so log entries persist even when the + caller's transaction is rolled back (e.g., on UserError). + Logging failures are swallowed so they never block the actual request. + """ + try: + if "spp.api.outgoing.log" not in self.env: + return + + from odoo.addons.spp_api_v2.services.outgoing_api_log_service import OutgoingApiLogService + + # Capture values from the current env before opening a new cursor, + # since self.data_source won't be accessible from the new cursor's env. + service_code = getattr(self.data_source, "code", None) or "dci" + origin_record_id = self.data_source.id if hasattr(self.data_source, "id") else None + user_id = self.env.uid + sanitized_envelope = self._copy_envelope_for_log(envelope) + + # Use a separate cursor so log entries survive transaction rollback. + with self.env.registry.cursor() as new_cr: + new_env = self.env(cr=new_cr) + service = OutgoingApiLogService( + new_env, + service_name="DCI Client", + service_code=service_code, + user_id=user_id, + ) + + service.log_call( + url=url, + endpoint=endpoint, + http_method="POST", + request_summary=sanitized_envelope, + response_summary=response_data, + response_status_code=status_code, + duration_ms=duration_ms, + origin_model="spp.dci.data.source", + origin_record_id=origin_record_id, + status=status, + error_detail=error_detail, + ) + except Exception: + _logger.warning("Failed to log outgoing API call (non-blocking)", exc_info=True) + + def _copy_envelope_for_log(self, envelope: dict) -> dict | None: + """Copy the request envelope for audit log storage. + + Returns a shallow copy suitable for JSON storage. The cryptographic + signature is preserved for auditability (non-repudiation). + + Args: + envelope: Request envelope dict + + Returns: + Copy of the envelope, or None if input is falsy + """ + if not envelope: + return None + + return dict(envelope) + # ========================================================================= # HELPER METHODS # ========================================================================= diff --git a/spp_dci_client/tests/__init__.py b/spp_dci_client/tests/__init__.py index 71410b98..8cd038a7 100644 --- a/spp_dci_client/tests/__init__.py +++ b/spp_dci_client/tests/__init__.py @@ -3,3 +3,4 @@ from . import test_data_source from . import test_client_service +from . import test_outgoing_log_integration diff --git a/spp_dci_client/tests/test_outgoing_log_integration.py b/spp_dci_client/tests/test_outgoing_log_integration.py new file mode 100644 index 00000000..1257ee47 --- /dev/null +++ b/spp_dci_client/tests/test_outgoing_log_integration.py @@ -0,0 +1,442 @@ +# Part of OpenSPP. See LICENSE file for full copyright and licensing details. +"""Tests for outgoing API log integration in DCI client""" + +import unittest +from unittest.mock import MagicMock, patch + +from odoo.tests import TransactionCase + + +class TestOutgoingLogClientMethods(TransactionCase): + """Test DCI client logging helper methods (no spp_api_v2 dependency needed)""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.DataSource = cls.env["spp.dci.data.source"] + + def _create_test_data_source(self, **kwargs): + """Helper to create a test data source""" + vals = { + "name": "Test CRVS", + "code": "test_crvs", + "base_url": "https://crvs.example.org/api", + "auth_type": "none", + "our_sender_id": "openspp.example.org", + "our_callback_uri": "https://openspp.example.org/callback", + "registry_type": "ns:org:RegistryType:Civil", + } + vals.update(kwargs) + return self.DataSource.create(vals) + + def test_log_skips_if_model_missing(self): + """_log_outgoing_call does nothing if spp.api.outgoing.log model is not installed""" + from ..services.client import DCIClient + + ds = self._create_test_data_source() + client = DCIClient(ds, self.env) + + # Mock env to not contain the outgoing log model + with patch.object(client, "env") as mock_env: + mock_env.__contains__ = lambda self, key: False + + # Should not raise + client._log_outgoing_call( + url="https://example.org/test", + endpoint="/test", + envelope={"header": {"action": "search"}, "message": {}}, + response_data=None, + status_code=None, + duration_ms=100, + status="success", + error_detail=None, + ) + + def test_copy_envelope_for_log_preserves_signature(self): + """_copy_envelope_for_log preserves signature for auditability""" + from ..services.client import DCIClient + + ds = self._create_test_data_source() + client = DCIClient(ds, self.env) + + envelope = { + "signature": "cryptographic_signature_12345", + "header": {"action": "search"}, + "message": {"test": "data"}, + } + + copied = client._copy_envelope_for_log(envelope) + + # Signature is preserved for audit trail / non-repudiation + self.assertEqual(copied["signature"], "cryptographic_signature_12345") + self.assertEqual(copied["header"], {"action": "search"}) + self.assertEqual(copied["message"], {"test": "data"}) + + def test_copy_envelope_for_log_none(self): + """_copy_envelope_for_log returns None for None/empty input""" + from ..services.client import DCIClient + + ds = self._create_test_data_source() + client = DCIClient(ds, self.env) + + self.assertIsNone(client._copy_envelope_for_log(None)) + self.assertIsNone(client._copy_envelope_for_log({})) + + def test_copy_envelope_for_log_returns_copy(self): + """_copy_envelope_for_log returns a copy, not the original""" + from ..services.client import DCIClient + + ds = self._create_test_data_source() + client = DCIClient(ds, self.env) + + envelope = {"header": {"action": "search"}, "message": {}} + copied = client._copy_envelope_for_log(envelope) + + self.assertIsNot(copied, envelope) + self.assertEqual(copied, envelope) + + +@unittest.skipUnless( + True, # Actual check is in setUpClass since we need the Odoo env + "spp_api_v2 required", +) +class TestOutgoingLogIntegration(TransactionCase): + """Test DCI client integration with outgoing API log. + + These tests require spp_api_v2 to be installed (provides spp.api.outgoing.log). + """ + + @classmethod + def setUpClass(cls): + super().setUpClass() + if "spp.api.outgoing.log" not in cls.env: + raise unittest.SkipTest("spp_api_v2 not installed (spp.api.outgoing.log model not available)") + cls.DataSource = cls.env["spp.dci.data.source"] + cls.OutgoingLog = cls.env["spp.api.outgoing.log"] + + def _create_test_data_source(self, **kwargs): + """Helper to create a test data source""" + vals = { + "name": "Test CRVS", + "code": "test_crvs", + "base_url": "https://crvs.example.org/api", + "auth_type": "none", + "our_sender_id": "openspp.example.org", + "our_callback_uri": "https://openspp.example.org/callback", + "registry_type": "ns:org:RegistryType:Civil", + } + vals.update(kwargs) + return self.DataSource.create(vals) + + def _make_mock_response(self, status_code=200, json_data=None): + """Helper to create a mock HTTP response""" + mock_response = MagicMock() + mock_response.status_code = status_code + mock_response.json.return_value = json_data or {"header": {"status": "success"}} + mock_response.text = str(json_data or {"header": {"status": "success"}}) + mock_response.raise_for_status = MagicMock() + return mock_response + + def _build_test_envelope(self, client): + """Helper to build a test envelope""" + return client._build_envelope(action="search", message={"test": "data"}) + + def _find_latest_log(self, url_filter="crvs.example.org"): + """Helper to find the most recent outgoing log entry""" + return self.OutgoingLog.search( + [("url", "like", url_filter)], + order="id desc", + limit=1, + ) + + @patch("httpx.Client") + def test_make_request_logs_success(self, mock_client_class): + """Successful request creates outgoing log with status=success""" + from ..services.client import DCIClient + + ds = self._create_test_data_source() + client = DCIClient(ds, self.env) + + mock_response = self._make_mock_response() + mock_http_client = MagicMock() + mock_http_client.post.return_value = mock_response + mock_http_client.__enter__.return_value = mock_http_client + mock_http_client.__exit__.return_value = None + mock_client_class.return_value = mock_http_client + + envelope = self._build_test_envelope(client) + client._make_request("/registry/sync/search", envelope) + + log = self._find_latest_log() + self.assertTrue(log, "Outgoing log should be created on success") + self.assertEqual(log.status, "success") + self.assertEqual(log.response_status_code, 200) + self.assertEqual(log.endpoint, "/registry/sync/search") + self.assertEqual(log.service_name, "DCI Client") + self.assertEqual(log.service_code, "test_crvs") + self.assertGreater(log.duration_ms, 0) + + @patch("httpx.Client") + def test_make_request_logs_http_error(self, mock_client_class): + """HTTP error creates outgoing log with status=http_error""" + import httpx + + from ..services.client import DCIClient + + ds = self._create_test_data_source() + client = DCIClient(ds, self.env) + + mock_response = MagicMock() + mock_response.status_code = 500 + mock_response.text = "Internal Server Error" + mock_response.json.return_value = {"error": "server_error"} + mock_request = MagicMock() + + def raise_status(): + raise httpx.HTTPStatusError("Server Error", request=mock_request, response=mock_response) + + mock_response.raise_for_status = raise_status + + mock_http_client = MagicMock() + mock_http_client.post.return_value = mock_response + mock_http_client.__enter__.return_value = mock_http_client + mock_http_client.__exit__.return_value = None + mock_client_class.return_value = mock_http_client + + envelope = self._build_test_envelope(client) + + from odoo.exceptions import UserError + + with self.assertRaises(UserError): + client._make_request("/registry/sync/search", envelope) + + log = self._find_latest_log() + self.assertTrue(log, "Outgoing log should be created on HTTP error") + self.assertEqual(log.status, "http_error") + self.assertEqual(log.response_status_code, 500) + self.assertTrue(log.error_detail) + + @patch("httpx.Client") + def test_make_request_logs_connection_error(self, mock_client_class): + """Connection error creates outgoing log with status=connection_error""" + import httpx + + from ..services.client import DCIClient + + ds = self._create_test_data_source() + client = DCIClient(ds, self.env) + + mock_http_client = MagicMock() + mock_http_client.post.side_effect = httpx.ConnectError("Connection refused") + mock_http_client.__enter__.return_value = mock_http_client + mock_http_client.__exit__.return_value = None + mock_client_class.return_value = mock_http_client + + envelope = self._build_test_envelope(client) + + from odoo.exceptions import UserError + + with self.assertRaises(UserError): + client._make_request("/registry/sync/search", envelope) + + log = self._find_latest_log() + self.assertTrue(log, "Outgoing log should be created on connection error") + self.assertEqual(log.status, "connection_error") + self.assertTrue(log.error_detail) + + @patch("httpx.Client") + def test_make_request_logs_timeout(self, mock_client_class): + """Timeout creates outgoing log with status=timeout""" + import httpx + + from ..services.client import DCIClient + + ds = self._create_test_data_source() + client = DCIClient(ds, self.env) + + mock_http_client = MagicMock() + mock_http_client.post.side_effect = httpx.ReadTimeout("Read timed out") + mock_http_client.__enter__.return_value = mock_http_client + mock_http_client.__exit__.return_value = None + mock_client_class.return_value = mock_http_client + + envelope = self._build_test_envelope(client) + + from odoo.exceptions import UserError + + with self.assertRaises(UserError): + client._make_request("/registry/sync/search", envelope) + + log = self._find_latest_log() + self.assertTrue(log, "Outgoing log should be created on timeout") + self.assertEqual(log.status, "timeout") + self.assertTrue(log.error_detail) + + @patch("httpx.Client") + def test_make_request_logs_ssl_error(self, mock_client_class): + """SSL error creates outgoing log with status=connection_error""" + import httpx + + from ..services.client import DCIClient + + ds = self._create_test_data_source() + client = DCIClient(ds, self.env) + + mock_http_client = MagicMock() + mock_http_client.post.side_effect = httpx.ConnectError( + "[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed" + ) + mock_http_client.__enter__.return_value = mock_http_client + mock_http_client.__exit__.return_value = None + mock_client_class.return_value = mock_http_client + + envelope = self._build_test_envelope(client) + + from odoo.exceptions import UserError + + with self.assertRaises(UserError): + client._make_request("/registry/sync/search", envelope) + + log = self._find_latest_log() + self.assertTrue(log, "Outgoing log should be created on SSL error") + self.assertEqual(log.status, "connection_error") + self.assertIn("SSL", log.error_detail.upper()) + + @patch("httpx.Client") + def test_make_request_logs_dns_error(self, mock_client_class): + """DNS resolution error creates outgoing log with status=connection_error""" + import httpx + + from ..services.client import DCIClient + + ds = self._create_test_data_source() + client = DCIClient(ds, self.env) + + mock_http_client = MagicMock() + mock_http_client.post.side_effect = httpx.ConnectError("Name or service not known") + mock_http_client.__enter__.return_value = mock_http_client + mock_http_client.__exit__.return_value = None + mock_client_class.return_value = mock_http_client + + envelope = self._build_test_envelope(client) + + from odoo.exceptions import UserError + + with self.assertRaises(UserError): + client._make_request("/registry/sync/search", envelope) + + log = self._find_latest_log() + self.assertTrue(log, "Outgoing log should be created on DNS error") + self.assertEqual(log.status, "connection_error") + self.assertTrue(log.error_detail) + + @patch("httpx.Client") + def test_make_request_logs_generic_exception(self, mock_client_class): + """Generic exception creates outgoing log with status=error""" + from ..services.client import DCIClient + + ds = self._create_test_data_source() + client = DCIClient(ds, self.env) + + mock_http_client = MagicMock() + mock_http_client.post.side_effect = RuntimeError("Something unexpected") + mock_http_client.__enter__.return_value = mock_http_client + mock_http_client.__exit__.return_value = None + mock_client_class.return_value = mock_http_client + + envelope = self._build_test_envelope(client) + + from odoo.exceptions import UserError + + with self.assertRaises(UserError): + client._make_request("/registry/sync/search", envelope) + + log = self._find_latest_log() + self.assertTrue(log, "Outgoing log should be created on generic exception") + self.assertEqual(log.status, "error") + self.assertTrue(log.error_detail) + + @patch("httpx.Client") + def test_401_retry_creates_two_log_entries(self, mock_client_class): + """401 retry path creates log entry for both the 401 and the retry result""" + from ..services.client import DCIClient + + ds = self._create_test_data_source(auth_type="oauth2") + client = DCIClient(ds, self.env) + + # First call returns 401, second returns 200 + mock_401_response = MagicMock() + mock_401_response.status_code = 401 + mock_401_response.text = "Unauthorized" + mock_401_response.json.return_value = {"error": "invalid_token"} + mock_401_response.raise_for_status = MagicMock() + + mock_200_response = self._make_mock_response() + + mock_http_client = MagicMock() + mock_http_client.post.side_effect = [mock_401_response, mock_200_response] + mock_http_client.__enter__.return_value = mock_http_client + mock_http_client.__exit__.return_value = None + mock_client_class.return_value = mock_http_client + + envelope = self._build_test_envelope(client) + + with patch.object(ds, "clear_oauth2_token_cache"): + client._make_request("/registry/sync/search", envelope) + + # Should have two log entries: one for 401, one for retry success + logs = self.OutgoingLog.search( + [("url", "like", "crvs.example.org")], + order="id desc", + limit=2, + ) + self.assertEqual(len(logs), 2, "Should have two log entries (401 + retry)") + + # Most recent should be the retry success + self.assertEqual(logs[0].status, "success") + self.assertEqual(logs[0].response_status_code, 200) + + # Earlier should be the 401 error + self.assertEqual(logs[1].status, "http_error") + self.assertEqual(logs[1].response_status_code, 401) + self.assertIn("retrying", logs[1].error_detail.lower()) + # M-1 fix: 401 response body should be captured + self.assertTrue(logs[1].response_summary) + + @patch("httpx.Client") + def test_log_failure_does_not_block_request(self, mock_client_class): + """Logging failure does not prevent the actual request from succeeding""" + from ..services.client import DCIClient + + ds = self._create_test_data_source() + client = DCIClient(ds, self.env) + + # Mock HTTP response (success) + mock_response = self._make_mock_response() + mock_http_client = MagicMock() + mock_http_client.post.return_value = mock_response + mock_http_client.__enter__.return_value = mock_http_client + mock_http_client.__exit__.return_value = None + mock_client_class.return_value = mock_http_client + + # Monkey-patch the model to raise + original_log_call = self.OutgoingLog.__class__.log_call + + def broken_log_call(self_model, **kwargs): + raise RuntimeError("Database error") + + self.OutgoingLog.__class__.log_call = broken_log_call + try: + # This should not raise despite broken logging + client._log_outgoing_call( + url="https://example.org/test", + endpoint="/test", + envelope={"header": {"action": "search"}, "message": {}}, + response_data=None, + status_code=200, + duration_ms=100, + status="success", + error_detail=None, + ) + finally: + self.OutgoingLog.__class__.log_call = original_log_call From b14a93190f1118b85ca7f4acc0a29260a09821f8 Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Wed, 18 Feb 2026 16:30:10 +0700 Subject: [PATCH 4/5] fix(spp_approval,spp_dci_client): prevent RPC exposure of system approval and narrow exception handling Rename action_approve_system to _action_approve_system to prevent Odoo from exposing it via the RPC interface, since it uses sudo() and skips _check_can_approve() authorization checks. Replace broad Exception catches in spp_dci_client with specific exception types (json.JSONDecodeError, KeyError, TypeError) and add json import. --- spp_approval/models/approval_mixin.py | 5 ++++- spp_dci_client/services/client.py | 5 +++-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/spp_approval/models/approval_mixin.py b/spp_approval/models/approval_mixin.py index a2f469bb..6e4c398c 100644 --- a/spp_approval/models/approval_mixin.py +++ b/spp_approval/models/approval_mixin.py @@ -332,7 +332,7 @@ def action_approve(self, comment=None): record._check_can_approve() record._do_approve(comment=comment) - def action_approve_system(self, comment=None): + def _action_approve_system(self, comment=None): """System-initiated approval bypassing user permission checks. Use this for automated approvals triggered by system events (e.g., DCI @@ -341,6 +341,9 @@ def action_approve_system(self, comment=None): This method uses sudo() to bypass access controls and skips the _check_can_approve() permission validation. + The underscore prefix is intentional — it prevents this method from being + callable via Odoo's RPC interface, since it must not be exposed to users. + Args: comment: Optional approval comment for audit trail """ diff --git a/spp_dci_client/services/client.py b/spp_dci_client/services/client.py index 5ca3a5b2..cefbe1c2 100644 --- a/spp_dci_client/services/client.py +++ b/spp_dci_client/services/client.py @@ -1,5 +1,6 @@ """DCI Client Service for making signed API requests.""" +import json import logging import time import uuid @@ -943,7 +944,7 @@ def _make_request(self, endpoint: str, envelope: dict, _retry_auth: bool = True) log_error_detail = "401 Unauthorized - retrying with fresh token" try: log_response_data = response.json() - except Exception: + except json.JSONDecodeError: log_response_data = None self.data_source.clear_oauth2_token_cache() return self._make_request(endpoint, envelope, _retry_auth=False) @@ -979,7 +980,7 @@ def _make_request(self, endpoint: str, envelope: dict, _retry_auth: bool = True) technical_detail += f": {error_data['header']['status_reason_message']}" else: technical_detail += f": {response_text}" - except Exception: + except (json.JSONDecodeError, KeyError, TypeError): technical_detail += f": {response_text}" log_error_detail = technical_detail From 3ea418e91969f4b06cf9832ddb82e0b6bee55753 Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Wed, 18 Feb 2026 17:24:02 +0700 Subject: [PATCH 5/5] fix(spp_approval,spp_dci_client): add auto=True to system approval and fix test log ordering Pass auto=True to _do_approve in _action_approve_system so automated approvals show "(auto)" in activity feedback and skip submitter notification, matching the intent of system-initiated approvals. Fix reversed log assertions in test_401_retry_creates_two_log_entries: due to try-finally semantics with recursive calls, the inner retry's finally creates the success log first (lower ID), so with order="id desc" logs[0] is the 401 entry and logs[1] is the success entry. --- spp_approval/models/approval_mixin.py | 2 +- spp_dci_client/models/data_source.py | 2 +- .../tests/test_outgoing_log_integration.py | 21 +++++++++++-------- 3 files changed, 14 insertions(+), 11 deletions(-) diff --git a/spp_approval/models/approval_mixin.py b/spp_approval/models/approval_mixin.py index 6e4c398c..cfc9c65b 100644 --- a/spp_approval/models/approval_mixin.py +++ b/spp_approval/models/approval_mixin.py @@ -356,7 +356,7 @@ def _action_approve_system(self, comment=None): record.approval_state, ) continue - record.sudo()._do_approve(comment=comment) + record.sudo()._do_approve(comment=comment, auto=True) # nosemgrep: odoo-sudo-without-context _logger.info( "System auto-approved %s %s: %s", record._name, diff --git a/spp_dci_client/models/data_source.py b/spp_dci_client/models/data_source.py index 08a84def..a1cb4d90 100644 --- a/spp_dci_client/models/data_source.py +++ b/spp_dci_client/models/data_source.py @@ -327,7 +327,7 @@ def get_oauth2_token(self, force_refresh=False): try: # Use sudo() to access OAuth2 credentials which are restricted to administrators - sudo_self = self.sudo() + sudo_self = self.sudo() # nosemgrep: odoo-sudo-without-context token_data = { "grant_type": "client_credentials", "client_id": sudo_self.oauth2_client_id, diff --git a/spp_dci_client/tests/test_outgoing_log_integration.py b/spp_dci_client/tests/test_outgoing_log_integration.py index 1257ee47..7cefec24 100644 --- a/spp_dci_client/tests/test_outgoing_log_integration.py +++ b/spp_dci_client/tests/test_outgoing_log_integration.py @@ -392,16 +392,19 @@ def test_401_retry_creates_two_log_entries(self, mock_client_class): ) self.assertEqual(len(logs), 2, "Should have two log entries (401 + retry)") - # Most recent should be the retry success - self.assertEqual(logs[0].status, "success") - self.assertEqual(logs[0].response_status_code, 200) - - # Earlier should be the 401 error - self.assertEqual(logs[1].status, "http_error") - self.assertEqual(logs[1].response_status_code, 401) - self.assertIn("retrying", logs[1].error_detail.lower()) + # With order="id desc", logs[0] has the highest ID (created last). + # Due to try-finally semantics, the recursive retry's finally runs first + # (lower ID = success), then the outer call's finally runs (higher ID = 401). + # So logs[0] (highest ID) is the 401 entry, logs[1] (lower ID) is the success. + self.assertEqual(logs[0].status, "http_error") + self.assertEqual(logs[0].response_status_code, 401) + self.assertIn("retrying", logs[0].error_detail.lower()) # M-1 fix: 401 response body should be captured - self.assertTrue(logs[1].response_summary) + self.assertTrue(logs[0].response_summary) + + # Earlier log (lower ID) is the retry success + self.assertEqual(logs[1].status, "success") + self.assertEqual(logs[1].response_status_code, 200) @patch("httpx.Client") def test_log_failure_does_not_block_request(self, mock_client_class):