fix: harden payment checks and sensitive data handling

buckaroo/http/client.py

@@ -244,27 +244,7 @@ def json(self) -> Dict[str, Any]:
 
     def is_successful_payment(self) -> bool:
         """Check if the payment was successful based on Buckaroo response."""
-        if not self.success:
-            return False
-
-        # Check Buckaroo-specific success indicators
-        if self._data and "Status" in self._data:
-            # Buckaroo status codes for successful payments
-            success_statuses = [190, 490, 491, 492, 790, 791, 792, 793]
-            status = self._data.get("Status", {})
-            if status and "Code" in status:
-                code = status.get("Code")
-                # Handle nested Code structure
-                if isinstance(code, dict):
-                    actual_code = code.get("Code")
-                elif isinstance(code, int):
-                    actual_code = code
-                else:
-                    actual_code = None
-
-                return actual_code in success_statuses if actual_code is not None else False
-
-        return self.success
+        return self.success and self.get_status_code() == 190
 
     def get_payment_key(self) -> Optional[str]:
         """Get the payment key from the response."""
@@ -287,25 +267,15 @@ def get_transaction_key(self) -> Optional[str]:
 
     def get_status_code(self) -> Optional[int]:
         """Get the Buckaroo status code."""
-        if not self._data:
+        if not isinstance(self._data, dict):
             return None
-
-        status = self._data.get("Status", {})
-        if not status:
+        status = self._data.get("Status")
+        if not isinstance(status, dict):
             return None
-
         code = status.get("Code")
-        if code is None:
-            return None
-
-        # Handle nested Code structure: {"Code": 490, "Description": "Failed"}
         if isinstance(code, dict):
-            return code.get("Code")
-        # Handle simple integer code
-        elif isinstance(code, int):
-            return code
-
-        return None
+            code = code.get("Code")
+        return code if type(code) is int else None
 
     def get_status_message(self) -> Optional[str]:
         """Get the Buckaroo status message."""

buckaroo/http/strategies/curl_strategy.py

