Script 'mail_helper' called by obssrc Hello community, here is the log from the commit of package python-altcha for openSUSE:Factory checked in at 2026-08-05 17:49:40 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ Comparing /work/SRC/openSUSE:Factory/python-altcha (Old) and /work/SRC/openSUSE:Factory/.python-altcha.new.16738 (New) ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Package is "python-altcha" Wed Aug 5 17:49:40 2026 rev:8 rq:1369601 version:2.1.0 Changes: -------- --- /work/SRC/openSUSE:Factory/python-altcha/python-altcha.changes 2026-06-30 15:12:58.559112544 +0200 +++ /work/SRC/openSUSE:Factory/.python-altcha.new.16738/python-altcha.changes 2026-08-05 17:50:36.210487048 +0200 @@ -1,0 +2,7 @@ +Tue Aug 4 22:10:18 UTC 2026 - Dirk Müller <[email protected]> + +- update to 2.1.0: + * feat: verify_server function for remote server verification + * fix: enforce key_prefix in PoW v2 fallback verification + +------------------------------------------------------------------- Old: ---- altcha-2.0.2.tar.gz New: ---- altcha-2.1.0.tar.gz ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ Other differences: ------------------ ++++++ python-altcha.spec ++++++ --- /var/tmp/diff_new_pack.oUcBBU/_old 2026-08-05 17:50:36.766506499 +0200 +++ /var/tmp/diff_new_pack.oUcBBU/_new 2026-08-05 17:50:36.766506499 +0200 @@ -17,7 +17,7 @@ Name: python-altcha -Version: 2.0.2 +Version: 2.1.0 Release: 0 Summary: A library for creating and verifying challenges for ALTCHA License: MIT @@ -32,8 +32,9 @@ %python_subpackages %description -The ALTCHA Python Library is a lightweight, zero-dependency library designed for creating and verifying -[ALTCHA](https://altcha.org) challenges, specifically tailored for Python applications. +The ALTCHA Python Library is a lightweight, zero-dependency library designed +for creating and verifying [ALTCHA](https://altcha.org) challenges, +specifically tailored for Python applications. %prep %autosetup -p1 -n altcha-%{version} ++++++ altcha-2.0.2.tar.gz -> altcha-2.1.0.tar.gz ++++++ diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/altcha-2.0.2/PKG-INFO new/altcha-2.1.0/PKG-INFO --- old/altcha-2.0.2/PKG-INFO 2026-06-20 12:15:40.391074000 +0200 +++ new/altcha-2.1.0/PKG-INFO 2026-07-27 09:45:51.430548200 +0200 @@ -1,6 +1,6 @@ Metadata-Version: 2.4 Name: altcha -Version: 2.0.2 +Version: 2.1.0 Summary: A library for creating and verifying challenges for ALTCHA. Home-page: https://github.com/altcha-org/altcha-lib-py Author: Daniel Regeci @@ -95,7 +95,7 @@ # Server: verify result = verify_solution(payload_b64, HMAC_SECRET) -print(result.verified) # True +print(result.verified) # True ``` ### Deterministic mode @@ -136,7 +136,8 @@ challenge = create_challenge( algorithm="PBKDF2/SHA-256", cost=5_000, - expires_at=datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(minutes=10), + expires_at=datetime.datetime.now(datetime.timezone.utc) + + datetime.timedelta(minutes=10), hmac_secret=HMAC_SECRET, ) ``` @@ -146,8 +147,8 @@ Pass your own `derive_key` function to use a custom or third-party KDF: ```python -def my_derive_key(parameters, salt: bytes, password: bytes) -> bytes: - ... +def my_derive_key(parameters, salt: bytes, password: bytes) -> bytes: ... + challenge = create_challenge( algorithm="MY-ALGO", @@ -256,6 +257,48 @@ Verifies an ALTCHA server signature. +#### `verify_server(payload, url, secret, *, headers, timeout, retries, retry_delay, retry_backoff, http_post) → VerifyServerResult` + +Verifies a payload remotely via the ALTCHA Sentinel `/v1/verify/signature` API, instead of +checking the HMAC signature locally. Avoids managing the HMAC secret on your server, at the +cost of a network round-trip. + +```python +from altcha import verify_server + +result = verify_server( + payload, # the payload received from POST /v1/verify + url="https://sentinel.example.com/v1/verify/signature";, + secret=API_KEY_SECRET, # optional, checked against the payload's API key + timeout=10, + retries=2, +) + +if result.verified: + ... +``` + +| Parameter | Type | Default | Description | +|---|---|---|---| +| `payload` | `str` \| `ServerSignaturePayload` \| `dict` | — | The payload to verify. | +| `url` | `str` | — | Full URL of the Sentinel `/v1/verify/signature` endpoint. | +| `secret` | `str` | `None` | API key secret, checked against the payload's API key. | +| `headers` | `dict` | `None` | Additional headers to send with the request. | +| `timeout` | `float` | `10` | Per-attempt request timeout in seconds. | +| `retries` | `int` | `0` | Number of retry attempts after the first try. | +| `retry_delay` | `float` | `0.3` | Base delay in seconds between retries. | +| `retry_backoff` | `str` | `'exponential'` | `'fixed'` or `'exponential'` backoff. | +| `http_post` | callable | stdlib `urllib` | Transport override: `(url, body, headers, timeout) -> (status, body)`. | + +Returns `VerifyServerResult` with fields: + +| Field | Type | Description | +|---|---|---| +| `verified` | `bool` | `True` if Sentinel confirmed the payload is valid. | +| `reason` | `str \| None` | Failure reason, if any. | +| `api_key` | `str \| None` | The API key associated with the payload, if returned. | +| `verification_data` | `dict \| None` | Parsed verification data, if returned. | + --- ### V1 (legacy) diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/altcha-2.0.2/README.md new/altcha-2.1.0/README.md --- old/altcha-2.0.2/README.md 2026-06-20 12:15:36.000000000 +0200 +++ new/altcha-2.1.0/README.md 2026-07-27 09:45:45.000000000 +0200 @@ -72,7 +72,7 @@ # Server: verify result = verify_solution(payload_b64, HMAC_SECRET) -print(result.verified) # True +print(result.verified) # True ``` ### Deterministic mode @@ -113,7 +113,8 @@ challenge = create_challenge( algorithm="PBKDF2/SHA-256", cost=5_000, - expires_at=datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(minutes=10), + expires_at=datetime.datetime.now(datetime.timezone.utc) + + datetime.timedelta(minutes=10), hmac_secret=HMAC_SECRET, ) ``` @@ -123,8 +124,8 @@ Pass your own `derive_key` function to use a custom or third-party KDF: ```python -def my_derive_key(parameters, salt: bytes, password: bytes) -> bytes: - ... +def my_derive_key(parameters, salt: bytes, password: bytes) -> bytes: ... + challenge = create_challenge( algorithm="MY-ALGO", @@ -233,6 +234,48 @@ Verifies an ALTCHA server signature. +#### `verify_server(payload, url, secret, *, headers, timeout, retries, retry_delay, retry_backoff, http_post) → VerifyServerResult` + +Verifies a payload remotely via the ALTCHA Sentinel `/v1/verify/signature` API, instead of +checking the HMAC signature locally. Avoids managing the HMAC secret on your server, at the +cost of a network round-trip. + +```python +from altcha import verify_server + +result = verify_server( + payload, # the payload received from POST /v1/verify + url="https://sentinel.example.com/v1/verify/signature";, + secret=API_KEY_SECRET, # optional, checked against the payload's API key + timeout=10, + retries=2, +) + +if result.verified: + ... +``` + +| Parameter | Type | Default | Description | +|---|---|---|---| +| `payload` | `str` \| `ServerSignaturePayload` \| `dict` | — | The payload to verify. | +| `url` | `str` | — | Full URL of the Sentinel `/v1/verify/signature` endpoint. | +| `secret` | `str` | `None` | API key secret, checked against the payload's API key. | +| `headers` | `dict` | `None` | Additional headers to send with the request. | +| `timeout` | `float` | `10` | Per-attempt request timeout in seconds. | +| `retries` | `int` | `0` | Number of retry attempts after the first try. | +| `retry_delay` | `float` | `0.3` | Base delay in seconds between retries. | +| `retry_backoff` | `str` | `'exponential'` | `'fixed'` or `'exponential'` backoff. | +| `http_post` | callable | stdlib `urllib` | Transport override: `(url, body, headers, timeout) -> (status, body)`. | + +Returns `VerifyServerResult` with fields: + +| Field | Type | Description | +|---|---|---| +| `verified` | `bool` | `True` if Sentinel confirmed the payload is valid. | +| `reason` | `str \| None` | Failure reason, if any. | +| `api_key` | `str \| None` | The API key associated with the payload, if returned. | +| `verification_data` | `dict \| None` | Parsed verification data, if returned. | + --- ### V1 (legacy) diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/altcha-2.0.2/altcha/__init__.py new/altcha-2.1.0/altcha/__init__.py --- old/altcha-2.0.2/altcha/__init__.py 2026-06-20 12:15:36.000000000 +0200 +++ new/altcha-2.1.0/altcha/__init__.py 2026-07-27 09:45:45.000000000 +0200 @@ -11,6 +11,8 @@ from .v2 import VerifyServerSignatureResult as VerifyServerSignatureResult from .v2 import parse_verification_data as parse_verification_data from .v2 import verify_server_signature as verify_server_signature +from .v2 import VerifyServerResult as VerifyServerResult +from .v2 import verify_server as verify_server from .v2 import create_challenge as create_challenge from .v2 import solve_challenge as solve_challenge from .v2 import verify_solution as verify_solution diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/altcha-2.0.2/altcha/v2.py new/altcha-2.1.0/altcha/v2.py --- old/altcha-2.0.2/altcha/v2.py 2026-06-20 12:15:36.000000000 +0200 +++ new/altcha-2.1.0/altcha/v2.py 2026-07-27 09:45:45.000000000 +0200 @@ -8,7 +8,9 @@ import secrets import struct import time +import urllib.error import urllib.parse +import urllib.request from typing import Callable, Literal import datetime @@ -663,7 +665,8 @@ verified=valid, ) - # 4b. Slow path: re-derive the key from the counter and compare. + # 4b. Slow path: re-derive the key from the counter and compare, and + # require it to satisfy the signed key prefix. if derive_key is None: derive_key = _select_derive_key(params.algorithm) @@ -672,7 +675,9 @@ password = _make_password(nonce_bytes, solution.counter) recomputed = derive_key(params, salt_bytes, password) recomputed_hex = recomputed.hex() - invalid = not _constant_time_equal(recomputed_hex, solution.derived_key) + key_matches = _constant_time_equal(recomputed_hex, solution.derived_key) + prefix_matches = recomputed_hex.startswith(params.key_prefix) + invalid = not (key_matches and prefix_matches) return VerifySolutionResult( expired=False, @@ -714,6 +719,14 @@ self.verification_data = verification_data self.verified = verified + def to_dict(self) -> dict: + return { + "algorithm": self.algorithm, + "signature": self.signature, + "verificationData": self.verification_data, + "verified": self.verified, + } + @classmethod def from_dict(cls, d: dict) -> ServerSignaturePayload: return cls( @@ -859,3 +872,141 @@ verified=verified, verification_data=verification_data if verified else None, ) + + +# --------------------------------------------------------------------------- +# Remote (Sentinel API) verification +# --------------------------------------------------------------------------- + + +class VerifyServerResult: + """ + Result of a remote Sentinel verification via :func:`verify_server`. + + Attributes: + verified: ``True`` if Sentinel confirmed the payload is valid. + reason: Failure reason (e.g. an error code or exception message), if any. + api_key: The API key associated with the payload, if returned. + verification_data: Parsed verification data, if returned. + """ + + def __init__( + self, + verified: bool, + reason: str | None = None, + api_key: str | None = None, + verification_data: dict | None = None, + ): + self.verified = verified + self.reason = reason + self.api_key = api_key + self.verification_data = verification_data + + @classmethod + def from_dict(cls, d: dict) -> VerifyServerResult: + return cls( + verified=bool(d.get("verified", False)), + reason=d.get("reason"), + api_key=d.get("apiKey"), + verification_data=d.get("verificationData"), + ) + + +def _default_http_post( + url: str, data: bytes, headers: dict[str, str], timeout: float +) -> tuple[int, bytes]: + """Minimal stdlib HTTP POST, used as the default transport for :func:`verify_server`.""" + req = urllib.request.Request(url, data=data, method="POST", headers=headers) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.status, resp.read() + except urllib.error.HTTPError as e: + return e.code, e.read() + + +def _sleep_backoff(attempt: int, delay: float, backoff: str) -> None: + wait = delay if backoff == "fixed" else delay * (2**attempt) + time.sleep(wait) + + +def verify_server( + payload: str | ServerSignaturePayload | dict, + url: str, + secret: str | None = None, + *, + headers: dict[str, str] | None = None, + timeout: float = 10.0, + retries: int = 0, + retry_delay: float = 0.3, + retry_backoff: Literal["fixed", "exponential"] = "exponential", + http_post: Callable[[str, bytes, dict[str, str], float], tuple[int, bytes]] + | None = None, +) -> VerifyServerResult: + """ + Verify a payload remotely via the ALTCHA Sentinel ``/v1/verify/signature`` API. + + Instead of verifying the HMAC signature locally (see :func:`verify_server_signature`), + this POSTs the payload to Sentinel and trusts its response. Avoids managing the HMAC + secret on your server, at the cost of a network round-trip. + + Args: + payload: The payload to verify, as received from ``POST /v1/verify`` — a base64 + string, a :class:`ServerSignaturePayload`, or an already-decoded dict. + url: Full URL of the Sentinel ``/v1/verify/signature`` endpoint. + secret: API key secret. If given, Sentinel checks it against the payload's API key. + headers: Additional headers to send with the request. + timeout: Per-attempt request timeout in seconds. Defaults to ``10``. + retries: Number of retry attempts after the first try. Defaults to ``0``. + retry_delay: Base delay in seconds between retries. Defaults to ``0.3``. + retry_backoff: ``'fixed'`` or ``'exponential'`` backoff applied to *retry_delay*. + http_post: Transport override — ``(url, body, headers, timeout) -> (status, body)``. + Defaults to a stdlib ``urllib.request``-based implementation. Swap in another + HTTP client (e.g. ``requests``, ``httpx``) here if preferred. + + Returns: + A :class:`VerifyServerResult` describing the outcome. Failures (HTTP errors, network + errors, or a negative Sentinel verdict) are reported via ``verified=False`` with a + ``reason``, never raised as exceptions. + """ + if isinstance(payload, ServerSignaturePayload): + payload_value: str | dict = payload.to_dict() + else: + payload_value = payload + + body: dict = {"payload": payload_value} + if secret is not None: + body["secret"] = secret + data = json.dumps(body).encode() + + req_headers = {"Content-Type": "application/json", **(headers or {})} + post = http_post or _default_http_post + + attempts = retries + 1 + for attempt in range(attempts): + try: + status, resp_body = post(url, data, req_headers, timeout) + except Exception as e: + if attempt >= attempts - 1: + return VerifyServerResult( + verified=False, reason=str(e) or type(e).__name__ + ) + _sleep_backoff(attempt, retry_delay, retry_backoff) + continue + + if status == 400: + try: + err_data = json.loads(resp_body.decode()) + except Exception: + err_data = None + reason = (err_data or {}).get("error") or f"HTTP_{status}" + return VerifyServerResult(verified=False, reason=reason) + + if not (200 <= status < 300): + if attempt >= attempts - 1: + return VerifyServerResult(verified=False, reason=f"HTTP_{status}") + _sleep_backoff(attempt, retry_delay, retry_backoff) + continue + + return VerifyServerResult.from_dict(json.loads(resp_body.decode())) + + return VerifyServerResult(verified=False, reason="NETWORK_ERROR") diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/altcha-2.0.2/altcha.egg-info/PKG-INFO new/altcha-2.1.0/altcha.egg-info/PKG-INFO --- old/altcha-2.0.2/altcha.egg-info/PKG-INFO 2026-06-20 12:15:40.000000000 +0200 +++ new/altcha-2.1.0/altcha.egg-info/PKG-INFO 2026-07-27 09:45:51.000000000 +0200 @@ -1,6 +1,6 @@ Metadata-Version: 2.4 Name: altcha -Version: 2.0.2 +Version: 2.1.0 Summary: A library for creating and verifying challenges for ALTCHA. Home-page: https://github.com/altcha-org/altcha-lib-py Author: Daniel Regeci @@ -95,7 +95,7 @@ # Server: verify result = verify_solution(payload_b64, HMAC_SECRET) -print(result.verified) # True +print(result.verified) # True ``` ### Deterministic mode @@ -136,7 +136,8 @@ challenge = create_challenge( algorithm="PBKDF2/SHA-256", cost=5_000, - expires_at=datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(minutes=10), + expires_at=datetime.datetime.now(datetime.timezone.utc) + + datetime.timedelta(minutes=10), hmac_secret=HMAC_SECRET, ) ``` @@ -146,8 +147,8 @@ Pass your own `derive_key` function to use a custom or third-party KDF: ```python -def my_derive_key(parameters, salt: bytes, password: bytes) -> bytes: - ... +def my_derive_key(parameters, salt: bytes, password: bytes) -> bytes: ... + challenge = create_challenge( algorithm="MY-ALGO", @@ -256,6 +257,48 @@ Verifies an ALTCHA server signature. +#### `verify_server(payload, url, secret, *, headers, timeout, retries, retry_delay, retry_backoff, http_post) → VerifyServerResult` + +Verifies a payload remotely via the ALTCHA Sentinel `/v1/verify/signature` API, instead of +checking the HMAC signature locally. Avoids managing the HMAC secret on your server, at the +cost of a network round-trip. + +```python +from altcha import verify_server + +result = verify_server( + payload, # the payload received from POST /v1/verify + url="https://sentinel.example.com/v1/verify/signature";, + secret=API_KEY_SECRET, # optional, checked against the payload's API key + timeout=10, + retries=2, +) + +if result.verified: + ... +``` + +| Parameter | Type | Default | Description | +|---|---|---|---| +| `payload` | `str` \| `ServerSignaturePayload` \| `dict` | — | The payload to verify. | +| `url` | `str` | — | Full URL of the Sentinel `/v1/verify/signature` endpoint. | +| `secret` | `str` | `None` | API key secret, checked against the payload's API key. | +| `headers` | `dict` | `None` | Additional headers to send with the request. | +| `timeout` | `float` | `10` | Per-attempt request timeout in seconds. | +| `retries` | `int` | `0` | Number of retry attempts after the first try. | +| `retry_delay` | `float` | `0.3` | Base delay in seconds between retries. | +| `retry_backoff` | `str` | `'exponential'` | `'fixed'` or `'exponential'` backoff. | +| `http_post` | callable | stdlib `urllib` | Transport override: `(url, body, headers, timeout) -> (status, body)`. | + +Returns `VerifyServerResult` with fields: + +| Field | Type | Description | +|---|---|---| +| `verified` | `bool` | `True` if Sentinel confirmed the payload is valid. | +| `reason` | `str \| None` | Failure reason, if any. | +| `api_key` | `str \| None` | The API key associated with the payload, if returned. | +| `verification_data` | `dict \| None` | Parsed verification data, if returned. | + --- ### V1 (legacy) diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/altcha-2.0.2/setup.py new/altcha-2.1.0/setup.py --- old/altcha-2.0.2/setup.py 2026-06-20 12:15:36.000000000 +0200 +++ new/altcha-2.1.0/setup.py 2026-07-27 09:45:45.000000000 +0200 @@ -2,7 +2,7 @@ setup( name="altcha", - version="2.0.2", + version="2.1.0", description="A library for creating and verifying challenges for ALTCHA.", long_description=open("README.md").read(), long_description_content_type="text/markdown", diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/altcha-2.0.2/tests/test_altcha_v2.py new/altcha-2.1.0/tests/test_altcha_v2.py --- old/altcha-2.0.2/tests/test_altcha_v2.py 2026-06-20 12:15:36.000000000 +0200 +++ new/altcha-2.1.0/tests/test_altcha_v2.py 2026-07-27 09:45:45.000000000 +0200 @@ -1,8 +1,12 @@ import datetime +import json import struct import unittest +import unittest.mock from altcha.v2 import ( + DEFAULT_HMAC_ALGORITHM, + Challenge, ChallengeParameters, Payload, ServerSignaturePayload, @@ -10,12 +14,14 @@ _canonical_json, _hmac_v2, _make_password, + _sign_challenge_v2, create_challenge, derive_key_pbkdf2, derive_key_scrypt, derive_key_sha, parse_verification_data, solve_challenge, + verify_server, verify_server_signature, verify_solution, ) @@ -307,6 +313,42 @@ self.assertFalse(result.verified) self.assertTrue(result.invalid_solution) + def test_slow_path_enforces_key_prefix(self): + # Regression test: the fallback (no key signature) verification path must + # reject a solution whose derived key is genuinely correct for its counter + # but does not satisfy the challenge's key_prefix. Previously only + # derived_key == KDF(counter) was checked, letting a client submit any + # counter after a single KDF execution and skip the prefix search entirely. + ch = create_challenge("SHA-256", cost=10, hmac_secret=HMAC_KEY) + + # Learn the honest KDF output for counter 0 with exactly one hash + # computation: solve a probe copy of the challenge whose key_prefix is "" + # (matches immediately, no search). + probe_params = ChallengeParameters( + **{**ch.parameters.__dict__, "key_prefix": ""} + ) + probe_ch = Challenge(parameters=probe_params, signature=None) + honest = solve_challenge(probe_ch) + assert honest is not None + self.assertEqual(honest.counter, 0) + + # Pick a key_prefix the honest key is guaranteed not to satisfy: a byte + # can't be both 0x00 and 0xff. + mismatched_prefix = "ff" if honest.derived_key.startswith("00") else "00" + ch.parameters.key_prefix = mismatched_prefix + signed = _sign_challenge_v2( + DEFAULT_HMAC_ALGORITHM, ch.parameters, None, HMAC_KEY + ) + + # Submit the honestly-derived key/counter pair (one KDF execution, no + # prefix search) against the challenge whose signed key_prefix it does + # not satisfy. + bad_sol = Solution(counter=honest.counter, derived_key=honest.derived_key) + payload = Payload(signed, bad_sol).to_base64() + result = verify_solution(payload, HMAC_KEY) + self.assertFalse(result.verified) + self.assertTrue(result.invalid_solution) + def test_payload_object(self): ch = create_challenge("SHA-256", cost=1, counter=2, hmac_secret=HMAC_KEY) sol = solve_challenge(ch) @@ -464,5 +506,148 @@ self.assertTrue(result.invalid_signature) +class TestVerifyServer(unittest.TestCase): + URL = "https://sentinel.example.com/v1/verify/signature"; + + def setUp(self): + self._sleep_calls: list[float] = [] + self._sleep_patcher = unittest.mock.patch( + "altcha.v2.time.sleep", side_effect=self._sleep_calls.append + ) + self._sleep_patcher.start() + self.addCleanup(self._sleep_patcher.stop) + + def test_success(self): + result_data = { + "apiKey": "key_1", + "verificationData": {"verified": True}, + "verified": True, + } + calls = [] + + def post(url, data, headers, timeout): + calls.append((url, data, headers, timeout)) + return 200, json.dumps(result_data).encode() + + result = verify_server("payload", self.URL, http_post=post) + self.assertTrue(result.verified) + self.assertEqual(result.api_key, "key_1") + self.assertEqual(result.verification_data, {"verified": True}) + self.assertIsNone(result.reason) + self.assertEqual(len(calls), 1) + + def test_secret_included_in_body(self): + calls = [] + + def post(url, data, headers, timeout): + calls.append(data) + return 200, json.dumps({"verified": True}).encode() + + verify_server("payload", self.URL, secret="sec_123", http_post=post) + self.assertEqual( + json.loads(calls[0]), {"payload": "payload", "secret": "sec_123"} + ) + + def test_secret_omitted_when_not_given(self): + calls = [] + + def post(url, data, headers, timeout): + calls.append(data) + return 200, json.dumps({"verified": True}).encode() + + verify_server("payload", self.URL, http_post=post) + self.assertEqual(json.loads(calls[0]), {"payload": "payload"}) + + def test_400_is_terminal_not_retried(self): + calls = [] + + def post(url, data, headers, timeout): + calls.append(1) + return 400, json.dumps({"error": "INVALID_PAYLOAD"}).encode() + + result = verify_server("payload", self.URL, retries=3, http_post=post) + self.assertFalse(result.verified) + self.assertEqual(result.reason, "INVALID_PAYLOAD") + self.assertEqual(len(calls), 1) + + def test_network_error_retried_then_fails(self): + calls = [] + + def post(url, data, headers, timeout): + calls.append(1) + raise OSError("fetch failed") + + result = verify_server("payload", self.URL, retries=2, http_post=post) + self.assertFalse(result.verified) + self.assertEqual(result.reason, "fetch failed") + self.assertEqual(len(calls), 3) + + def test_network_error_then_success(self): + calls = [] + + def post(url, data, headers, timeout): + calls.append(1) + if len(calls) < 2: + raise OSError("fetch failed") + return 200, json.dumps({"verified": True}).encode() + + result = verify_server("payload", self.URL, retries=2, http_post=post) + self.assertTrue(result.verified) + self.assertEqual(len(calls), 2) + + def test_server_signature_payload_serialized(self): + payload = ServerSignaturePayload( + algorithm="SHA-256", + signature="sig", + verification_data="verified=true", + verified=True, + ) + calls = [] + + def post(url, data, headers, timeout): + calls.append(data) + return 200, json.dumps({"verified": True}).encode() + + verify_server(payload, self.URL, http_post=post) + sent = json.loads(calls[0]) + self.assertEqual( + sent["payload"], + { + "algorithm": "SHA-256", + "signature": "sig", + "verificationData": "verified=true", + "verified": True, + }, + ) + + def test_fixed_backoff(self): + def post(url, data, headers, timeout): + raise OSError("fail") + + verify_server( + "payload", + self.URL, + retries=2, + retry_delay=0.5, + retry_backoff="fixed", + http_post=post, + ) + self.assertEqual(self._sleep_calls, [0.5, 0.5]) + + def test_exponential_backoff(self): + def post(url, data, headers, timeout): + raise OSError("fail") + + verify_server( + "payload", + self.URL, + retries=2, + retry_delay=0.5, + retry_backoff="exponential", + http_post=post, + ) + self.assertEqual(self._sleep_calls, [0.5, 1.0]) + + if __name__ == "__main__": unittest.main()
