client.py

"""
CapGlyph API SDK — typed HTTP client for capglyphd
OpenAPI surface: /v1/{seal,open,embed,verify,extract,info,consume,revoke} + /health
Works with httpx (preferred) or stdlib fallback.
"""

from __future__ import annotations

import json
import os
from dataclasses import dataclass
from typing import Any, Optional

try:
    import httpx  # type: ignore

    _HAS_HTTPX = True
except ImportError:
    _HAS_HTTPX = False
    httpx = None  # type: ignore


class CapglyphApiError(Exception):
    def __init__(self, code: str, message: str, status: int = 0):
        super().__init__(message)
        self.code = code
        self.status = status


@dataclass
class CapglyphdConfig:
    base_url: str
    api_key: Optional[str] = None


class CapglyphClient:
    """Typed HTTP client for capglyphd."""

    def __init__(self, base_url: str, api_key: Optional[str] = None, timeout: float = 30.0):
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.timeout = timeout
        if _HAS_HTTPX:
            headers = {}
            if api_key:
                headers["Authorization"] = f"Bearer {api_key}"
            self._client = httpx.Client(base_url=self.base_url, headers=headers, timeout=timeout)  # type: ignore
        else:
            self._client = None  # use urllib fallback per-request

    def _request(self, method: str, path: str, body: Any = None) -> Any:
        url = f"{self.base_url}{path}"
        headers = {"Content-Type": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"

        if _HAS_HTTPX:
            resp = self._client.request(
                method, path, json=body, headers=headers if not self._client else None
            )  # type: ignore
            text = resp.text
            try:
                data = json.loads(text) if text else None
            except Exception:
                data = None
            if resp.status_code >= 400:
                code = (
                    (data or {}).get("code")
                    or (data or {}).get("error")
                    or f"E_HTTP_{resp.status_code}"
                )
                msg = (
                    (data or {}).get("message")
                    or (data or {}).get("error")
                    or text
                    or resp.reason_phrase
                )
                raise CapglyphApiError(code, msg, resp.status_code)
            return data
        else:
            # stdlib fallback
            import urllib.error
            import urllib.request

            data_bytes = json.dumps(body).encode() if body is not None else None
            req = urllib.request.Request(url, data=data_bytes, method=method, headers=headers)
            try:
                with urllib.request.urlopen(req, timeout=self.timeout) as r:  # type: ignore
                    text = r.read().decode()
                    return json.loads(text) if text else None
            except urllib.error.HTTPError as e:
                text = e.read().decode()
                try:
                    data = json.loads(text) if text else None
                except Exception:
                    data = None
                code = (data or {}).get("code") or f"E_HTTP_{e.code}"
                msg = (data or {}).get("message") or text or str(e)
                raise CapglyphApiError(code, msg, e.code) from e

    # -- Framing (no image) ------------------------------------------

    def seal(
        self,
        payload_hex: str,
        k_mac_hex: str,
        version: int = 1,
        payload_type: int = 1,
        flags: int = 0,
    ) -> dict:
        return self._request(
            "POST",
            "/v1/seal",
            {
                "payload_hex": payload_hex,
                "k_mac_hex": k_mac_hex,
                "version": version,
                "payload_type": payload_type,
                "flags": flags,
            },
        )

    def open(self, sealed_hex: str, k_mac_hex: str) -> dict:
        return self._request("POST", "/v1/open", {"sealed_hex": sealed_hex, "k_mac_hex": k_mac_hex})

    def validate(self, sealed_hex: str) -> dict:
        return self._request("POST", "/v1/validate", {"sealed_hex": sealed_hex})

    # -- Image carrier ------------------------------------------------

    def embed_image(self, image_base64: str, mode: str, payload_hex: str, k_mac_hex: str) -> dict:
        return self._request(
            "POST",
            "/v1/embed",
            {
                "image_base64": image_base64,
                "mode": mode,
                "payload_hex": payload_hex,
                "k_mac_hex": k_mac_hex,
            },
        )

    def verify_image(self, image_base64: str, mode: str, k_mac_hex: str | None = None) -> dict:
        body: dict[str, Any] = {"image_base64": image_base64, "mode": mode}
        if k_mac_hex:
            body["k_mac_hex"] = k_mac_hex
        return self._request("POST", "/v1/verify", body)

    def extract_image(self, image_base64: str, mode: str, id_length: int = 16) -> dict:
        return self._request(
            "POST",
            "/v1/extract",
            {"image_base64": image_base64, "mode": mode, "id_length": id_length},
        )

    # -- Credential lifecycle -----------------------------------------

    def consume(self, token_id: str, now: str | None = None) -> dict:
        body: dict[str, Any] = {"token_id": token_id}
        if now:
            body["now"] = now
        return self._request("POST", "/v1/consume", body)

    def revoke(self, token_id: str, reason: str | None = None) -> dict:
        body: dict[str, Any] = {"token_id": token_id}
        if reason:
            body["reason"] = reason
        return self._request("POST", "/v1/revoke", body)

    def info(self, image_base64: str, mode: str | None = None) -> Any:
        body: dict[str, Any] = {"image_base64": image_base64}
        if mode:
            body["mode"] = mode
        return self._request("POST", "/v1/info", body)

    def health(self) -> dict:
        return self._request("GET", "/health")

    def close(self):
        if _HAS_HTTPX and self._client:
            self._client.close()

    def __enter__(self):
        return self

    def __exit__(self, *_):
        self.close()


def client_from_env() -> CapglyphClient:
    base_url = os.environ.get("CAPGLYPH_API_URL", "http://localhost:8080")
    api_key = os.environ.get("CAPGLYPH_API_KEY")
    return CapglyphClient(base_url, api_key)