From 7b1e9f630943a789db239840f95773db1a69dfd9 Mon Sep 17 00:00:00 2001 From: Andrew Miller Date: Thu, 4 Jun 2026 13:53:16 -0400 Subject: [PATCH] Add proof-measure client + Gitea PyPI publish; bump to 0.4.0 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. --- .gitea/workflows/publish.yml | 37 +++++++ README.md | 39 +++++++ prime_sdk/__init__.py | 31 +++++- prime_sdk/models.py | 166 ++++++++++++++++++++++++++++ prime_sdk/proof_measure.py | 64 +++++++++++ prime_sdk/unauthenticated_client.py | 89 +++++++++++++++ pyproject.toml | 2 +- tests/test_client.py | 12 +- tests/test_proof_measure.py | 145 ++++++++++++++++++++++++ 9 files changed, 581 insertions(+), 4 deletions(-) create mode 100644 .gitea/workflows/publish.yml create mode 100644 prime_sdk/proof_measure.py create mode 100644 prime_sdk/unauthenticated_client.py create mode 100644 tests/test_proof_measure.py diff --git a/.gitea/workflows/publish.yml b/.gitea/workflows/publish.yml new file mode 100644 index 0000000..f64e03d --- /dev/null +++ b/.gitea/workflows/publish.yml @@ -0,0 +1,37 @@ +name: Publish to PyPI Registry + +on: + release: + types: [created, published] + workflow_dispatch: + +jobs: + publish: + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v3 + + - name: Setup Python + uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Install build tooling + run: python -m pip install --upgrade pip build twine + + - name: Run tests + run: | + python -m pip install -e '.[dev]' + python -m pytest -q + + - name: Build package + run: python -m build + + - name: Publish to Gitea PyPI registry + env: + TWINE_USERNAME: sa-dc-build + TWINE_PASSWORD: ${{ secrets.SA_DC_BUILD_API_TOKEN }} + TWINE_REPOSITORY_URL: https://git.dragonchain.com/api/packages/dragonchain/pypi + run: twine upload --non-interactive dist/* diff --git a/README.md b/README.md index fa9ecfd..2dbf9ee 100644 --- a/README.md +++ b/README.md @@ -146,6 +146,45 @@ client = DragonchainSDK( ### Block - `block.get(block_id)` — Get block by ID +### Proof Measure (public, unauthenticated) +- `proof_measure.get_security(network, since=None)` — A network's accumulated security since `since` (unix seconds), as a raw measure + USD valuation. `network` = `"BTC"` or `"ETH"`. +- `proof_measure.report(req)` — Per-transaction "securedBy" report over interchain anchors. +- `proof_measure.health()` — Service liveness. + +## Proof Measure + +`proof-measure` is a separate, **public, unauthenticated** Dragonchain service +(the measured-immutability / "securedBy" metric) at +`https://proof-measure.dragonchain.com`. It needs no credentials. Decimal fields +are returned as strings (full precision) and timestamps as int unix seconds. + +```python +import time +from prime_sdk import ProofMeasureClient, ReportRequest, ReportAnchorInput + +# Via the main SDK (targets the default public endpoint): +sec = client.proof_measure.get_security("BTC", int(time.time()) - 3600) +print(f"BTC secured by {sec.value_usd_formatted} ({sec.raw.value} {sec.raw.unit})") + +# Or standalone — no prime credentials needed: +pm = ProofMeasureClient() # or ProofMeasureClient("http://localhost:9481") + +report = pm.report( + ReportRequest( + transaction_id="tx-123", + prime_id="my-prime", + block_id="42", + anchors=[ + ReportAnchorInput(tx_hash="0x...", timestamp=anchor_unix, network="BTC"), + ReportAnchorInput(tx_hash="0x...", timestamp=anchor_unix, network="ETH"), + ], + ) +) +print(f"Secured by {report.total_value_usd_formatted} across {len(report.anchors)} anchors") + +pm.health() +``` + ## Authentication The SDK uses HMAC-SHA256 authentication. You need to provide: diff --git a/prime_sdk/__init__.py b/prime_sdk/__init__.py index 1eecae1..b5c23ef 100644 --- a/prime_sdk/__init__.py +++ b/prime_sdk/__init__.py @@ -71,11 +71,24 @@ from .models import ( TransactionTypeCreateResponse, VerificationBlock, ) +from .models import ( + AnchorSecurity, + HashPower, + HealthResponse, + RawMeasure, + ReportAnchorInput, + ReportRequest, + SecurityResult, + TransactionReport, +) +from .proof_measure import DEFAULT_BASE_URL as PROOF_MEASURE_DEFAULT_BASE_URL +from .proof_measure import ProofMeasureClient from .system import SystemClient from .transaction import TransactionClient from .transaction_type import TransactionTypeClient +from .unauthenticated_client import UnauthenticatedClient -__version__ = "0.2.0" +__version__ = "0.4.0" class DragonchainSDK: @@ -111,6 +124,10 @@ class DragonchainSDK: self.contract = ContractClient(self._client) self.block = BlockClient(self._client) self.system = SystemClient(self._client) + # proof-measure is a separate, unauthenticated public service; this + # handle targets its default public endpoint. For a custom endpoint + # construct a ``ProofMeasureClient`` directly. + self.proof_measure = ProofMeasureClient() def get_client(self) -> Client: """Return the underlying HTTP client (advanced use).""" @@ -128,6 +145,9 @@ __all__ = [ "SystemClient", "TransactionClient", "TransactionTypeClient", + "ProofMeasureClient", + "UnauthenticatedClient", + "PROOF_MEASURE_DEFAULT_BASE_URL", # models "Block", "BlockHeader", @@ -155,4 +175,13 @@ __all__ = [ "TransactionTypeCreateRequest", "TransactionTypeCreateResponse", "VerificationBlock", + # proof-measure models + "RawMeasure", + "SecurityResult", + "ReportAnchorInput", + "ReportRequest", + "AnchorSecurity", + "HashPower", + "TransactionReport", + "HealthResponse", ] diff --git a/prime_sdk/models.py b/prime_sdk/models.py index 90c85ef..39d29f4 100644 --- a/prime_sdk/models.py +++ b/prime_sdk/models.py @@ -492,3 +492,169 @@ class ListResponse: @classmethod def from_dict(cls, d: Dict[str, Any]) -> "ListResponse": return cls(items=d.get("items") or [], total_count=d.get("total_count", 0)) + + +# --------------------------------------------------------------------------- # +# Proof Measure (measured immutability / "securedBy") +# --------------------------------------------------------------------------- # +# +# Decimal-valued fields are kept as strings (full precision); timestamps are int +# unix seconds. JSON keys are camelCase, preserved verbatim from the service. + + +@dataclass +class RawMeasure: + """A network's native cumulative security measure: cumulative hashes for + PoW, stake-seconds for PoS.""" + + value: str = "" # human-scaled mantissa, e.g. "484.44" + unit: str = "" # e.g. "Zettahashes", "ETH·s" + base: str = "" # unscaled base amount (hashes or stake-seconds) + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "RawMeasure": + return cls( + value=d.get("value", ""), + unit=d.get("unit", ""), + base=d.get("base", ""), + ) + + +@dataclass +class SecurityResult: + """Security a single network accrued for a window/anchor, as the raw native + measure AND a USD valuation, plus a normalized 0..1 score.""" + + network: str = "" + consensus: str = "" # "pow" | "pos" + raw: RawMeasure = field(default_factory=RawMeasure) + value_usd: str = "" + value_usd_formatted: str = "" + label: str = "" + normalized_score: str = "" + since: int = 0 # window/anchor start (unix seconds) + as_of: int = 0 # latest sample time (unix seconds) + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "SecurityResult": + return cls( + network=d.get("network", ""), + consensus=d.get("consensus", ""), + raw=RawMeasure.from_dict(d.get("raw") or {}), + value_usd=d.get("valueUsd", ""), + value_usd_formatted=d.get("valueUsdFormatted", ""), + label=d.get("label", ""), + normalized_score=d.get("normalizedScore", ""), + since=d.get("since", 0), + as_of=d.get("asOf", 0), + ) + + +@dataclass +class ReportAnchorInput: + """One anchor supplied in a report request. Provide either ``network`` + ("BTC"/"ETH") or the public-chain numeric ``chain_id``; network wins.""" + + tx_hash: str + timestamp: int # anchor time, unix seconds + network: Optional[str] = None + chain_id: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + d: Dict[str, Any] = {"txHash": self.tx_hash, "timestamp": self.timestamp} + if self.network: + d["network"] = self.network + if self.chain_id: + d["chainId"] = self.chain_id + return d + + +@dataclass +class ReportRequest: + """Body of ``ProofMeasureClient.report``: a transaction's interchain anchors.""" + + transaction_id: str + prime_id: str + block_id: str + anchors: List[ReportAnchorInput] = field(default_factory=list) + + def to_dict(self) -> Dict[str, Any]: + return { + "transactionId": self.transaction_id, + "primeId": self.prime_id, + "blockId": self.block_id, + "anchors": [a.to_dict() for a in self.anchors], + } + + +@dataclass +class AnchorSecurity: + """One interchain anchor with the security its public network has accumulated + since the anchor was placed.""" + + network: str = "" + anchor_timestamp: int = 0 # unix seconds + anchor_tx_hash: str = "" + security: SecurityResult = field(default_factory=SecurityResult) + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "AnchorSecurity": + return cls( + network=d.get("network", ""), + anchor_timestamp=d.get("anchorTimestamp", 0), + anchor_tx_hash=d.get("anchorTxHash", ""), + security=SecurityResult.from_dict(d.get("security") or {}), + ) + + +@dataclass +class HashPower: + """Combined raw hash power across a report's PoW anchors.""" + + value: str = "" + units: str = "" + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "HashPower": + return cls(value=d.get("value", ""), units=d.get("units", "")) + + +@dataclass +class TransactionReport: + """Per-transaction "securedBy" report: every public-chain anchor covering the + transaction's block with both raw and USD security, plus combined totals. + ``hash_power`` is None when there are no PoW anchors.""" + + transaction_id: str = "" + prime_id: str = "" + block_id: str = "" + anchors: List[AnchorSecurity] = field(default_factory=list) + total_value_usd: str = "" + total_value_usd_formatted: str = "" + hash_power: Optional[HashPower] = None + total_normalized_score: str = "" + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "TransactionReport": + hp = d.get("hashPower") + return cls( + transaction_id=d.get("transactionId", ""), + prime_id=d.get("primeId", ""), + block_id=d.get("blockId", ""), + anchors=[AnchorSecurity.from_dict(a) for a in (d.get("anchors") or [])], + total_value_usd=d.get("totalValueUsd", ""), + total_value_usd_formatted=d.get("totalValueUsdFormatted", ""), + hash_power=HashPower.from_dict(hp) if hp else None, + total_normalized_score=d.get("totalNormalizedScore", ""), + ) + + +@dataclass +class HealthResponse: + """proof-measure liveness payload.""" + + status: str = "" + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "HealthResponse": + return cls(status=d.get("status", "")) diff --git a/prime_sdk/proof_measure.py b/prime_sdk/proof_measure.py new file mode 100644 index 0000000..17e4ee9 --- /dev/null +++ b/prime_sdk/proof_measure.py @@ -0,0 +1,64 @@ +"""Client for the Dragonchain proof-measure service. + +proof-measure is the measured-immutability / "securedBy" metric for L1–L5 +verification chains. It is a separate, public, UNauthenticated service (no API +keys), so this client needs only a base URL. It exposes a network's accumulated +security as both a raw measure (cumulative hashes / stake-seconds) and a USD +valuation, plus a per-transaction "securedBy" report over interchain anchors. +""" + +from typing import Optional + +import requests + +from .client import DEFAULT_TIMEOUT +from .models import ( + CONTENT_TYPE_JSON, + HealthResponse, + ReportRequest, + SecurityResult, + TransactionReport, +) +from .unauthenticated_client import UnauthenticatedClient + +#: Public production proof-measure endpoint. +DEFAULT_BASE_URL = "https://proof-measure.dragonchain.com" + + +class ProofMeasureClient: + """Calls the proof-measure HTTP API. + + With no arguments it targets the public production endpoint, so + ``ProofMeasureClient()`` works with zero configuration. + """ + + def __init__( + self, + base_url: str = DEFAULT_BASE_URL, + timeout: int = DEFAULT_TIMEOUT, + session: "Optional[requests.Session]" = None, + ): + self._client = UnauthenticatedClient( + base_url or DEFAULT_BASE_URL, timeout=timeout, session=session + ) + + def get_security(self, network: str, since: Optional[int] = None) -> SecurityResult: + """Return the security a public network (``network`` = "BTC" or "ETH") + has accumulated since ``since`` (unix seconds), as a raw measure + USD + valuation. Omit ``since`` to use the service's default window.""" + path = f"/api/v1/security/{network}" + if since and since > 0: + path += f"?since={since}" + return self._client.get(path, SecurityResult) + + def report(self, req: ReportRequest) -> TransactionReport: + """Compute the per-transaction "securedBy" report for the supplied + interchain anchors: each anchor's raw + USD security since it was placed, + plus combined totals.""" + return self._client.post( + "/api/v1/report", CONTENT_TYPE_JSON, req, TransactionReport + ) + + def health(self) -> HealthResponse: + """Report service liveness and DB reachability.""" + return self._client.get("/api/v1/health", HealthResponse) diff --git a/prime_sdk/unauthenticated_client.py b/prime_sdk/unauthenticated_client.py new file mode 100644 index 0000000..03185fe --- /dev/null +++ b/prime_sdk/unauthenticated_client.py @@ -0,0 +1,89 @@ +"""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) diff --git a/pyproject.toml b/pyproject.toml index fe172b4..54cc175 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "prime-sdk-python" -version = "0.2.0" +version = "0.4.0" description = "A self-contained Python SDK for interacting with Dragonchain nodes." readme = "README.md" requires-python = ">=3.9" diff --git a/tests/test_client.py b/tests/test_client.py index 4ff5a60..7df9346 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -93,8 +93,16 @@ class _FakeSession: self.response = response self.calls = [] - def request(self, method, url, data=None, headers=None, timeout=None): - self.calls.append({"method": method, "url": url, "data": data, "headers": headers}) + def request(self, method, url, data=None, headers=None, timeout=None, allow_redirects=None): + self.calls.append( + { + "method": method, + "url": url, + "data": data, + "headers": headers, + "allow_redirects": allow_redirects, + } + ) return self.response diff --git a/tests/test_proof_measure.py b/tests/test_proof_measure.py new file mode 100644 index 0000000..f0a145e --- /dev/null +++ b/tests/test_proof_measure.py @@ -0,0 +1,145 @@ +"""Offline tests for the proof-measure client. No network access required.""" + +import json + +import pytest + +from prime_sdk import ( + DragonchainAPIError, + ProofMeasureClient, + ReportAnchorInput, + ReportRequest, +) +from prime_sdk.proof_measure import DEFAULT_BASE_URL + + +class _FakeResponse: + def __init__(self, status_code, text): + self.status_code = status_code + self.text = text + + +class _FakeSession: + def __init__(self, response): + self.response = response + self.calls = [] + + def request(self, method, url, data=None, headers=None, timeout=None, allow_redirects=None): + self.calls.append( + { + "method": method, + "url": url, + "data": data, + "headers": headers, + "allow_redirects": allow_redirects, + } + ) + return self.response + + +def _client_with(session) -> ProofMeasureClient: + pm = ProofMeasureClient() + pm._client._session = session + return pm + + +def test_default_base_url(): + assert DEFAULT_BASE_URL == "https://proof-measure.dragonchain.com" + assert ProofMeasureClient()._client.endpoint() == DEFAULT_BASE_URL + + +def test_get_security_builds_path_and_parses(): + body = json.dumps( + { + "network": "BTC", + "consensus": "pow", + "raw": {"value": "71.63", "unit": "Yottahashes", "base": "7.16e25"}, + "valueUsd": "13928170.26", + "valueUsdFormatted": "$13,928,170.26", + "label": "$13,928,170.26 energy consumed", + "normalizedScore": "0.9234", + "since": 1780504894, + "asOf": 1780591253, + } + ) + fake = _FakeSession(_FakeResponse(200, body)) + res = _client_with(fake).get_security("BTC", 1780504894) + + call = fake.calls[0] + assert call["method"] == "GET" + assert call["url"] == DEFAULT_BASE_URL + "/api/v1/security/BTC?since=1780504894" + assert call["allow_redirects"] is False + # no auth headers on the unauthenticated client + assert "Authorization" not in call["headers"] + assert res.network == "BTC" + assert res.consensus == "pow" + assert res.raw.unit == "Yottahashes" + assert res.value_usd == "13928170.26" + assert res.value_usd_formatted == "$13,928,170.26" + assert res.since == 1780504894 and res.as_of == 1780591253 + + +def test_get_security_omits_since_when_not_positive(): + fake = _FakeSession(_FakeResponse(200, json.dumps({"network": "ETH", "consensus": "pos"}))) + _client_with(fake).get_security("ETH") + assert fake.calls[0]["url"] == DEFAULT_BASE_URL + "/api/v1/security/ETH" + + +def test_report_posts_request_and_parses(): + resp_body = json.dumps( + { + "transactionId": "tx-1", + "primeId": "p-1", + "blockId": "42", + "anchors": [ + { + "network": "BTC", + "anchorTimestamp": 1712345678, + "anchorTxHash": "0xabc", + "security": {"network": "BTC", "consensus": "pow", "valueUsd": "100.00"}, + } + ], + "totalValueUsd": "100.00", + "totalValueUsdFormatted": "$100.00", + "hashPower": {"value": "40.06", "units": "Yottahashes"}, + "totalNormalizedScore": "0.9", + } + ) + fake = _FakeSession(_FakeResponse(200, resp_body)) + req = ReportRequest( + transaction_id="tx-1", + prime_id="p-1", + block_id="42", + anchors=[ReportAnchorInput(tx_hash="0xabc", timestamp=1712345678, network="BTC")], + ) + rep = _client_with(fake).report(req) + + call = fake.calls[0] + assert call["method"] == "POST" + assert call["url"] == DEFAULT_BASE_URL + "/api/v1/report" + assert call["headers"]["Content-Type"] == "application/json" + sent = json.loads(call["data"]) + assert sent == { + "transactionId": "tx-1", + "primeId": "p-1", + "blockId": "42", + "anchors": [{"txHash": "0xabc", "timestamp": 1712345678, "network": "BTC"}], + } + assert rep.total_value_usd_formatted == "$100.00" + assert rep.hash_power is not None and rep.hash_power.units == "Yottahashes" + assert len(rep.anchors) == 1 and rep.anchors[0].security.network == "BTC" + + +def test_health_parses(): + fake = _FakeSession(_FakeResponse(200, json.dumps({"status": "ok"}))) + res = _client_with(fake).health() + assert fake.calls[0]["url"] == DEFAULT_BASE_URL + "/api/v1/health" + assert res.status == "ok" + + +def test_error_status_raises(): + fake = _FakeSession(_FakeResponse(400, " bad request ")) + with pytest.raises(DragonchainAPIError) as exc: + _client_with(fake).health() + assert exc.value.status_code == 400 + assert exc.value.body == "bad request"