diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..282fb0a --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,21 @@ +# Changelog + +## v0.2.0 — server sync + +- Refactored package into modular files: + - `panel_sdk/client.py` + - `panel_sdk/rater.py` + - `panel_sdk/_signing.py` + - `panel_sdk/_scrubber.py` + - `panel_sdk/_retry.py` + - `panel_sdk/errors.py` + - `panel_sdk/types.py` +- Added operator-side methods: `ingest_units`, `score_unit`, `skill_review`, `ingest_trace_and_wait` +- Added typed trace ingest results (`TraceResult`) for sync/pending responses +- Added dual-secret support with `site_secret_source="raw"` and `x-panel-ingest-secret` +- Added scrubber dispatch modes: `off`, `self-sign`, `proxy` +- Added typed errors: `PanelRateLimitError`, `PanelScrubberError` +- Added rater clients: `RaterClient`, `AsyncRaterClient` +- Added 429/5xx retry and backoff behavior +- Added unit tests covering HMAC signing, scrubber JWT, async trace polling, and 429 retry +- Kept `verify_token(token)` public signature unchanged diff --git a/README.md b/README.md index 64518f8..8dbbd01 100644 --- a/README.md +++ b/README.md @@ -1,28 +1,52 @@ # panel-sdk (python) -thin client for [panel](https://github.com/UltraInstinct0x/panel). python 3.10+. +Thin client for [panel](https://github.com/UltraInstinct0x/panel). Python 3.10+. -``` +## Install + +```bash pip install panel-sdk ``` +## v0.2.0 options + ```python from panel_sdk import PanelClient panel = PanelClient( base_url="https://panel.example.com", - site_key=os.environ["PANEL_SITE_KEY"], - site_secret=os.environ["PANEL_SITE_SECRET"], - scrubber_secret=os.environ.get("SCRUBBER_JWT_SECRET"), # omit for first-party keys + site_key="pk_live_xxx", + site_secret="secret", + site_secret_source="env", # or "raw" for dual-secret mode + scrubber_mode="off", # off | self-sign | proxy + scrubber_secret=None, # required when scrubber_mode="self-sign" + scrubber_url=None, # required when scrubber_mode="proxy" + engine_version="0.2.0", + timeout_seconds=10.0, + max_retries=3, ) +``` -v = panel.verify_token(request.json["panel_token"]) -if not v.ok or (v.trust or 0) < 0.5: - abort(403) +## PanelClient methods -panel.ingest_trace(trace_id=f"tr_{uuid4()}", source_agent="myapp", - blob={"messages": [...]}) -``` +- `ingest_trace(source_agent, blob, trace_id=None) -> TraceResult` +- `ingest_trace_and_wait(source_agent, blob, trace_id=None, max_wait_seconds=60, poll_interval_seconds=1.5) -> dict` +- `fetch_trace(trace_id) -> dict` +- `ingest_units(units) -> dict` +- `score_unit(ref=None, unit_id=None) -> dict` +- `skill_review(skill_name, diff, ...) -> dict` +- `verify_token(token) -> VerifyResult` + +`verify_token(token)` signature remains unchanged. + +## Rater clients + +- `RaterClient.next_unit(pool, rater_id)` +- `RaterClient.submit_judgment(unit_id, choice)` +- Async parity via `AsyncRaterClient` + +## Errors -methods (sync + async via `AsyncPanelClient`): `ingest_unit`, `ingest_trace`, `verify_token`, `fetch_unit`, `fetch_trace`. -auth: HMAC-SHA256 (`x-panel-ingest-sig`) + optional scrubber JWT. +- `PanelError` +- `PanelRateLimitError` (includes `scope`, `retry_after_s`) +- `PanelScrubberError` diff --git a/panel_sdk/__init__.py b/panel_sdk/__init__.py index e3e8060..c40137d 100644 --- a/panel_sdk/__init__.py +++ b/panel_sdk/__init__.py @@ -1,192 +1,20 @@ -"""panel_sdk — thin client for the panel HTTP api. - -usage: - from panel_sdk import PanelClient - c = PanelClient(base_url="https://panel.example.com", - site_key="pk_live_xxx", site_secret="...", - scrubber_secret="...") # omit for first-party keys - c.verify_token(token) -""" -from __future__ import annotations - -import base64 -import hashlib -import hmac -import json -import secrets -import time -from dataclasses import dataclass -from typing import Any, Mapping, MutableMapping - -import httpx - -__version__ = "0.1.0" -__all__ = ["PanelClient", "AsyncPanelClient", "PanelError", "VerifyResult"] - - -class PanelError(Exception): - def __init__(self, status: int, body: Any, message: str) -> None: - super().__init__(message) - self.status = status - self.body = body - - -@dataclass -class VerifyResult: - ok: bool - trust: float | None = None - tier_used: str | None = None - unit_ids: list[str] | None = None - reason: str | None = None - - @classmethod - def from_json(cls, j: Mapping[str, Any]) -> "VerifyResult": - return cls( - ok=bool(j.get("ok")), - trust=j.get("trust"), - tier_used=j.get("tier_used"), - unit_ids=j.get("unit_ids"), - reason=j.get("reason"), - ) - - -def _b64u(b: bytes) -> str: - return base64.urlsafe_b64encode(b).rstrip(b"=").decode("ascii") - - -def _hmac_hex(secret: str, body: str) -> str: - return hmac.new(secret.encode(), body.encode(), hashlib.sha256).hexdigest() - - -def _sha256_hex(s: str) -> str: - return hashlib.sha256(s.encode()).hexdigest() - - -def _jwt_hs256(secret: str, payload: Mapping[str, Any]) -> str: - header = {"alg": "HS256", "typ": "JWT"} - si = f"{_b64u(json.dumps(header, separators=(',', ':')).encode())}.{_b64u(json.dumps(payload, separators=(',', ':')).encode())}" - sig = _b64u(hmac.new(secret.encode(), si.encode(), hashlib.sha256).digest()) - return f"{si}.{sig}" - - -def _attest(scrubber_secret: str, body: str, engine_version: str) -> str: - now = int(time.time()) - return _jwt_hs256(scrubber_secret, { - "jti": secrets.token_hex(16), - "iat": now, - "exp": now + 300, - "input_hash": "x", - "output_hash": _sha256_hex(body), - "mode": "text", - "engine_version": engine_version, - }) - - -def _check(resp: httpx.Response) -> Any: - try: - j = resp.json() - except Exception: - j = {"raw": resp.text} - if not (200 <= resp.status_code < 300): - raise PanelError(resp.status_code, j, f"panel {resp.status_code}: {resp.text[:300]}") - return j - - -def _ingest_headers(site_key: str, site_secret: str, body: str, - scrubber_secret: str | None, engine_version: str) -> MutableMapping[str, str]: - h: MutableMapping[str, str] = { - "content-type": "application/json", - "x-panel-site-key": site_key, - "x-panel-ingest-sig": _hmac_hex(site_secret, body), - } - if scrubber_secret: - h["x-scrubber-attestation"] = _attest(scrubber_secret, body, engine_version) - return h - - -class PanelClient: - def __init__(self, base_url: str, site_key: str, site_secret: str, - scrubber_secret: str | None = None, engine_version: str = "0.2.0", - client: httpx.Client | None = None, timeout: float = 10.0) -> None: - self.base = base_url.rstrip("/") - self.site_key = site_key - self.site_secret = site_secret - self.scrubber_secret = scrubber_secret - self.engine_version = engine_version - self._client = client or httpx.Client(timeout=timeout) - - def ingest_unit(self, *, type: str, payload: dict, pool: str | None = None) -> dict: - d: dict[str, Any] = {"type": type, "payload": payload} - if pool is not None: d["pool"] = pool - body = json.dumps(d, separators=(",", ":")) - r = self._client.post(self.base + "/api/units/ingest", - headers=_ingest_headers(self.site_key, self.site_secret, body, - self.scrubber_secret, self.engine_version), - content=body) - return _check(r) - - def ingest_trace(self, *, trace_id: str, source_agent: str, blob: dict) -> dict: - body = json.dumps({"trace_id": trace_id, "source_agent": source_agent, "blob": blob}, separators=(",", ":")) - r = self._client.post(self.base + "/api/v1/traces", - headers=_ingest_headers(self.site_key, self.site_secret, body, - self.scrubber_secret, self.engine_version), - content=body) - return _check(r) - - def verify_token(self, token: str) -> VerifyResult: - r = self._client.post(self.base + "/v1/verify", - headers={"content-type": "application/json"}, - json={"token": token, "site_key": self.site_key}) - return VerifyResult.from_json(_check(r)) - - def fetch_unit(self, unit_id: str) -> dict: - r = self._client.get(f"{self.base}/api/units/{unit_id}") - return _check(r) - - def fetch_trace(self, trace_id: str) -> dict: - r = self._client.get(f"{self.base}/api/v1/traces/{trace_id}") - return _check(r) - - -class AsyncPanelClient: - def __init__(self, base_url: str, site_key: str, site_secret: str, - scrubber_secret: str | None = None, engine_version: str = "0.2.0", - client: httpx.AsyncClient | None = None, timeout: float = 10.0) -> None: - self.base = base_url.rstrip("/") - self.site_key = site_key - self.site_secret = site_secret - self.scrubber_secret = scrubber_secret - self.engine_version = engine_version - self._client = client or httpx.AsyncClient(timeout=timeout) - - async def ingest_unit(self, *, type: str, payload: dict, pool: str | None = None) -> dict: - d: dict[str, Any] = {"type": type, "payload": payload} - if pool is not None: d["pool"] = pool - body = json.dumps(d, separators=(",", ":")) - r = await self._client.post(self.base + "/api/units/ingest", - headers=_ingest_headers(self.site_key, self.site_secret, body, - self.scrubber_secret, self.engine_version), - content=body) - return _check(r) - - async def ingest_trace(self, *, trace_id: str, source_agent: str, blob: dict) -> dict: - body = json.dumps({"trace_id": trace_id, "source_agent": source_agent, "blob": blob}, separators=(",", ":")) - r = await self._client.post(self.base + "/api/v1/traces", - headers=_ingest_headers(self.site_key, self.site_secret, body, - self.scrubber_secret, self.engine_version), - content=body) - return _check(r) - - async def verify_token(self, token: str) -> VerifyResult: - r = await self._client.post(self.base + "/v1/verify", - headers={"content-type": "application/json"}, - json={"token": token, "site_key": self.site_key}) - return VerifyResult.from_json(_check(r)) - - async def fetch_unit(self, unit_id: str) -> dict: - r = await self._client.get(f"{self.base}/api/units/{unit_id}") - return _check(r) - - async def fetch_trace(self, trace_id: str) -> dict: - r = await self._client.get(f"{self.base}/api/v1/traces/{trace_id}") - return _check(r) +"""Public package exports for panel-sdk.""" + +from panel_sdk.client import AsyncPanelClient, PanelClient +from panel_sdk.errors import PanelError, PanelRateLimitError, PanelScrubberError +from panel_sdk.rater import AsyncRaterClient, RaterClient +from panel_sdk.types import TraceResult, VerifyResult + +__version__ = "0.2.0" + +__all__ = [ + "AsyncPanelClient", + "AsyncRaterClient", + "PanelClient", + "PanelError", + "PanelRateLimitError", + "PanelScrubberError", + "RaterClient", + "TraceResult", + "VerifyResult", +] diff --git a/panel_sdk/_retry.py b/panel_sdk/_retry.py new file mode 100644 index 0000000..89c736b --- /dev/null +++ b/panel_sdk/_retry.py @@ -0,0 +1,43 @@ +"""HTTP retry helpers for rate-limit and server errors.""" + +from __future__ import annotations + +import time +from typing import Any + +from panel_sdk.errors import PanelError, PanelRateLimitError + +SERVER_BACKOFF_S = (0.5, 1.5, 3.5) + + +def parse_error_response(status_code: int, data: Any, text: str) -> Any: + """Raise typed errors for non-success responses.""" + if status_code == 429: + scope = data.get("scope") if isinstance(data, dict) else None + retry_after = data.get("retry_after_s") if isinstance(data, dict) else None + if isinstance(retry_after, int): + retry_after = float(retry_after) + if retry_after is not None and not isinstance(retry_after, float): + retry_after = None + raise PanelRateLimitError(status=429, body=data, scope=scope, retry_after_s=retry_after) + raise PanelError(status=status_code, body=data, message=f"panel {status_code}: {text[:300]}") + + +def should_retry(status_code: int, attempt: int, max_retries: int) -> bool: + """Return whether a request should be retried for this status.""" + if attempt >= max_retries: + return False + return status_code == 429 or status_code >= 500 + + +def retry_delay_seconds(status_code: int, attempt: int, retry_after_s: float | None) -> float: + """Return capped sleep duration before next retry.""" + if status_code == 429: + return min(30.0, retry_after_s or 1.0) + idx = min(attempt, len(SERVER_BACKOFF_S) - 1) + return SERVER_BACKOFF_S[idx] + + +def blocking_sleep(seconds: float) -> None: + """Sleep helper for sync retries.""" + time.sleep(seconds) diff --git a/panel_sdk/_scrubber.py b/panel_sdk/_scrubber.py new file mode 100644 index 0000000..05d9865 --- /dev/null +++ b/panel_sdk/_scrubber.py @@ -0,0 +1,108 @@ +"""Scrubber attestation helpers for off/self-sign/proxy modes.""" + +from __future__ import annotations + +import secrets +import time +from typing import Any + +import httpx + +from panel_sdk._signing import jwt_hs256, sha256_hex +from panel_sdk.errors import PanelScrubberError + + +def build_scrubber_headers( + *, + mode: str, + body: str, + engine_version: str, + scrubber_secret: str | None, + scrubber_url: str | None, + client: httpx.Client, + timeout_seconds: float, +) -> tuple[str, dict[str, str]]: + """Build scrubber attestation header and potentially scrubbed body.""" + if mode == "off": + return body, {} + if mode == "self-sign": + if not scrubber_secret: + raise PanelScrubberError(400, {"error": "missing_scrubber_secret"}, "scrubber_secret required") + now = int(time.time()) + token = jwt_hs256( + scrubber_secret, + { + "jti": secrets.token_hex(16), + "iat": now, + "exp": now + 300, + "input_hash": "x", + "output_hash": sha256_hex(body), + "mode": "text", + "engine_version": engine_version, + }, + ) + return body, {"x-scrubber-attestation": token} + if mode == "proxy": + if not scrubber_url: + raise PanelScrubberError(400, {"error": "missing_scrubber_url"}, "scrubber_url required") + resp = client.post( + f"{scrubber_url.rstrip('/')}/scrub", + headers={"content-type": "application/json"}, + content=body, + timeout=timeout_seconds, + ) + if resp.status_code >= 400: + raise PanelScrubberError(resp.status_code, {"raw": resp.text}, "scrubber proxy failed") + token = resp.headers.get("x-scrubber-attestation") + if not token: + raise PanelScrubberError(502, {"error": "missing_attestation"}, "scrubber proxy did not return attestation") + return resp.text, {"x-scrubber-attestation": token} + raise PanelScrubberError(400, {"error": "invalid_scrubber_mode", "mode": mode}, "invalid scrubber_mode") + + +async def build_scrubber_headers_async( + *, + mode: str, + body: str, + engine_version: str, + scrubber_secret: str | None, + scrubber_url: str | None, + client: httpx.AsyncClient, + timeout_seconds: float, +) -> tuple[str, dict[str, str]]: + """Async version of scrubber dispatch.""" + if mode == "off": + return body, {} + if mode == "self-sign": + if not scrubber_secret: + raise PanelScrubberError(400, {"error": "missing_scrubber_secret"}, "scrubber_secret required") + now = int(time.time()) + token = jwt_hs256( + scrubber_secret, + { + "jti": secrets.token_hex(16), + "iat": now, + "exp": now + 300, + "input_hash": "x", + "output_hash": sha256_hex(body), + "mode": "text", + "engine_version": engine_version, + }, + ) + return body, {"x-scrubber-attestation": token} + if mode == "proxy": + if not scrubber_url: + raise PanelScrubberError(400, {"error": "missing_scrubber_url"}, "scrubber_url required") + resp = await client.post( + f"{scrubber_url.rstrip('/')}/scrub", + headers={"content-type": "application/json"}, + content=body, + timeout=timeout_seconds, + ) + if resp.status_code >= 400: + raise PanelScrubberError(resp.status_code, {"raw": resp.text}, "scrubber proxy failed") + token = resp.headers.get("x-scrubber-attestation") + if not token: + raise PanelScrubberError(502, {"error": "missing_attestation"}, "scrubber proxy did not return attestation") + return resp.text, {"x-scrubber-attestation": token} + raise PanelScrubberError(400, {"error": "invalid_scrubber_mode", "mode": mode}, "invalid scrubber_mode") diff --git a/panel_sdk/_signing.py b/panel_sdk/_signing.py new file mode 100644 index 0000000..2c628a6 --- /dev/null +++ b/panel_sdk/_signing.py @@ -0,0 +1,44 @@ +"""Signing and canonicalization helpers for panel auth.""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +from typing import Any, Mapping + + +def canonical_json(value: Any) -> str: + """Return compact canonical JSON string.""" + return json.dumps(value, separators=(",", ":"), sort_keys=False) + + +def hmac_sha256_hex(secret: str, text: str) -> str: + """Return lowercase hex HMAC SHA-256 signature.""" + return hmac.new(secret.encode("utf-8"), text.encode("utf-8"), hashlib.sha256).hexdigest() + + +def sha256_hex(text: str) -> str: + """Return lowercase SHA-256 hex digest.""" + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _b64u(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def jwt_hs256(secret: str, payload: Mapping[str, Any]) -> str: + """Create a JWT with HS256 signature.""" + header = {"alg": "HS256", "typ": "JWT"} + signing_input = ( + f"{_b64u(canonical_json(header).encode('utf-8'))}." + f"{_b64u(canonical_json(payload).encode('utf-8'))}" + ) + signature = _b64u(hmac.new(secret.encode("utf-8"), signing_input.encode("utf-8"), hashlib.sha256).digest()) + return f"{signing_input}.{signature}" + + +def canonical_score_string(site_key: str, ref: str | None = None, unit_id: str | None = None) -> str: + """Build canonical query signature string for score endpoint.""" + return f"GET\n/api/units/score\nref={ref or ''}\nid={unit_id or ''}\nsite={site_key}" diff --git a/panel_sdk/client.py b/panel_sdk/client.py new file mode 100644 index 0000000..b0042e3 --- /dev/null +++ b/panel_sdk/client.py @@ -0,0 +1,363 @@ +"""Panel operator clients (sync + async).""" + +from __future__ import annotations + +import asyncio +import time +from typing import Any, Mapping + +import httpx + +from panel_sdk._retry import blocking_sleep, parse_error_response, retry_delay_seconds, should_retry +from panel_sdk._scrubber import build_scrubber_headers, build_scrubber_headers_async +from panel_sdk._signing import canonical_json, canonical_score_string, hmac_sha256_hex +from panel_sdk.errors import PanelRateLimitError +from panel_sdk.types import TraceResult, VerifyResult + + +def _json_or_raw(response: httpx.Response) -> Any: + try: + return response.json() + except Exception: + return {"raw": response.text} + + +class PanelClient: + """Synchronous operator client for panel API.""" + + def __init__( + self, + *, + base_url: str, + site_key: str, + site_secret: str, + site_secret_source: str = "env", + scrubber_mode: str = "off", + scrubber_secret: str | None = None, + scrubber_url: str | None = None, + engine_version: str = "0.2.0", + timeout_seconds: float = 10.0, + max_retries: int = 3, + client: httpx.Client | None = None, + ) -> None: + """Initialize PanelClient with server-sync options.""" + self.base_url = base_url.rstrip("/") + self.site_key = site_key + self.site_secret = site_secret + self.site_secret_source = site_secret_source + self.scrubber_mode = scrubber_mode + self.scrubber_secret = scrubber_secret + self.scrubber_url = scrubber_url + self.engine_version = engine_version + self.timeout_seconds = timeout_seconds + self.max_retries = max_retries + self._client = client or httpx.Client(timeout=timeout_seconds) + + def _signed_request(self, method: str, path: str, *, data: Any | None = None, params: dict[str, str] | None = None) -> Any: + body_obj = data if data is not None else {} + body = canonical_json(body_obj) + body, scrubber_headers = build_scrubber_headers( + mode=self.scrubber_mode, + body=body, + engine_version=self.engine_version, + scrubber_secret=self.scrubber_secret, + scrubber_url=self.scrubber_url, + client=self._client, + timeout_seconds=self.timeout_seconds, + ) + + signature = hmac_sha256_hex(self.site_secret, body) + if method.upper() == "GET" and path == "/api/units/score": + signature = hmac_sha256_hex( + self.site_secret, + canonical_score_string(self.site_key, ref=params.get("ref") if params else None, unit_id=params.get("id") if params else None), + ) + + headers: dict[str, str] = { + "content-type": "application/json", + "x-panel-site-key": self.site_key, + "x-panel-ingest-sig": signature, + **scrubber_headers, + } + if self.site_secret_source == "raw": + headers["x-panel-ingest-secret"] = self.site_secret + + attempt = 0 + while True: + response = self._client.request( + method, + f"{self.base_url}{path}", + headers=headers, + content=body if method.upper() != "GET" else None, + params=params, + ) + payload = _json_or_raw(response) + if 200 <= response.status_code < 300: + return payload + retry_after_s: float | None = None + if response.status_code == 429: + ra = response.headers.get("Retry-After") + if ra: + try: + retry_after_s = float(ra) + except ValueError: + retry_after_s = None + if isinstance(payload, dict) and payload.get("retry_after_s") is not None: + retry_after_s = float(payload["retry_after_s"]) + if should_retry(response.status_code, attempt, self.max_retries): + delay = retry_delay_seconds(response.status_code, attempt, retry_after_s) + blocking_sleep(delay) + attempt += 1 + continue + parse_error_response(response.status_code, payload, response.text) + + def ingest_trace(self, *, source_agent: str, blob: Mapping[str, Any], trace_id: str | None = None) -> TraceResult: + """Ingest a trace payload and return done/pending typed result.""" + body: dict[str, Any] = {"source_agent": source_agent, "blob": blob} + if trace_id: + body["trace_id"] = trace_id + data = self._signed_request("POST", "/api/v1/traces", data=body) + return TraceResult.from_json(data, self.base_url) + + def ingest_trace_and_wait(self, *, source_agent: str, blob: Mapping[str, Any], trace_id: str | None = None, max_wait_seconds: float = 60.0, poll_interval_seconds: float = 1.5) -> dict[str, Any]: + """Ingest a trace and poll until completed or timeout.""" + result = self.ingest_trace(source_agent=source_agent, blob=blob, trace_id=trace_id) + if result.status != "pending": + return {"status": result.status, "trace_id": result.trace_id, "unit_ids": result.unit_ids} + deadline = time.time() + max_wait_seconds + while time.time() < deadline: + polled = self.fetch_trace(result.trace_id) + status = str(polled.get("status", "")) + if status and status != "pending": + return polled + blocking_sleep(poll_interval_seconds) + raise TimeoutError("trace polling timed out") + + def fetch_trace(self, trace_id: str) -> dict[str, Any]: + """Fetch trace status by trace ID.""" + response = self._client.get(f"{self.base_url}/api/v1/traces/{trace_id}") + payload = _json_or_raw(response) + if not (200 <= response.status_code < 300): + parse_error_response(response.status_code, payload, response.text) + return payload + + def ingest_units(self, units: list[dict[str, Any]] | dict[str, Any]) -> dict[str, Any]: + """Ingest one or many units using /api/units/ingest.""" + return self._signed_request("POST", "/api/units/ingest", data={"units": units} if isinstance(units, list) else units) + + def score_unit(self, *, ref: str | None = None, unit_id: str | None = None) -> dict[str, Any]: + """Lookup aggregate score by external ref or unit ID.""" + params: dict[str, str] = {} + if ref is not None: + params["ref"] = ref + if unit_id is not None: + params["id"] = unit_id + return self._signed_request("GET", "/api/units/score", params=params) + + def skill_review( + self, + *, + skill_name: str, + diff: str, + external_ref: str | None = None, + context: str | None = None, + source_agent: str | None = None, + yes_label: str | None = None, + no_label: str | None = None, + trusted_pool_only: bool | None = None, + ) -> dict[str, Any]: + """Call skill review convenience endpoint.""" + payload: dict[str, Any] = {"skill_name": skill_name, "diff": diff} + for key, value in { + "external_ref": external_ref, + "context": context, + "source_agent": source_agent, + "yes_label": yes_label, + "no_label": no_label, + "trusted_pool_only": trusted_pool_only, + }.items(): + if value is not None: + payload[key] = value + return self._signed_request("POST", "/api/v1/skill-review", data=payload) + + def verify_token(self, token: str) -> VerifyResult: + """Verify a widget token. Public API signature intentionally unchanged.""" + response = self._client.post( + f"{self.base_url}/v1/verify", + headers={"content-type": "application/json"}, + json={"token": token, "site_key": self.site_key}, + ) + payload = _json_or_raw(response) + if not (200 <= response.status_code < 300): + parse_error_response(response.status_code, payload, response.text) + return VerifyResult.from_json(payload) + + +class AsyncPanelClient: + """Asynchronous operator client for panel API.""" + + def __init__( + self, + *, + base_url: str, + site_key: str, + site_secret: str, + site_secret_source: str = "env", + scrubber_mode: str = "off", + scrubber_secret: str | None = None, + scrubber_url: str | None = None, + engine_version: str = "0.2.0", + timeout_seconds: float = 10.0, + max_retries: int = 3, + client: httpx.AsyncClient | None = None, + ) -> None: + """Initialize AsyncPanelClient with server-sync options.""" + self.base_url = base_url.rstrip("/") + self.site_key = site_key + self.site_secret = site_secret + self.site_secret_source = site_secret_source + self.scrubber_mode = scrubber_mode + self.scrubber_secret = scrubber_secret + self.scrubber_url = scrubber_url + self.engine_version = engine_version + self.timeout_seconds = timeout_seconds + self.max_retries = max_retries + self._client = client or httpx.AsyncClient(timeout=timeout_seconds) + + async def _signed_request(self, method: str, path: str, *, data: Any | None = None, params: dict[str, str] | None = None) -> Any: + body_obj = data if data is not None else {} + body = canonical_json(body_obj) + body, scrubber_headers = await build_scrubber_headers_async( + mode=self.scrubber_mode, + body=body, + engine_version=self.engine_version, + scrubber_secret=self.scrubber_secret, + scrubber_url=self.scrubber_url, + client=self._client, + timeout_seconds=self.timeout_seconds, + ) + + signature = hmac_sha256_hex(self.site_secret, body) + if method.upper() == "GET" and path == "/api/units/score": + signature = hmac_sha256_hex( + self.site_secret, + canonical_score_string(self.site_key, ref=params.get("ref") if params else None, unit_id=params.get("id") if params else None), + ) + + headers: dict[str, str] = { + "content-type": "application/json", + "x-panel-site-key": self.site_key, + "x-panel-ingest-sig": signature, + **scrubber_headers, + } + if self.site_secret_source == "raw": + headers["x-panel-ingest-secret"] = self.site_secret + + attempt = 0 + while True: + response = await self._client.request( + method, + f"{self.base_url}{path}", + headers=headers, + content=body if method.upper() != "GET" else None, + params=params, + ) + payload = _json_or_raw(response) + if 200 <= response.status_code < 300: + return payload + retry_after_s: float | None = None + if response.status_code == 429: + ra = response.headers.get("Retry-After") + if ra: + try: + retry_after_s = float(ra) + except ValueError: + retry_after_s = None + if isinstance(payload, dict) and payload.get("retry_after_s") is not None: + retry_after_s = float(payload["retry_after_s"]) + if should_retry(response.status_code, attempt, self.max_retries): + await asyncio.sleep(retry_delay_seconds(response.status_code, attempt, retry_after_s)) + attempt += 1 + continue + parse_error_response(response.status_code, payload, response.text) + + async def ingest_trace(self, *, source_agent: str, blob: Mapping[str, Any], trace_id: str | None = None) -> TraceResult: + """Ingest a trace payload and return done/pending typed result.""" + body: dict[str, Any] = {"source_agent": source_agent, "blob": blob} + if trace_id: + body["trace_id"] = trace_id + data = await self._signed_request("POST", "/api/v1/traces", data=body) + return TraceResult.from_json(data, self.base_url) + + async def ingest_trace_and_wait(self, *, source_agent: str, blob: Mapping[str, Any], trace_id: str | None = None, max_wait_seconds: float = 60.0, poll_interval_seconds: float = 1.5) -> dict[str, Any]: + """Ingest a trace and poll until completed or timeout.""" + result = await self.ingest_trace(source_agent=source_agent, blob=blob, trace_id=trace_id) + if result.status != "pending": + return {"status": result.status, "trace_id": result.trace_id, "unit_ids": result.unit_ids} + deadline = time.time() + max_wait_seconds + while time.time() < deadline: + polled = await self.fetch_trace(result.trace_id) + status = str(polled.get("status", "")) + if status and status != "pending": + return polled + await asyncio.sleep(poll_interval_seconds) + raise TimeoutError("trace polling timed out") + + async def fetch_trace(self, trace_id: str) -> dict[str, Any]: + """Fetch trace status by trace ID.""" + response = await self._client.get(f"{self.base_url}/api/v1/traces/{trace_id}") + payload = _json_or_raw(response) + if not (200 <= response.status_code < 300): + parse_error_response(response.status_code, payload, response.text) + return payload + + async def ingest_units(self, units: list[dict[str, Any]] | dict[str, Any]) -> dict[str, Any]: + """Ingest one or many units using /api/units/ingest.""" + return await self._signed_request("POST", "/api/units/ingest", data={"units": units} if isinstance(units, list) else units) + + async def score_unit(self, *, ref: str | None = None, unit_id: str | None = None) -> dict[str, Any]: + """Lookup aggregate score by external ref or unit ID.""" + params: dict[str, str] = {} + if ref is not None: + params["ref"] = ref + if unit_id is not None: + params["id"] = unit_id + return await self._signed_request("GET", "/api/units/score", params=params) + + async def skill_review( + self, + *, + skill_name: str, + diff: str, + external_ref: str | None = None, + context: str | None = None, + source_agent: str | None = None, + yes_label: str | None = None, + no_label: str | None = None, + trusted_pool_only: bool | None = None, + ) -> dict[str, Any]: + """Call skill review convenience endpoint.""" + payload: dict[str, Any] = {"skill_name": skill_name, "diff": diff} + for key, value in { + "external_ref": external_ref, + "context": context, + "source_agent": source_agent, + "yes_label": yes_label, + "no_label": no_label, + "trusted_pool_only": trusted_pool_only, + }.items(): + if value is not None: + payload[key] = value + return await self._signed_request("POST", "/api/v1/skill-review", data=payload) + + async def verify_token(self, token: str) -> VerifyResult: + """Verify a widget token. Public API signature intentionally unchanged.""" + response = await self._client.post( + f"{self.base_url}/v1/verify", + headers={"content-type": "application/json"}, + json={"token": token, "site_key": self.site_key}, + ) + payload = _json_or_raw(response) + if not (200 <= response.status_code < 300): + parse_error_response(response.status_code, payload, response.text) + return VerifyResult.from_json(payload) diff --git a/panel_sdk/errors.py b/panel_sdk/errors.py new file mode 100644 index 0000000..9079d5b --- /dev/null +++ b/panel_sdk/errors.py @@ -0,0 +1,27 @@ +"""Typed exceptions for panel-sdk clients.""" + +from __future__ import annotations + +from typing import Any + + +class PanelError(Exception): + """Base exception for panel server errors.""" + + def __init__(self, status: int, body: Any, message: str) -> None: + super().__init__(message) + self.status = status + self.body = body + + +class PanelRateLimitError(PanelError): + """Exception raised when panel returns HTTP 429 rate limit responses.""" + + def __init__(self, status: int, body: Any, scope: str | None, retry_after_s: float | None) -> None: + super().__init__(status=status, body=body, message="panel rate limited") + self.scope = scope + self.retry_after_s = retry_after_s + + +class PanelScrubberError(PanelError): + """Exception raised when scrubber proxying or attestation fails.""" diff --git a/panel_sdk/rater.py b/panel_sdk/rater.py new file mode 100644 index 0000000..83308d4 --- /dev/null +++ b/panel_sdk/rater.py @@ -0,0 +1,84 @@ +"""Rater clients for pool fetch and judgment submission.""" + +from __future__ import annotations + +from typing import Any + +import httpx + +from panel_sdk._retry import parse_error_response + + +def _json_or_raw(response: httpx.Response) -> Any: + try: + return response.json() + except Exception: + return {"raw": response.text} + + +class RaterClient: + """Synchronous rater client using site-key auth only.""" + + def __init__(self, *, base_url: str, site_key: str, timeout_seconds: float = 10.0, client: httpx.Client | None = None) -> None: + """Initialize RaterClient.""" + self.base_url = base_url.rstrip("/") + self.site_key = site_key + self._client = client or httpx.Client(timeout=timeout_seconds) + + def next_unit(self, pool: str, rater_id: str) -> dict[str, Any]: + """Fetch next unit from rater pool.""" + response = self._client.get( + f"{self.base_url}/api/rater/next", + headers={"x-panel-site-key": self.site_key}, + params={"pool": pool, "rater_id": rater_id}, + ) + payload = _json_or_raw(response) + if not (200 <= response.status_code < 300): + parse_error_response(response.status_code, payload, response.text) + return payload + + def submit_judgment(self, unit_id: str, choice: str) -> dict[str, Any]: + """Submit a rater judgment choice.""" + response = self._client.post( + f"{self.base_url}/api/rater/judgment", + headers={"x-panel-site-key": self.site_key, "content-type": "application/json"}, + json={"unit_id": unit_id, "choice": choice}, + ) + payload = _json_or_raw(response) + if not (200 <= response.status_code < 300): + parse_error_response(response.status_code, payload, response.text) + return payload + + +class AsyncRaterClient: + """Asynchronous rater client using site-key auth only.""" + + def __init__(self, *, base_url: str, site_key: str, timeout_seconds: float = 10.0, client: httpx.AsyncClient | None = None) -> None: + """Initialize AsyncRaterClient.""" + self.base_url = base_url.rstrip("/") + self.site_key = site_key + self._client = client or httpx.AsyncClient(timeout=timeout_seconds) + + async def next_unit(self, pool: str, rater_id: str) -> dict[str, Any]: + """Fetch next unit from rater pool.""" + response = await self._client.get( + f"{self.base_url}/api/rater/next", + headers={"x-panel-site-key": self.site_key}, + params={"pool": pool, "rater_id": rater_id}, + ) + payload = _json_or_raw(response) + if not (200 <= response.status_code < 300): + parse_error_response(response.status_code, payload, response.text) + return payload + + async def submit_judgment(self, unit_id: str, choice: str) -> dict[str, Any]: + """Submit a rater judgment choice.""" + response = await self._client.post( + f"{self.base_url}/api/rater/judgment", + headers={"x-panel-site-key": self.site_key, "content-type": "application/json"}, + json={"unit_id": unit_id, "choice": choice}, + ) + payload = _json_or_raw(response) + if not (200 <= response.status_code < 300): + parse_error_response(response.status_code, payload, response.text) + return payload diff --git a/panel_sdk/types.py b/panel_sdk/types.py new file mode 100644 index 0000000..6baebae --- /dev/null +++ b/panel_sdk/types.py @@ -0,0 +1,59 @@ +"""Dataclass result types for panel-sdk.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Mapping + + +@dataclass +class VerifyResult: + """Parsed response for POST /v1/verify.""" + + ok: bool + trust: float | None = None + tier_used: str | None = None + unit_ids: list[str] | None = None + reason: str | None = None + + @classmethod + def from_json(cls, payload: Mapping[str, Any]) -> "VerifyResult": + """Construct VerifyResult from panel JSON response.""" + return cls( + ok=bool(payload.get("ok")), + trust=payload.get("trust"), + tier_used=payload.get("tier_used"), + unit_ids=payload.get("unit_ids"), + reason=payload.get("reason"), + ) + + +@dataclass +class TraceResult: + """Typed result for trace ingest sync/async behavior.""" + + status: str + trace_id: str + unit_ids: list[str] | None = None + structural_count: int | None = None + llm_count: int | None = None + skipped_count: int | None = None + poll_url: str | None = None + + @classmethod + def from_json(cls, payload: Mapping[str, Any], base_url: str) -> "TraceResult": + """Construct TraceResult from panel JSON response.""" + status = str(payload.get("status") or "done") + poll = payload.get("poll") + poll_url = None + if isinstance(poll, str): + poll_url = f"{base_url.rstrip('/')}{poll}" + return cls( + status=status, + trace_id=str(payload.get("trace_id", "")), + unit_ids=payload.get("unit_ids"), + structural_count=payload.get("structural_count"), + llm_count=payload.get("llm_count"), + skipped_count=payload.get("skipped_count"), + poll_url=poll_url, + ) diff --git a/pyproject.toml b/pyproject.toml index 4e8ccda..2eef602 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "panel-sdk" -version = "0.1.0" +version = "0.2.0" description = "thin server SDK for the panel HTTP api" requires-python = ">=3.10" dependencies = ["httpx>=0.27"] diff --git a/tests/test_client.py b/tests/test_client.py deleted file mode 100644 index 2c18259..0000000 --- a/tests/test_client.py +++ /dev/null @@ -1,91 +0,0 @@ -import base64 -import hashlib -import hmac -import json - -import httpx -import pytest -import respx - -from panel_sdk import AsyncPanelClient, PanelClient, PanelError, VerifyResult - -BASE = "https://p.test" -SITE_KEY = "pk_test_sdk" -SITE_SECRET = "site-secret-abc" -SCRUBBER = "scrubber-secret-xyz" - - -def _expected_sig(body: str) -> str: - return hmac.new(SITE_SECRET.encode(), body.encode(), hashlib.sha256).hexdigest() - - -@respx.mock -def test_ingest_unit_signs_body(): - seen = {} - - def handler(request: httpx.Request) -> httpx.Response: - seen["sig"] = request.headers.get("x-panel-ingest-sig") - seen["key"] = request.headers.get("x-panel-site-key") - seen["attest"] = request.headers.get("x-scrubber-attestation") - return httpx.Response(200, json={"id": "u_1"}) - - respx.post(f"{BASE}/api/units/ingest").mock(side_effect=handler) - c = PanelClient(base_url=BASE, site_key=SITE_KEY, site_secret=SITE_SECRET) - r = c.ingest_unit(type="step_validity", payload={"foo": 1}) - assert r == {"id": "u_1"} - assert seen["key"] == SITE_KEY - assert seen["attest"] is None # no scrubber secret → no attestation - assert seen["sig"] == _expected_sig(json.dumps({"type": "step_validity", "payload": {"foo": 1}}, separators=(",", ":"))) - - -@respx.mock -def test_ingest_trace_attaches_attestation_when_scrubber_secret_set(): - seen = {} - - def handler(request: httpx.Request) -> httpx.Response: - seen["attest"] = request.headers["x-scrubber-attestation"] - seen["body"] = request.content.decode() - return httpx.Response(200, json={"trace_id": "tr_1", "units_emitted": 5}) - - respx.post(f"{BASE}/api/v1/traces").mock(side_effect=handler) - c = PanelClient(base_url=BASE, site_key=SITE_KEY, site_secret=SITE_SECRET, scrubber_secret=SCRUBBER) - r = c.ingest_trace(trace_id="tr_1", source_agent="hermes", blob={"messages": []}) - assert r["units_emitted"] == 5 - parts = seen["attest"].split(".") - assert len(parts) == 3 - payload_b64 = parts[1] + "=" * (-len(parts[1]) % 4) - payload = json.loads(base64.urlsafe_b64decode(payload_b64)) - expected_output = hashlib.sha256(seen["body"].encode()).hexdigest() - assert payload["output_hash"] == expected_output - - -@respx.mock -def test_verify_token_returns_parsed_result(): - respx.post(f"{BASE}/v1/verify").mock( - return_value=httpx.Response(200, json={"ok": True, "trust": 0.9, "tier_used": "C1", "unit_ids": ["u_a"]}) - ) - c = PanelClient(base_url=BASE, site_key=SITE_KEY, site_secret=SITE_SECRET) - v = c.verify_token("t.t.t") - assert isinstance(v, VerifyResult) - assert v.ok is True and v.tier_used == "C1" and v.trust == 0.9 - - -@respx.mock -def test_raises_panel_error_on_non_2xx(): - respx.post(f"{BASE}/api/units/ingest").mock( - return_value=httpx.Response(422, json={"error": "scrubber_attestation_required"}) - ) - c = PanelClient(base_url=BASE, site_key=SITE_KEY, site_secret=SITE_SECRET) - with pytest.raises(PanelError) as ei: - c.ingest_unit(type="x", payload={}) - assert ei.value.status == 422 - assert ei.value.body == {"error": "scrubber_attestation_required"} - - -@respx.mock -async def test_async_client_parity(): - respx.post(f"{BASE}/v1/verify").mock(return_value=httpx.Response(200, json={"ok": True, "trust": 0.6, "tier_used": "C2"})) - async with httpx.AsyncClient() as session: - c = AsyncPanelClient(base_url=BASE, site_key=SITE_KEY, site_secret=SITE_SECRET, client=session) - v = await c.verify_token("z.z.z") - assert v.ok and v.tier_used == "C2" diff --git a/tests/test_sdk_sync.py b/tests/test_sdk_sync.py new file mode 100644 index 0000000..7ac91f1 --- /dev/null +++ b/tests/test_sdk_sync.py @@ -0,0 +1,125 @@ +import base64 +import hashlib +import hmac +import json + +import httpx +import pytest +import respx + +from panel_sdk import PanelClient, PanelRateLimitError, TraceResult + +BASE = "https://p.test" +SITE_KEY = "pk_test_sdk" +SITE_SECRET = "site-secret-abc" +SCRUBBER_SECRET = "scrubber-secret-xyz" + + +def test_hmac_signature_for_ingest_body() -> None: + seen: dict[str, str | None] = {} + + @respx.mock + def run() -> None: + def handler(request: httpx.Request) -> httpx.Response: + seen["sig"] = request.headers.get("x-panel-ingest-sig") + body = request.content.decode("utf-8") + expected = hmac.new(SITE_SECRET.encode("utf-8"), body.encode("utf-8"), hashlib.sha256).hexdigest() + assert seen["sig"] == expected + return httpx.Response(200, json={"ok": True}) + + respx.post(f"{BASE}/api/units/ingest").mock(side_effect=handler) + client = PanelClient(base_url=BASE, site_key=SITE_KEY, site_secret=SITE_SECRET) + client.ingest_units({"type": "process_output_rating", "passage": "hello"}) + + run() + + +@respx.mock +def test_hmac_signature_for_score_canonical_query() -> None: + seen: dict[str, str | None] = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["sig"] = request.headers.get("x-panel-ingest-sig") + return httpx.Response(200, json={"counts": {"yes": 1}, "trust_weighted_score": 0.8}) + + respx.get(f"{BASE}/api/units/score").mock(side_effect=handler) + client = PanelClient(base_url=BASE, site_key=SITE_KEY, site_secret=SITE_SECRET) + client.score_unit(ref="ext_123") + canonical = f"GET\n/api/units/score\nref=ext_123\nid=\nsite={SITE_KEY}" + expected = hmac.new(SITE_SECRET.encode("utf-8"), canonical.encode("utf-8"), hashlib.sha256).hexdigest() + assert seen["sig"] == expected + + +@respx.mock +def test_scrubber_self_sign_jwt_structure() -> None: + seen: dict[str, str] = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["token"] = request.headers["x-scrubber-attestation"] + seen["body"] = request.content.decode("utf-8") + return httpx.Response(200, json={"trace_id": "tr_1", "unit_ids": [], "structural_count": 1, "llm_count": 0, "skipped_count": 0}) + + respx.post(f"{BASE}/api/v1/traces").mock(side_effect=handler) + client = PanelClient( + base_url=BASE, + site_key=SITE_KEY, + site_secret=SITE_SECRET, + scrubber_mode="self-sign", + scrubber_secret=SCRUBBER_SECRET, + ) + client.ingest_trace(source_agent="agent", blob={"messages": []}, trace_id="tr_1") + parts = seen["token"].split(".") + assert len(parts) == 3 + payload_b64 = parts[1] + "=" * (-len(parts[1]) % 4) + payload = json.loads(base64.urlsafe_b64decode(payload_b64.encode("ascii"))) + assert payload["engine_version"] == "0.2.0" + assert payload["mode"] == "text" + assert payload["output_hash"] == hashlib.sha256(seen["body"].encode("utf-8")).hexdigest() + + +@respx.mock +def test_ingest_trace_and_wait_completes() -> None: + respx.post(f"{BASE}/api/v1/traces").mock( + return_value=httpx.Response(202, json={"trace_id": "tr_2", "status": "pending", "poll": "/v1/traces/tr_2"}) + ) + route = respx.get(f"{BASE}/api/v1/traces/tr_2").mock( + side_effect=[ + httpx.Response(202, json={"trace_id": "tr_2", "status": "pending"}), + httpx.Response(200, json={"trace_id": "tr_2", "status": "done", "unit_ids": ["u1"]}), + ] + ) + client = PanelClient(base_url=BASE, site_key=SITE_KEY, site_secret=SITE_SECRET) + result = client.ingest_trace_and_wait(source_agent="agent", blob={"messages": []}, trace_id="tr_2", max_wait_seconds=2, poll_interval_seconds=0) + assert result["status"] == "done" + assert route.call_count == 2 + + +@respx.mock +def test_429_retry_after_and_then_raises() -> None: + route = respx.post(f"{BASE}/api/units/ingest").mock( + side_effect=[ + httpx.Response(429, json={"error": "rate_limited", "scope": "ingest", "retry_after_s": 0}, headers={"Retry-After": "0"}), + httpx.Response(429, json={"error": "rate_limited", "scope": "ingest", "retry_after_s": 0}, headers={"Retry-After": "0"}), + ] + ) + client = PanelClient(base_url=BASE, site_key=SITE_KEY, site_secret=SITE_SECRET, max_retries=1) + with pytest.raises(PanelRateLimitError) as exc: + client.ingest_units({"type": "process_output_rating", "passage": "x"}) + assert exc.value.scope == "ingest" + assert exc.value.retry_after_s == 0.0 + assert route.call_count == 2 + + +@respx.mock +@pytest.mark.asyncio +async def test_async_ingest_trace_returns_typed_result() -> None: + from panel_sdk import AsyncPanelClient + + respx.post(f"{BASE}/api/v1/traces").mock( + return_value=httpx.Response(200, json={"trace_id": "tr_9", "unit_ids": ["u9"], "structural_count": 1, "llm_count": 1, "skipped_count": 0}) + ) + async with httpx.AsyncClient() as session: + client = AsyncPanelClient(base_url=BASE, site_key=SITE_KEY, site_secret=SITE_SECRET, client=session) + result = await client.ingest_trace(source_agent="agent", blob={"messages": []}, trace_id="tr_9") + assert isinstance(result, TraceResult) + assert result.status == "done"