feat: add reply validators and hosted fields oauth service

Author: vildanbina

buckaroo/services/hosted_fields_service.py

@@ -0,0 +1,112 @@
+"""Hosted Fields token service for Buckaroo credit card tokenization.
+
+The Buckaroo Hosted Fields iframe tokenizes card data client-side, keeping
+PAN/CVV out of the merchant server. The iframe needs a short-lived JWT,
+minted via the OAuth2 ``client_credentials`` flow against
+``auth.buckaroo.io`` using the merchant's Hosted Fields client_id /
+client_secret pair (distinct from the Buckaroo store key / secret key
+used for HMAC-signed API calls).
+
+The PHP SDK does not include this service; OAuth is out of scope for the
+core transaction API. This module exists only to consolidate the Hosted
+Fields token exchange in one place.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+from typing import Any, Dict, Optional
+from urllib.parse import urlencode
+
+from ..exceptions._buckaroo_error import BuckarooError
+from ..http.strategies import HttpStrategy, HttpStrategyFactory
+
+
+class HostedFieldsService:
+    """Mint OAuth access tokens for the Buckaroo Hosted Fields iframe."""
+
+    OAUTH_TOKEN_URL = "https://auth.buckaroo.io/oauth/token"
+    GRANT_TYPE = "client_credentials"
+    DEFAULT_SCOPE = "hostedfields:save"
+
+    def __init__(
+        self,
+        client_id: str,
+        client_secret: str,
+        http_strategy: Optional[HttpStrategy] = None,
+        timeout: int = 10,
+    ) -> None:
+        if client_id is None or not client_id.strip():
+            raise ValueError("Client ID must be provided")
+        if client_secret is None or not client_secret.strip():
+            raise ValueError("Client secret must be provided")
+
+        self._client_id = client_id.strip()
+        self._client_secret = client_secret.strip()
+        self._timeout = timeout
+        self._injected_strategy = http_strategy
+        self._strategy: Optional[HttpStrategy] = None
+
+    def _get_strategy(self) -> HttpStrategy:
+        if self._injected_strategy is not None:
+            return self._injected_strategy
+        if self._strategy is None:
+            self._strategy = HttpStrategyFactory.create_strategy()
+            self._strategy.configure()
+        return self._strategy
+
+    def get_token(self, scope: Optional[str] = None) -> Dict[str, Any]:
+        """Mint a Hosted Fields access token. Returns the decoded JSON body."""
+        scope = scope or self.DEFAULT_SCOPE
+        credentials = base64.b64encode(
+            f"{self._client_id}:{self._client_secret}".encode("utf-8")
+        ).decode("ascii")
+        body = urlencode({"scope": scope, "grant_type": self.GRANT_TYPE})
+
+        try:
+            response = self._get_strategy().request(
+                method="POST",
+                url=self.OAUTH_TOKEN_URL,
+                headers={
+                    "Authorization": f"Basic {credentials}",
+                    "Content-Type": "application/x-www-form-urlencoded",
+                },
+                data=body,
+                timeout=self._timeout,
+            )
+        except Exception as exc:
+            if isinstance(exc, BuckarooError):
+                raise
+            raise BuckarooError(f"Hosted Fields token request failed: {exc}") from exc
+
+        if not response.success:
+            error = self._extract_error(response.text)
+            raise BuckarooError(
+                f"Hosted Fields token request failed (status {response.status_code}): {error}"
+            )
+
+        text = response.text or ""
+        if not text.strip():
+            raise BuckarooError("Hosted Fields token response was empty")
+        try:
+            return json.loads(text)
+        except json.JSONDecodeError as exc:
+            raise BuckarooError(
+                f"Failed to parse Hosted Fields token response JSON: {exc}"
+            ) from exc
+
+    @staticmethod
+    def _extract_error(text: Optional[str]) -> str:
+        """Pull RFC6749 error fields from the response without echoing the full body."""
+        if not text:
+            return "no body"
+        try:
+            data = json.loads(text)
+        except json.JSONDecodeError:
+            return "non-JSON body"
+        if not isinstance(data, dict):
+            return "unexpected body shape"
+        code = data.get("error") or "unknown_error"
+        description = data.get("error_description")
+        return f"{code}: {description}" if description else code

buckaroo/services/reply/__init__.py

@@ -0,0 +1,15 @@
+"""Reply (push notification) verification handlers.
+
+Mirrors the PHP SDK ``Handlers/Reply/`` folder. Pick the strategy that
+matches the wire format of the incoming Buckaroo push:
+
+* :class:`buckaroo.services.reply.http_post.HttpPost` — form-encoded pushes
+  (SHA-1 over ``brq_signature``).
+* :class:`buckaroo.services.reply.json_reply.Json` — JSON pushes (HMAC-SHA256
+  over the ``Authorization`` header).
+"""
+
+from .http_post import HttpPost
+from .json_reply import Json
+
+__all__ = ["HttpPost", "Json"]

buckaroo/services/reply/http_post.py

@@ -0,0 +1,58 @@
+"""Form-encoded push verifier (SHA-1 over ``brq_signature``).
+
+Mirrors PHP ``Handlers/Reply/HttpPost.php``. Buckaroo signs form-style push
+notifications with a SHA-1 digest over the sorted ``brq_*`` / ``add_*`` /
+``cust_*`` parameters concatenated with the merchant secret key, delivered
+in the ``brq_signature`` field of the form body.
+"""
+
+from __future__ import annotations
+
+import hmac
+from hashlib import sha1
+from typing import Mapping
+from urllib.parse import unquote_plus
+
+
+class HttpPost:
+    """Verify SHA-1 signatures on form-style Buckaroo pushes."""
+
+    SIGNATURE_FIELD = "brq_signature"
+    INCLUDE_PREFIXES = ("add_", "brq_", "cust_")
+
+    def __init__(self, secret_key: str) -> None:
+        if secret_key is None or not secret_key.strip():
+            raise ValueError("Secret key must be provided")
+        self.secret_key = secret_key.strip()
+
+    def validate(self, params: Mapping[str, object]) -> bool:
+        """Return True if ``params['brq_signature']`` matches the computed signature."""
+        provided = next(
+            (
+                v for k, v in params.items()
+                if k.lower() == self.SIGNATURE_FIELD
+            ),
+            None,
+        )
+        if not provided or not isinstance(provided, str):
+            return False
+        expected = self.compute_signature(params)
+        return hmac.compare_digest(provided, expected)
+
+    def compute_signature(self, params: Mapping[str, object]) -> str:
+        """Compute the expected SHA-1 signature for a form-style push."""
+        decoded = [
+            (k, unquote_plus(v) if isinstance(v, str) else v)
+            for k, v in params.items()
+            if k.lower() != self.SIGNATURE_FIELD
+        ]
+        filtered = [
+            (k, v) for k, v in decoded
+            if any(k.lower().startswith(p) for p in self.INCLUDE_PREFIXES)
+        ]
+        sorted_items = sorted(filtered, key=lambda pair: pair[0].lower())
+        sign_string = "".join(
+            f"{k}={v if v is not None else ''}" for k, v in sorted_items
+        )
+        sign_string += self.secret_key
+        return sha1(sign_string.encode("utf-8")).hexdigest()

buckaroo/services/reply/json_reply.py

@@ -0,0 +1,81 @@
+"""JSON push verifier (HMAC-SHA256 over the ``Authorization`` header).
+
+Mirrors PHP ``Handlers/Reply/Json.php`` (which delegates to
+``Handlers/HMAC/Validator.php``). Buckaroo signs JSON push notifications
+with an HMAC-SHA256 hash carried in the
+``Authorization: hmac key:hash:nonce:time`` header. Verification recomputes
+the hash from the URL, method, body, and merchant credentials, then
+compares constant-time.
+
+Module file is ``json_reply.py`` (not ``json.py``) to avoid shadowing the
+stdlib :mod:`json`. Class name is :class:`Json` for parity with the PHP SDK.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac as _hmac
+from typing import Optional, Union
+from urllib.parse import quote
+
+
+class Json:
+    """Verify HMAC-SHA256 signatures on JSON-style Buckaroo pushes."""
+
+    def __init__(self, store_key: str, secret_key: str) -> None:
+        if store_key is None or not store_key.strip():
+            raise ValueError("Store key must be provided")
+        if secret_key is None or not secret_key.strip():
+            raise ValueError("Secret key must be provided")
+        self.store_key = store_key.strip()
+        self.secret_key = secret_key.strip()
+
+    def validate(
+        self,
+        authorization: Optional[str],
+        uri: str,
+        method: str,
+        body: Union[str, bytes, None] = "",
+    ) -> bool:
+        """Return True if ``authorization`` is a valid HMAC for the request."""
+        if not authorization:
+            return False
+
+        parts = authorization.split(":")
+        if len(parts) != 4:
+            return False
+        provided_hash = parts[1]
+        nonce = parts[2]
+        timestamp = parts[3]
+
+        content_b64 = self._md5_b64(body)
+        encoded_url = self._encode_url(uri)
+        signing_string = (
+            f"{self.store_key}{method}{encoded_url}{timestamp}{nonce}{content_b64}"
+        )
+        expected = base64.b64encode(
+            _hmac.new(
+                self.secret_key.encode("utf-8"),
+                signing_string.encode("utf-8"),
+                hashlib.sha256,
+            ).digest()
+        ).decode("ascii")
+
+        return _hmac.compare_digest(provided_hash, expected)
+
+    @staticmethod
+    def _md5_b64(body: Union[str, bytes, None]) -> str:
+        if not body:
+            return ""
+        if isinstance(body, str):
+            body = body.encode("utf-8")
+        return base64.b64encode(hashlib.md5(body).digest()).decode("ascii")
+
+    @staticmethod
+    def _encode_url(url: str) -> str:
+        if url.startswith("https://"):
+            url = url[8:]
+        elif url.startswith("http://"):
+            url = url[7:]
+        return quote(url, safe="").lower()

