From 8b4358af313e78df4a4f0025f8224c127ad8e57f Mon Sep 17 00:00:00 2001 From: Vildan Bina Date: Mon, 5 Oct 2026 15:15:00 +0200 Subject: [PATCH 1/2] fix(PYT-44): harden payment checks and sensitive data handling - Require explicit paid status in the low-level payment predicate - Pass curl credentials and request bodies through escaped stdin config - Omit log bodies when masking cannot safely process them - Cover security regressions and preserve high-level success flags --- buckaroo/http/client.py | 42 ++------ buckaroo/http/strategies/curl_strategy.py | 46 +++++---- buckaroo/observers/logging_observer.py | 6 +- .../strategies/test_curl_private_input.py | 96 +++++++++++++++++++ .../http/strategies/test_curl_strategy.py | 65 +------------ tests/unit/http/test_payment_success.py | 61 ++++++++++++ tests/unit/http/test_response.py | 56 ----------- tests/unit/observers/test_logging_observer.py | 50 +++++++--- 8 files changed, 234 insertions(+), 188 deletions(-) create mode 100644 tests/unit/http/strategies/test_curl_private_input.py create mode 100644 tests/unit/http/test_payment_success.py diff --git a/buckaroo/http/client.py b/buckaroo/http/client.py index 17f0f1c..51eb021 100644 --- a/buckaroo/http/client.py +++ b/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.""" diff --git a/buckaroo/http/strategies/curl_strategy.py b/buckaroo/http/strategies/curl_strategy.py index b20c8f5..1d0dd94 100644 --- a/buckaroo/http/strategies/curl_strategy.py +++ b/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. diff --git a/buckaroo/observers/logging_observer.py b/buckaroo/observers/logging_observer.py index 46e669e..8c3de1a 100644 --- a/buckaroo/observers/logging_observer.py +++ b/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, diff --git a/tests/unit/http/strategies/test_curl_private_input.py b/tests/unit/http/strategies/test_curl_private_input.py new file mode 100644 index 0000000..99a015f --- /dev/null +++ b/tests/unit/http/strategies/test_curl_private_input.py @@ -0,0 +1,96 @@ +"""Real curl requests preserve payloads without exposing them in argv.""" + +import base64 +import shutil +import subprocess +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import pytest + +from buckaroo.http.strategies import CurlStrategy +from buckaroo.services.hosted_fields_service import HostedFieldsService + + +@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) + + +@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" + + +@pytest.mark.skipif(not shutil.which("curl"), reason="curl is required") +def test_hosted_fields_basic_credentials_are_private(local_endpoint, monkeypatch): + url, received = local_endpoint + service = HostedFieldsService("synthetic-id", "synthetic-secret", http_strategy=CurlStrategy()) + monkeypatch.setattr(service, "OAUTH_TOKEN_URL", url) + authorization = "Basic " + base64.b64encode(b"synthetic-id:synthetic-secret").decode() + real_run = subprocess.run + + def inspect_command(cmd, **kwargs): + assert authorization not in " ".join(cmd) + return real_run(cmd, **kwargs) + + monkeypatch.setattr(subprocess, "run", inspect_command) + assert service.get_token() == {"access_token": "synthetic-token"} + assert received[0][1]["Authorization"] == authorization + assert received[0][2] == b"scope=hostedfields%3Asave&grant_type=client_credentials" diff --git a/tests/unit/http/strategies/test_curl_strategy.py b/tests/unit/http/strategies/test_curl_strategy.py index e04c0c2..bafe13f 100644 --- a/tests/unit/http/strategies/test_curl_strategy.py +++ b/tests/unit/http/strategies/test_curl_strategy.py @@ -94,68 +94,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 +282,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 diff --git a/tests/unit/http/test_payment_success.py b/tests/unit/http/test_payment_success.py new file mode 100644 index 0000000..a0cfd5c --- /dev/null +++ b/tests/unit/http/test_payment_success.py @@ -0,0 +1,61 @@ +"""Low-level payment verdicts and existing high-level response contracts.""" + +import json + +import pytest + +from buckaroo.http.client import BuckarooResponse +from buckaroo.http.strategies import HttpResponse + + +@pytest.mark.parametrize( + "status,expected", + [ + ({"Code": code}, code == 190) + for code in [190, 490, 491, 492, 690, 790, 791, 792, 793, 794, 890, 891, 999] + ] + + [({"Code": {"Code": code}}, code == 190) for code in [190, 490, 791]] + + [ + (status, False) + for status in [ + None, + {}, + [], + "invalid", + 190, + {"Code": None}, + {"Code": []}, + {"Code": "190"}, + {"Code": 190.0}, + ] + ], +) +def test_payment_success_requires_explicit_paid_status(status, expected): + data = {"Status": status, "RequiredAction": {"RedirectURL": "https://example.invalid/pay"}} + raw = HttpResponse(200, {}, json.dumps(data), True) + assert BuckarooResponse(raw).is_successful_payment() is expected + + +@pytest.mark.parametrize("body", ["", "{}", "null", "[]", '"invalid"']) +def test_missing_payment_status_is_not_paid(body): + assert BuckarooResponse(HttpResponse(200, {}, body, True)).is_successful_payment() is False + + +def test_http_failure_is_not_payment_success(): + from buckaroo.models.payment_response import PaymentResponse + + raw = HttpResponse(500, {}, '{"Status":{"Code":190}}', False) + response = BuckarooResponse(raw) + assert response.is_successful_payment() is False + assert PaymentResponse(response.to_dict()).is_successful() is False + + +@pytest.mark.parametrize("flag", [True, False]) +def test_high_level_response_preserves_explicit_success_flag(flag): + from buckaroo.models.payment_response import PaymentResponse + + # The high-level wrapper accepts a caller-supplied verdict independently + # of the transport response; keep that existing contract. + for envelope in [{}, {"status_code": 200, "success": True, "data": {"Status": {"Code": 190}}}]: + response = PaymentResponse({**envelope, "is_successful_payment": flag}) + assert response.is_successful() is flag diff --git a/tests/unit/http/test_response.py b/tests/unit/http/test_response.py index c151fcb..4f37b26 100644 --- a/tests/unit/http/test_response.py +++ b/tests/unit/http/test_response.py @@ -86,62 +86,6 @@ def test_headers_pass_through(self): assert response.headers == headers -class TestIsSuccessfulPayment: - def test_returns_false_when_http_failed(self): - response = BuckarooResponse( - make_response(status_code=500, text='{"Status": {"Code": 190}}') - ) - - assert response.is_successful_payment() is False - - @pytest.mark.parametrize("code", [190, 490, 491, 492, 790, 791, 792, 793]) - def test_true_for_each_buckaroo_success_code(self, code): - response = BuckarooResponse(make_response(text=f'{{"Status": {{"Code": {code}}}}}')) - - assert response.is_successful_payment() is True - - def test_false_for_non_success_buckaroo_code(self): - response = BuckarooResponse(make_response(text='{"Status": {"Code": 491000}}')) - - assert response.is_successful_payment() is False - - def test_handles_nested_code_dict_shape(self): - response = BuckarooResponse( - make_response(text='{"Status": {"Code": {"Code": 190, "Description": "Success"}}}') - ) - - assert response.is_successful_payment() is True - - def test_returns_success_when_no_status_field(self): - # HTTP 2xx but no "Status" in body — falls through to self.success. - response = BuckarooResponse(make_response(text='{"Other": "field"}')) - - assert response.is_successful_payment() is True - - def test_true_when_status_code_missing_from_status(self): - response = BuckarooResponse(make_response(text='{"Status": {"Other": 1}}')) - - # Status present but no "Code" key — falls through to self.success. - assert response.is_successful_payment() is True - - def test_false_when_code_is_unknown_type(self): - response = BuckarooResponse(make_response(text='{"Status": {"Code": "oops"}}')) - - assert response.is_successful_payment() is False - - def test_true_when_status_is_falsy(self): - # Status present but falsy — skips the Buckaroo-code branch. - response = BuckarooResponse(make_response(text='{"Status": null}')) - - assert response.is_successful_payment() is True - - def test_true_when_data_is_empty_but_http_ok(self): - # No _data at all -> falls through to self.success. - response = BuckarooResponse(make_response(text="")) - - assert response.is_successful_payment() is True - - class TestGetStatusCode: def test_returns_simple_int_code(self): response = BuckarooResponse(make_response(text='{"Status": {"Code": 190}}')) diff --git a/tests/unit/observers/test_logging_observer.py b/tests/unit/observers/test_logging_observer.py index 45dd62c..53952f1 100644 --- a/tests/unit/observers/test_logging_observer.py +++ b/tests/unit/observers/test_logging_observer.py @@ -170,9 +170,9 @@ def test_format_json_parses_json_string_and_masks(): assert parsed["amount"] == 10 -def test_format_json_non_json_string_returned_verbatim(): +def test_format_json_non_json_string_is_omitted(): obs = _observer() - assert obs._format_json("not json at all") == "not json at all" + assert obs._format_json("not json at all") == "***BODY_OMITTED***" def test_format_json_dict_input_produces_masked_json(): @@ -183,19 +183,21 @@ def test_format_json_dict_input_produces_masked_json(): assert parsed["currency"] == "EUR" -def test_format_json_non_serialisable_falls_back_to_str(): - class NotJSON: - def __repr__(self): - return "" +def test_format_json_non_serialisable_is_omitted(): + obs = _observer() + assert obs._format_json({object(): "value"}) == "***BODY_OMITTED***" + + +def test_opaque_object_body_is_omitted(caplog): + class OpaqueBody: + def __str__(self): + return "synthetic-secret" obs = _observer() - # json.dumps handles most things via default=str; force a failure by - # triggering an exception path — a dict with a non-serialisable key - # (keys must be str/int/float/bool/None) raises TypeError. - weird = {object(): "value"} - result = obs._format_json(weird) - assert isinstance(result, str) - assert result == str(weird) + with caplog.at_level(logging.INFO, logger="buckaroo_sdk"): + obs.log_response(200, body=OpaqueBody()) + assert "synthetic-secret" not in caplog.text + assert "***BODY_OMITTED***" in caplog.text # --- Case-insensitive substring matching --- @@ -692,3 +694,25 @@ def test_child_log_info_family_merges_context(caplog, method, expected_level): assert rec.levelno == expected_level assert "abc" in rec.message assert "hello" in rec.message + + +@pytest.mark.parametrize( + "body", + [ + "token=synthetic-secret&iban=NL00SYNTHETIC", + '{"token":"synthetic-secret"', + {1: "bad key", "token": "synthetic-secret"}, + ], +) +@pytest.mark.parametrize("direction", ["request", "response"]) +@pytest.mark.parametrize("mask", [True, False]) +def test_log_body_failure_paths_respect_masking(body, direction, mask, caplog): + observer = _observer(mask=mask) + with caplog.at_level(logging.INFO, logger="buckaroo_sdk"): + if direction == "request": + observer.log_request("POST", "https://example.invalid", body=body) + else: + observer.log_response(200, body=body) + assert ("synthetic-secret" in caplog.text) is (not mask) + if mask: + assert "***BODY_OMITTED***" in caplog.text From 25b139ae558a462e060cbc1667e1afb1fc52c90c Mon Sep 17 00:00:00 2001 From: Vildan Bina Date: Mon, 5 Oct 2026 15:23:20 +0200 Subject: [PATCH 2/2] tests(PYT-44): align security tests with source modules - Move regression tests into the existing HTTP, model and service test files - Share the loopback fixture in unit conftest and remove standalone test modules --- tests/unit/conftest.py | 33 +++++++ .../strategies/test_curl_private_input.py | 96 ------------------- .../http/strategies/test_curl_strategy.py | 37 +++++++ tests/unit/http/test_payment_success.py | 61 ------------ tests/unit/http/test_response.py | 41 ++++++++ tests/unit/models/test_payment_response.py | 9 ++ .../services/test_hosted_fields_service.py | 21 ++++ 7 files changed, 141 insertions(+), 157 deletions(-) delete mode 100644 tests/unit/http/strategies/test_curl_private_input.py delete mode 100644 tests/unit/http/test_payment_success.py diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 198fdae..f5e8d28 100644 --- a/tests/unit/conftest.py +++ b/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) diff --git a/tests/unit/http/strategies/test_curl_private_input.py b/tests/unit/http/strategies/test_curl_private_input.py deleted file mode 100644 index 99a015f..0000000 --- a/tests/unit/http/strategies/test_curl_private_input.py +++ /dev/null @@ -1,96 +0,0 @@ -"""Real curl requests preserve payloads without exposing them in argv.""" - -import base64 -import shutil -import subprocess -import threading -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer - -import pytest - -from buckaroo.http.strategies import CurlStrategy -from buckaroo.services.hosted_fields_service import HostedFieldsService - - -@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) - - -@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" - - -@pytest.mark.skipif(not shutil.which("curl"), reason="curl is required") -def test_hosted_fields_basic_credentials_are_private(local_endpoint, monkeypatch): - url, received = local_endpoint - service = HostedFieldsService("synthetic-id", "synthetic-secret", http_strategy=CurlStrategy()) - monkeypatch.setattr(service, "OAUTH_TOKEN_URL", url) - authorization = "Basic " + base64.b64encode(b"synthetic-id:synthetic-secret").decode() - real_run = subprocess.run - - def inspect_command(cmd, **kwargs): - assert authorization not in " ".join(cmd) - return real_run(cmd, **kwargs) - - monkeypatch.setattr(subprocess, "run", inspect_command) - assert service.get_token() == {"access_token": "synthetic-token"} - assert received[0][1]["Authorization"] == authorization - assert received[0][2] == b"scope=hostedfields%3Asave&grant_type=client_credentials" diff --git a/tests/unit/http/strategies/test_curl_strategy.py b/tests/unit/http/strategies/test_curl_strategy.py index bafe13f..dff8b6c 100644 --- a/tests/unit/http/strategies/test_curl_strategy.py +++ b/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 @@ -391,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" diff --git a/tests/unit/http/test_payment_success.py b/tests/unit/http/test_payment_success.py deleted file mode 100644 index a0cfd5c..0000000 --- a/tests/unit/http/test_payment_success.py +++ /dev/null @@ -1,61 +0,0 @@ -"""Low-level payment verdicts and existing high-level response contracts.""" - -import json - -import pytest - -from buckaroo.http.client import BuckarooResponse -from buckaroo.http.strategies import HttpResponse - - -@pytest.mark.parametrize( - "status,expected", - [ - ({"Code": code}, code == 190) - for code in [190, 490, 491, 492, 690, 790, 791, 792, 793, 794, 890, 891, 999] - ] - + [({"Code": {"Code": code}}, code == 190) for code in [190, 490, 791]] - + [ - (status, False) - for status in [ - None, - {}, - [], - "invalid", - 190, - {"Code": None}, - {"Code": []}, - {"Code": "190"}, - {"Code": 190.0}, - ] - ], -) -def test_payment_success_requires_explicit_paid_status(status, expected): - data = {"Status": status, "RequiredAction": {"RedirectURL": "https://example.invalid/pay"}} - raw = HttpResponse(200, {}, json.dumps(data), True) - assert BuckarooResponse(raw).is_successful_payment() is expected - - -@pytest.mark.parametrize("body", ["", "{}", "null", "[]", '"invalid"']) -def test_missing_payment_status_is_not_paid(body): - assert BuckarooResponse(HttpResponse(200, {}, body, True)).is_successful_payment() is False - - -def test_http_failure_is_not_payment_success(): - from buckaroo.models.payment_response import PaymentResponse - - raw = HttpResponse(500, {}, '{"Status":{"Code":190}}', False) - response = BuckarooResponse(raw) - assert response.is_successful_payment() is False - assert PaymentResponse(response.to_dict()).is_successful() is False - - -@pytest.mark.parametrize("flag", [True, False]) -def test_high_level_response_preserves_explicit_success_flag(flag): - from buckaroo.models.payment_response import PaymentResponse - - # The high-level wrapper accepts a caller-supplied verdict independently - # of the transport response; keep that existing contract. - for envelope in [{}, {"status_code": 200, "success": True, "data": {"Status": {"Code": 190}}}]: - response = PaymentResponse({**envelope, "is_successful_payment": flag}) - assert response.is_successful() is flag diff --git a/tests/unit/http/test_response.py b/tests/unit/http/test_response.py index 4f37b26..2760561 100644 --- a/tests/unit/http/test_response.py +++ b/tests/unit/http/test_response.py @@ -5,6 +5,8 @@ directly — no HTTP client, no mocks. """ +import json + import pytest from buckaroo.http.client import BuckarooResponse @@ -283,3 +285,42 @@ def test_includes_status_success_data_and_headers(self): "data": {"Key": "k"}, "headers": headers, } + + +@pytest.mark.parametrize( + "status,expected", + [ + ({"Code": code}, code == 190) + for code in [190, 490, 491, 492, 690, 790, 791, 792, 793, 794, 890, 891, 999] + ] + + [({"Code": {"Code": code}}, code == 190) for code in [190, 490, 791]] + + [ + (status, False) + for status in [ + None, + {}, + [], + "invalid", + 190, + {"Code": None}, + {"Code": []}, + {"Code": "190"}, + {"Code": 190.0}, + ] + ], +) +def test_payment_success_requires_explicit_paid_status(status, expected): + data = {"Status": status, "RequiredAction": {"RedirectURL": "https://example.invalid/pay"}} + raw = HttpResponse(200, {}, json.dumps(data), True) + assert BuckarooResponse(raw).is_successful_payment() is expected + + +@pytest.mark.parametrize("body", ["", "{}", "null", "[]", '"invalid"']) +def test_missing_payment_status_is_not_paid(body): + assert BuckarooResponse(HttpResponse(200, {}, body, True)).is_successful_payment() is False + + +def test_http_failure_is_not_payment_success(): + raw = HttpResponse(500, {}, '{"Status":{"Code":190}}', False) + response = BuckarooResponse(raw) + assert response.is_successful_payment() is False diff --git a/tests/unit/models/test_payment_response.py b/tests/unit/models/test_payment_response.py index 10aa80b..5d265c6 100644 --- a/tests/unit/models/test_payment_response.py +++ b/tests/unit/models/test_payment_response.py @@ -673,3 +673,12 @@ def test_has_error_false_when_request_errors_is_scalar(): assert response.has_error() is False assert response.get_first_error() == {} assert response.get_some_error() == "" + + +@pytest.mark.parametrize("flag", [True, False]) +def test_high_level_response_preserves_explicit_success_flag(flag): + # The high-level wrapper accepts a caller-supplied verdict independently + # of the transport response; keep that existing contract. + for envelope in [{}, {"status_code": 200, "success": True, "data": {"Status": {"Code": 190}}}]: + response = PaymentResponse({**envelope, "is_successful_payment": flag}) + assert response.is_successful() is flag diff --git a/tests/unit/services/test_hosted_fields_service.py b/tests/unit/services/test_hosted_fields_service.py index ab63942..f6c34a0 100644 --- a/tests/unit/services/test_hosted_fields_service.py +++ b/tests/unit/services/test_hosted_fields_service.py @@ -3,9 +3,12 @@ from __future__ import annotations import base64 +import shutil +import subprocess import pytest +from buckaroo.http.strategies import CurlStrategy from buckaroo.exceptions._buckaroo_error import BuckarooError from buckaroo.services.hosted_fields_service import HostedFieldsService from tests.support.mock_request import BuckarooMockRequest @@ -184,3 +187,21 @@ def test_rejects_empty_client_id(self, mock_strategy): def test_rejects_empty_client_secret(self, mock_strategy): with pytest.raises(ValueError): HostedFieldsService(client_id="x", client_secret="", http_strategy=mock_strategy) + + +@pytest.mark.skipif(not shutil.which("curl"), reason="curl is required") +def test_hosted_fields_basic_credentials_are_private(local_endpoint, monkeypatch): + url, received = local_endpoint + service = HostedFieldsService("synthetic-id", "synthetic-secret", http_strategy=CurlStrategy()) + monkeypatch.setattr(service, "OAUTH_TOKEN_URL", url) + authorization = "Basic " + base64.b64encode(b"synthetic-id:synthetic-secret").decode() + real_run = subprocess.run + + def inspect_command(cmd, **kwargs): + assert authorization not in " ".join(cmd) + return real_run(cmd, **kwargs) + + monkeypatch.setattr(subprocess, "run", inspect_command) + assert service.get_token() == {"access_token": "synthetic-token"} + assert received[0][1]["Authorization"] == authorization + assert received[0][2] == b"scope=hostedfields%3Asave&grant_type=client_credentials"