@@ -54,12 +54,18 @@ def request(
         cmd = self._build_curl_command(
             method=method,
             url=url,
-            headers=headers,
-            data=data,
             timeout=timeout or self._timeout,
             verify_ssl=verify_ssl,
         )
 
+        # Keep credentials and payloads off the process command line.
+        config_lines = []
+        for key, value in {**self._default_headers, **(headers or {})}.items():
+            config_lines.append(f"header = {self._quote_config_value(f'{key}: {value}')}")
+        if data and method.upper() in ["POST", "PUT", "PATCH"]:
+            config_lines.append(f"data-raw = {self._quote_config_value(data)}")
+        config_input = "\n".join(config_lines) + "\n"
+
         # Execute curl with retry logic
         if self._retry_attempts <= 0:
             raise Exception("Request failed after all retry attempts")
@@ -68,8 +74,10 @@ def request(
             try:
                 result = subprocess.run(
                     cmd,
+                    input=config_input,
                     capture_output=True,
                     text=True,
+                    encoding="utf-8",
                     timeout=timeout or self._timeout,
                     check=False,  # Don't raise on non-zero exit codes
                 )
@@ -95,8 +103,6 @@ def _build_curl_command(
         self,
         method: str,
         url: str,
-        headers: Optional[Dict[str, str]] = None,
-        data: Optional[str] = None,
         timeout: int = 30,
         verify_ssl: bool = True,
     ) -> List[str]:
@@ -106,8 +112,6 @@ def _build_curl_command(
         Args:
             method: HTTP method
             url: Request URL
-            headers: Request headers
-            data: Request body data
             timeout: Request timeout
             verify_ssl: Whether to verify SSL
 
@@ -116,6 +120,8 @@ def _build_curl_command(
         """
         cmd = [
             "curl",
+            "--config",
+            "-",
             "-X",
             method.upper(),
             "--location",  # Follow redirects
@@ -131,23 +137,27 @@ def _build_curl_command(
         if not verify_ssl:
             cmd.extend(["--insecure"])
 
-        # Add headers
-        all_headers = {**self._default_headers}
-        if headers:
-            all_headers.update(headers)
-
-        for key, value in all_headers.items():
-            cmd.extend(["-H", f"{key}: {value}"])
-
-        # Add data for POST/PUT requests
-        if data and method.upper() in ["POST", "PUT", "PATCH"]:
-            cmd.extend(["--data", data])
-
         # Add URL last
         cmd.append(url)
 
         return cmd
 
+    @staticmethod
+    def _quote_config_value(value: str) -> str:
+        """Quote a literal curl config value without allowing new options."""
+        if "\0" in value:
+            raise ValueError("curl input cannot contain NUL characters")
+        for original, escaped in (
+            ("\\", "\\\\"),
+            ('"', '\\"'),
+            ("\n", "\\n"),
+            ("\r", "\\r"),
+            ("\t", "\\t"),
+            ("\v", "\\v"),
+        ):
+            value = value.replace(original, escaped)
+        return f'"{value}"'
+
     def _parse_curl_output(self, result: subprocess.CompletedProcess) -> HttpResponse:
         """
         Parse curl output into HttpResponse object.

buckaroo/observers/logging_observer.py

@@ -165,12 +165,14 @@ def _format_json(self, data: Any) -> str:
                 try:
                     data = json.loads(data)
                 except json.JSONDecodeError:
-                    return data
+                    return "***BODY_OMITTED***" if self.config.mask_sensitive_data else data
 
             masked_data = self._mask_sensitive_data(data)
+            if self.config.mask_sensitive_data:
+                return json.dumps(masked_data, indent=2)
             return json.dumps(masked_data, indent=2, default=str)
         except Exception:
-            return str(data)
+            return "***BODY_OMITTED***" if self.config.mask_sensitive_data else str(data)
 
     def log_request(
         self,

tests/unit/conftest.py

@@ -1,5 +1,8 @@
 """Shared fixtures for unit tests."""
 
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
 import pytest
 
 from buckaroo._buckaroo_client import BuckarooClient
@@ -40,3 +43,33 @@ def env_credentials(monkeypatch):
     monkeypatch.setenv("BUCKAROO_STORE_KEY", "sk")
     monkeypatch.setenv("BUCKAROO_SECRET_KEY", "ss")
     return monkeypatch
+
+
+@pytest.fixture
+def local_endpoint():
+    received = []
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_POST(self):
+            body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
+            received.append((self.command, dict(self.headers), body))
+            self.send_response(200)
+            self.end_headers()
+            self.wfile.write(b'{"access_token":"synthetic-token"}')
+
+        do_PUT = do_POST
+        do_PATCH = do_POST
+        do_GET = do_POST
+
+        def log_message(self, *args):
+            pass
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield f"http://127.0.0.1:{server.server_port}/token", received
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)

tests/unit/http/strategies/test_curl_strategy.py

@@ -1,5 +1,6 @@
 """Unit tests for buckaroo.http.strategies.curl_strategy."""
 
+import shutil
 import subprocess
 from types import SimpleNamespace
 from unittest.mock import patch
@@ -94,68 +95,6 @@ def test_omits_insecure_when_verify_ssl_true(self):
 
         assert "--insecure" not in cmd
 
-    def test_merges_default_headers_with_per_call_headers(self):
-        strategy = CurlStrategy()
-        strategy.configure(default_headers={"X-Default": "d", "X-Shared": "default"})
-
-        cmd = strategy._build_curl_command(
-            method="GET",
-            url="https://x",
-            headers={"X-Call": "c", "X-Shared": "perCall"},
-        )
-
-        header_values = [cmd[i + 1] for i, arg in enumerate(cmd) if arg == "-H"]
-        assert "X-Default: d" in header_values
-        assert "X-Call: c" in header_values
-        # per-call wins over default for overlapping key
-        assert "X-Shared: perCall" in header_values
-        assert "X-Shared: default" not in header_values
-
-    def test_headers_omitted_when_neither_default_nor_per_call_provided(self):
-        strategy = CurlStrategy()
-
-        cmd = strategy._build_curl_command(method="GET", url="https://x")
-
-        assert "-H" not in cmd
-
-    @pytest.mark.parametrize("method", ["POST", "PUT", "PATCH"])
-    def test_data_attached_for_write_methods(self, method):
-        strategy = CurlStrategy()
-
-        cmd = strategy._build_curl_command(method=method, url="https://x", data='{"a":1}')
-
-        assert "--data" in cmd
-        assert cmd[cmd.index("--data") + 1] == '{"a":1}'
-
-    @pytest.mark.parametrize("method", ["GET", "DELETE"])
-    def test_data_omitted_for_read_methods(self, method):
-        strategy = CurlStrategy()
-
-        cmd = strategy._build_curl_command(method=method, url="https://x", data='{"a":1}')
-
-        assert "--data" not in cmd
-
-    def test_data_omitted_when_data_is_none_even_for_post(self):
-        strategy = CurlStrategy()
-
-        cmd = strategy._build_curl_command(method="POST", url="https://x", data=None)
-
-        assert "--data" not in cmd
-
-    def test_url_is_last_argument(self):
-        strategy = CurlStrategy()
-        strategy.configure(default_headers={"X-Default": "d"})
-
-        cmd = strategy._build_curl_command(
-            method="POST",
-            url="https://api.test/path",
-            headers={"X-Call": "c"},
-            data="body",
-            verify_ssl=False,
-        )
-
-        assert cmd[-1] == "https://api.test/path"
-
     def test_lowercase_method_is_uppercased(self):
         strategy = CurlStrategy()
 
@@ -344,7 +283,8 @@ def test_returns_parsed_response_on_first_success(self):
         assert cmd[0] == "curl"
         assert cmd[-1] == "https://api.test/path"
         assert "--insecure" in cmd
-        assert "--data" in cmd
+        assert "--data" not in cmd
+        assert 'data-raw = "payload"' in kwargs["input"]
         assert kwargs["timeout"] == 10
         assert kwargs["capture_output"] is True
         assert kwargs["text"] is True
@@ -452,3 +392,39 @@ def test_returns_false_when_curl_missing(self):
 class TestGetName:
     def test_returns_curl(self):
         assert CurlStrategy().get_name() == "curl"
+
+
+@pytest.mark.skipif(not shutil.which("curl"), reason="curl is required")
+@pytest.mark.parametrize("method", ["POST", "PUT", "PATCH"])
+def test_private_headers_and_body_reach_server_unchanged(local_endpoint, monkeypatch, method):
+    url, received = local_endpoint
+    strategy = CurlStrategy()
+    strategy.configure(default_headers={"X-Default": "synthetic-default", "X-Shared": "default"})
+    body = '@not-a-file\n"\nurl = "http://127.0.0.1:1/injected"\n# café\t\\end\r\n'
+    authorization = "hmac synthetic:signature:nonce:timestamp"
+    header_value = 'synthetic-"quoted"-\\value'
+    real_run = subprocess.run
+
+    def inspect_command(cmd, **kwargs):
+        for secret in (body, authorization, header_value, "synthetic-default"):
+            assert secret not in " ".join(cmd)
+        assert kwargs.get("shell", False) is False
+        return real_run(cmd, **kwargs)
+
+    monkeypatch.setattr(subprocess, "run", inspect_command)
+    response = strategy.request(
+        method,
+        url,
+        headers={"Authorization": authorization, "X-Test": header_value, "X-Shared": "per-call"},
+        data=body,
+        timeout=5,
+    )
+    assert response.status_code == 200
+    assert len(received) == 1
+    actual_method, headers, actual_body = received[0]
+    assert actual_method == method
+    assert actual_body == body.encode("utf-8")
+    assert headers["Authorization"] == authorization
+    assert headers["X-Test"] == header_value
+    assert headers["X-Default"] == "synthetic-default"
+    assert headers["X-Shared"] == "per-call"