Andrew Miller
7b1e9f6309
All checks were successful
Publish to PyPI Registry / publish (release) Successful in 1m14s
proof-measure is a separate, public, unauthenticated Dragonchain service. Adds: - UnauthenticatedClient: HMAC-free transport mirroring Client (session injection, allow_redirects=False, from_dict decoding). - ProofMeasureClient: get_security / report / health; default base URL https://proof-measure.dragonchain.com. Standalone (ProofMeasureClient()) and via DragonchainSDK.proof_measure. - Proof-measure dataclass models (decimals as strings, timestamps as int). - .gitea/workflows/publish.yml: build + twine upload to the Gitea PyPI registry on release (the SDK had no publish workflow before). Also fixes 3 pre-existing failing tests: _FakeSession.request didn't accept the allow_redirects kwarg the client now passes (added by the prior redirect change), so the suite was red at HEAD.
90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
"""HTTP client for public Dragonchain services that require no credentials.
|
|
|
|
Mirrors :class:`prime_sdk.client.Client`'s request plumbing — body marshaling,
|
|
``allow_redirects=False``, status handling, and ``from_dict`` decoding — but
|
|
sends no DC1-HMAC-SHA256 Authorization/Dragonchain/Timestamp headers. Used by the
|
|
proof-measure client, which talks to a separate, public, unauthenticated service.
|
|
"""
|
|
|
|
import json
|
|
from typing import Any, Optional, Type, TypeVar
|
|
|
|
import requests
|
|
|
|
from .client import CONTENT_TYPE_JSON, DEFAULT_TIMEOUT, _to_jsonable
|
|
from .errors import DragonchainAPIError
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class UnauthenticatedClient:
|
|
"""Unauthenticated HTTP client for a single public service base URL."""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
timeout: int = DEFAULT_TIMEOUT,
|
|
session: Optional[requests.Session] = None,
|
|
):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.timeout = timeout
|
|
# A caller may inject a pre-configured Session (e.g. one with an
|
|
# SSRF-guarded transport adapter) when base_url is attacker-influenced.
|
|
self._session = session if session is not None else requests.Session()
|
|
|
|
def endpoint(self) -> str:
|
|
return self.base_url
|
|
|
|
def _do_request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
content_type: str = "",
|
|
body: Any = None,
|
|
response_cls: Optional[Type[T]] = None,
|
|
) -> Optional[T]:
|
|
if isinstance(body, (bytes, bytearray)):
|
|
body_bytes = bytes(body)
|
|
elif body is not None:
|
|
body_bytes = json.dumps(_to_jsonable(body)).encode("utf-8")
|
|
content_type = CONTENT_TYPE_JSON
|
|
else:
|
|
body_bytes = b""
|
|
|
|
headers = {"Accept": CONTENT_TYPE_JSON}
|
|
if content_type:
|
|
headers["Content-Type"] = content_type
|
|
|
|
url = self.base_url + path
|
|
# allow_redirects=False for the same reason as the authenticated client:
|
|
# a JSON API never legitimately 3xx-redirects, and following one is an
|
|
# SSRF vector.
|
|
resp = self._session.request(
|
|
method,
|
|
url,
|
|
data=body_bytes,
|
|
headers=headers,
|
|
timeout=self.timeout,
|
|
allow_redirects=False,
|
|
)
|
|
|
|
resp_text = resp.text
|
|
if resp.status_code >= 400:
|
|
raise DragonchainAPIError(resp.status_code, resp_text.strip())
|
|
|
|
if response_cls is not None and resp_text:
|
|
return response_cls.from_dict(json.loads(resp_text)) # type: ignore[attr-defined]
|
|
return None
|
|
|
|
def get(self, path: str, response_cls: Optional[Type[T]] = None) -> Optional[T]:
|
|
return self._do_request("GET", path, response_cls=response_cls)
|
|
|
|
def post(
|
|
self,
|
|
path: str,
|
|
content_type: str,
|
|
body: Any,
|
|
response_cls: Optional[Type[T]] = None,
|
|
) -> Optional[T]:
|
|
return self._do_request("POST", path, content_type, body, response_cls)
|