tests/unit/services/reply/__init__.py

@@ -0,0 +1,106 @@
+"""Tests for :class:`buckaroo.services.reply.http_post.HttpPost`."""
+
+from __future__ import annotations
+
+from hashlib import sha1
+
+import pytest
+
+from buckaroo.services.reply.http_post import HttpPost
+
+
+SECRET = "secretkey"
+
+
+def _sign(params: dict, secret: str = SECRET) -> str:
+    """Reference SHA-1 signature per Buckaroo's documented algorithm."""
+    items = [(k, v) for k, v in params.items() if k.lower() != "brq_signature"]
+    filtered = [
+        (k, v) for k, v in items
+        if any(k.lower().startswith(p) for p in ("add_", "brq_", "cust_"))
+    ]
+    sorted_items = sorted(filtered, key=lambda pair: pair[0].lower())
+    sign_string = "".join(f"{k}={v if v is not None else ''}" for k, v in sorted_items)
+    sign_string += secret
+    return sha1(sign_string.encode("utf-8")).hexdigest()
+
+
+class TestComputeSignature:
+    def test_sorts_keys_case_insensitively(self):
+        h = HttpPost(SECRET)
+        params = {"brq_b": "2", "BRQ_a": "1"}
+        assert h.compute_signature(params) == sha1(
+            f"BRQ_a=1brq_b=2{SECRET}".encode("utf-8")
+        ).hexdigest()
+
+    def test_excludes_brq_signature_field(self):
+        h = HttpPost(SECRET)
+        params = {"brq_x": "1", "brq_signature": "deadbeef"}
+        expected = sha1(f"brq_x=1{SECRET}".encode("utf-8")).hexdigest()
+        assert h.compute_signature(params) == expected
+
+    def test_filters_to_brq_add_cust_prefixes_only(self):
+        h = HttpPost(SECRET)
+        params = {"brq_a": "1", "add_b": "2", "cust_c": "3", "other_d": "4"}
+        expected = sha1(f"add_b=2brq_a=1cust_c=3{SECRET}".encode("utf-8")).hexdigest()
+        assert h.compute_signature(params) == expected
+
+    def test_url_decodes_values(self):
+        h = HttpPost(SECRET)
+        params = {"brq_x": "hello+world"}
+        expected = sha1(f"brq_x=hello world{SECRET}".encode("utf-8")).hexdigest()
+        assert h.compute_signature(params) == expected
+
+    def test_handles_empty_value(self):
+        h = HttpPost(SECRET)
+        params = {"brq_x": ""}
+        expected = sha1(f"brq_x={SECRET}".encode("utf-8")).hexdigest()
+        assert h.compute_signature(params) == expected
+
+
+class TestValidate:
+    def test_returns_true_for_valid_signature(self):
+        h = HttpPost(SECRET)
+        params = {"brq_amount": "10.00", "brq_invoicenumber": "INV-1"}
+        params["brq_signature"] = _sign(params)
+        assert h.validate(params) is True
+
+    def test_returns_false_for_invalid_signature(self):
+        h = HttpPost(SECRET)
+        params = {"brq_amount": "10.00", "brq_signature": "not-a-real-sig"}
+        assert h.validate(params) is False
+
+    def test_returns_false_when_signature_missing(self):
+        h = HttpPost(SECRET)
+        assert h.validate({"brq_amount": "10.00"}) is False
+
+    def test_returns_false_when_signature_empty(self):
+        h = HttpPost(SECRET)
+        assert h.validate({"brq_amount": "10.00", "brq_signature": ""}) is False
+
+    def test_signature_field_lookup_is_case_insensitive(self):
+        h = HttpPost(SECRET)
+        params = {"brq_amount": "10.00"}
+        params["BRQ_SIGNATURE"] = _sign(params)
+        assert h.validate(params) is True
+
+    def test_uses_constant_time_compare(self):
+        h = HttpPost(SECRET)
+        params = {"brq_amount": "10.00"}
+        valid = _sign(params)
+        params["brq_signature"] = "0" * len(valid)
+        assert h.validate(params) is False
+
+
+class TestConstruction:
+    def test_rejects_empty_secret_key(self):
+        with pytest.raises(ValueError):
+            HttpPost("")
+
+    def test_rejects_whitespace_only_secret_key(self):
+        with pytest.raises(ValueError):
+            HttpPost("   ")
+
+    def test_strips_secret_key(self):
+        h = HttpPost("  sec  ")
+        assert h.secret_key == "sec"