halvorai-ingest 0.5.0: device helpers for Flutter apps

CHANGELOG.md

@@ -1,5 +1,18 @@
 # Changelog
 
+## 0.5.0
+
+### Added
+
+- `halvorai_ingest.device`, the backend half of mobile apps built with `halvorai_ingest_flutter`, whose installs sign their own events:
+  - `register_device_source` registers an install's public Ed25519 key (a key with `d` is refused).
+  - `forward_envelopes` checks each event's owner and environment, then passes the batch to `POST /v1/events` byte for byte and returns HalvorAI's answer unchanged.
+  - Both have `_async` versions for FastAPI and other async frameworks. Same behaviour as `@halvorai/ingest/device` 0.4.0.
+
+### Changed
+
+- `init` registers its source through the same code. A refused registration still raises `RuntimeError` (now `SourceRegistrationError`, a subclass), and a network failure says `POST /v1/sources failed: <reason>`.
+
 ## 0.4.1
 
 ### Fixed

README.md

@@ -115,6 +115,61 @@ Each 402 challenge becomes a `payment.requested` event. The payment that answers
 
 Stripe, MPP and MCP adapters are in the Node.js SDK, [`@halvorai/ingest`](https://github.com/halvorAI/ingest-js). Guides for each: [docs.halvorai.com](https://docs.halvorai.com).
 
+## Mobile apps (Flutter)
+
+Apps built with [`halvorai_ingest_flutter`](https://github.com/halvorAI/ingest-dart) sign events on the phone, with a key that each install makes for itself. The app never holds your API key. Your backend registers each install's public key and forwards its signed batches; `halvorai_ingest.device` provides both. With FastAPI:
+
+```python
+from fastapi import FastAPI, Request, Response
+
+from halvorai_ingest.device import (
+    DeviceRequestError,
+    forward_envelopes_async,
+    register_device_source_async,
+)
+
+app = FastAPI()
+UNAUTHORIZED = b'{"error":{"code":"unauthorized","message":"sign in first"}}'
+
+
+@app.post("/halvor/sources")  # the app's provisionUrl
+async def halvor_sources(request: Request) -> Response:
+    user = await current_user(request)  # your own session check
+    if user is None:
+        return Response(UNAUTHORIZED, status_code=401, media_type="application/json")
+    body = await request.json()
+    try:
+        source_id = await register_device_source_async(body.get("public_jwk"), name=body.get("name"))
+    except DeviceRequestError as exc:
+        return Response(exc.body(), status_code=exc.status, media_type="application/json")
+    return Response(f'{{"source_id":"{source_id}"}}', media_type="application/json")
+
+
+@app.post("/halvor/events")  # the app's forwardUrl
+async def halvor_events(request: Request) -> Response:
+    user = await current_user(request)
+    if user is None:
+        return Response(UNAUTHORIZED, status_code=401, media_type="application/json")
+    out = await forward_envelopes_async(await request.body(), expected_owner_id=user.id)
+    return Response(out.body, status_code=out.status, media_type=out.content_type)
+```
+
+Both functions read `HALVOR_API_KEY` (and `HALVOR_ENDPOINT`) from the environment, or take `api_key=`. Sync versions without `_async` work the same way in Flask or Django.
+
+What they do:
+
+- **`register_device_source`** accepts only the public half of an Ed25519 key, refusing a key with `d`. It registers the key with `POST /v1/sources` and returns the `src_` id.
+- **`forward_envelopes`**:
+  - Checks that every event's `owner.id` is the signed-in user and its `environment` is the key's.
+  - Sends the request body to `POST /v1/events` **byte for byte**, never a parsed and re-serialized copy, so the device's signatures stay valid.
+  - Returns HalvorAI's status and body unchanged, which the app reads its per-event results from.
+- **Status codes tell the app what to do:**
+  - 401 or 403 means "try later": the app keeps the batch on disk and retries.
+  - 400, 413 and 422 are permanent: the app drops the batch, or splits it on 413.
+  - 502 means HalvorAI couldn't be reached, and the app retries.
+
+Set the signed-in user as `owner` in the app's agent context; events without it are refused.
+
 ## Develop
 
 ```bash

halvorai_ingest/device.py

@@ -0,0 +1,269 @@
+"""Server-side helpers for apps whose installs sign their own events (`halvorai_ingest_flutter`).
+
+The app never holds your API key. Your backend:
+
+- registers each install's public key (`register_device_source`) and answers the app's
+  provisioning request with `{"source_id": "src_…"}`;
+- checks and forwards each signed batch (`forward_envelopes`), answering with HalvorAI's response
+  unchanged.
+
+Both come in sync and async forms; use the async ones in FastAPI and other async frameworks.
+The status codes tell the app what to do: 401 or 403 (your own session check) means try later,
+400, 413 and 422 are permanent, and 502 means HalvorAI could not be reached.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+from dataclasses import dataclass
+from typing import Any
+
+import httpx
+
+from .client import API_KEY_PATTERN, default_endpoint
+from .sources import SourceRegistrationError, post_source, post_source_async
+
+DEFAULT_MAX_ENVELOPES = 100
+DEFAULT_MAX_BYTES = 1_000_000
+
+_BASE64URL_32_BYTES = re.compile(r"^[A-Za-z0-9_-]{43}$")
+
+
+class DeviceRequestError(Exception):
+    """A request the app sent that your backend must answer with `status`.
+
+    `body()` is HalvorAI's error format, so the Flutter SDK reports `code` as `serverCode`.
+    """
+
+    def __init__(self, status: int, code: str, message: str) -> None:
+        super().__init__(message)
+        self.status = status
+        self.code = code
+        self.message = message
+
+    def body(self) -> bytes:
+        return json.dumps({"error": {"code": self.code, "message": self.message}}).encode("utf-8")
+
+
+@dataclass(frozen=True)
+class ForwardResult:
+    """What to answer the app: HalvorAI's response unchanged, or why the batch was not sent."""
+
+    status: int
+    body: bytes
+    content_type: str
+
+
+@dataclass(frozen=True)
+class _Config:
+    api_key: str
+    environment: str
+    endpoint: str
+
+
+def _config(api_key: str | None, endpoint: str | None) -> _Config:
+    """A misconfigured backend is a programming error, so this raises ValueError."""
+    key = api_key or os.environ.get("HALVOR_API_KEY")
+    if not key or not API_KEY_PATTERN.match(key):
+        raise ValueError("HalvorAI api key is missing or malformed (expected hk_test_… or hk_live_…)")
+    environment = "live" if key.startswith("hk_live_") else "test"
+    base = endpoint or os.environ.get("HALVOR_ENDPOINT") or default_endpoint(key, environment)
+    base = base.rstrip("/")
+    if base.endswith("/v1/events"):
+        base = base[: -len("/v1/events")]
+    return _Config(key, environment, base)
+
+
+def public_device_jwk(jwk: object) -> dict[str, str]:
+    """The public half of an Ed25519 key, reduced to `kty`, `crv` and `x`.
+
+    A key with `d` is refused: the app sent its private key, which must never leave the device.
+    """
+    if not isinstance(jwk, dict):
+        raise DeviceRequestError(400, "invalid_request", "public_jwk must be a JSON object")
+    if "d" in jwk:
+        raise DeviceRequestError(
+            400, "invalid_request", "public_jwk contains a private key (d); only the public half may be sent"
+        )
+    if jwk.get("kty") != "OKP" or jwk.get("crv") != "Ed25519":
+        raise DeviceRequestError(400, "invalid_request", "public_jwk must be an Ed25519 key")
+    x = jwk.get("x")
+    if not isinstance(x, str) or not _BASE64URL_32_BYTES.match(x):
+        raise DeviceRequestError(400, "invalid_request", "public_jwk.x must be 32 bytes, base64url")
+    return {"kty": "OKP", "crv": "Ed25519", "x": x}
+
+
+def _name(name: str | None) -> str:
+    return ((name or "").strip() or "app install")[:100]
+
+
+def register_device_source(
+    public_jwk: object,
+    *,
+    name: str | None = None,
+    api_key: str | None = None,
+    endpoint: str | None = None,
+    transport: httpx.BaseTransport | None = None,
+) -> str:
+    """Register an app install's public key and return the `src_` id the app signs with.
+
+    Call it from the route the app's provisioning request reaches, after checking the user is
+    signed in. Raises `DeviceRequestError`: 400 for a key the app should not have sent, 502 when
+    HalvorAI does not register it (the app retries).
+    """
+    config = _config(api_key, endpoint)
+    jwk = public_device_jwk(public_jwk)
+    try:
+        with httpx.Client(timeout=30.0, transport=transport) as http:
+            return post_source(http, config.endpoint, config.api_key, _name(name), jwk)
+    except SourceRegistrationError as exc:
+        raise DeviceRequestError(502, "upstream_error", str(exc)) from exc
+
+
+async def register_device_source_async(
+    public_jwk: object,
+    *,
+    name: str | None = None,
+    api_key: str | None = None,
+    endpoint: str | None = None,
+    transport: httpx.AsyncBaseTransport | None = None,
+) -> str:
+    """`register_device_source` for async frameworks."""
+    config = _config(api_key, endpoint)
+    jwk = public_device_jwk(public_jwk)
+    try:
+        async with httpx.AsyncClient(timeout=30.0, transport=transport) as http:
+            return await post_source_async(http, config.endpoint, config.api_key, _name(name), jwk)
+    except SourceRegistrationError as exc:
+        raise DeviceRequestError(502, "upstream_error", str(exc)) from exc
+
+
+def _check_batch(body: bytes, expected_owner_id: str, environment: str, max_envelopes: int, max_bytes: int) -> None:
+    if len(body) > max_bytes:
+        raise DeviceRequestError(413, "payload_too_large", f"body is over {max_bytes} bytes")
+    try:
+        parsed: Any = json.loads(body.decode("utf-8"))
+    except (UnicodeDecodeError, ValueError):
+        raise DeviceRequestError(400, "invalid_request", "body is not UTF-8 JSON") from None
+    events = parsed.get("events") if isinstance(parsed, dict) else None
+    if not isinstance(events, list) or not events:
+        raise DeviceRequestError(400, "invalid_request", 'body must be {"events":[…]} with at least one envelope')
+    if len(events) > max_envelopes:
+        raise DeviceRequestError(413, "payload_too_large", f"more than {max_envelopes} envelopes")
+    for i, envelope in enumerate(events):
+        event = envelope.get("event") if isinstance(envelope, dict) else None
+        if not isinstance(event, dict) or not isinstance(envelope.get("sig"), str):
+            raise DeviceRequestError(400, "invalid_request", f'events[{i}] must be {{"event":{{…}},"sig":"…"}}')
+        owner = event.get("owner")
+        if not isinstance(owner, dict) or owner.get("id") != expected_owner_id:
+            raise DeviceRequestError(422, "owner_mismatch", f"events[{i}].event.owner.id is not the signed-in user")
+        if event.get("environment") != environment:
+            raise DeviceRequestError(
+                422,
+                "environment_mismatch",
+                f"events[{i}].event.environment must be {environment}, the API key's environment",
+            )
+
+
+def _prepare(
+    body: bytes | str,
+    expected_owner_id: str,
+    api_key: str | None,
+    endpoint: str | None,
+    max_envelopes: int,
+    max_bytes: int,
+) -> tuple[_Config, bytes, ForwardResult | None]:
+    config = _config(api_key, endpoint)
+    raw = body.encode("utf-8") if isinstance(body, str) else bytes(body)
+    try:
+        _check_batch(raw, expected_owner_id, config.environment, max_envelopes, max_bytes)
+    except DeviceRequestError as exc:
+        return config, raw, _error(exc)
+    return config, raw, None
+
+
+def _error(exc: DeviceRequestError) -> ForwardResult:
+    return ForwardResult(exc.status, exc.body(), "application/json")
+
+
+def _headers(config: _Config) -> dict[str, str]:
+    return {
+        "Authorization": f"Bearer {config.api_key}",
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+    }
+
+
+def _result(res: httpx.Response) -> ForwardResult:
+    return ForwardResult(res.status_code, res.content, res.headers.get("content-type", "application/json"))
+
+
+def _unreachable(exc: Exception) -> ForwardResult:
+    return _error(DeviceRequestError(502, "upstream_error", f"HalvorAI could not be reached: {exc}"))
+
+
+def forward_envelopes(
+    body: bytes | str,
+    *,
+    expected_owner_id: str,
+    api_key: str | None = None,
+    endpoint: str | None = None,
+    max_envelopes: int = DEFAULT_MAX_ENVELOPES,
+    max_bytes: int = DEFAULT_MAX_BYTES,
+    transport: httpx.BaseTransport | None = None,
+) -> ForwardResult:
+    """Check a batch an app install sent and pass it to `POST /v1/events` byte for byte.
+
+    Returns HalvorAI's status and body unchanged; the app reads its per-event results from them.
+    The body is parsed only to check it: re-serializing could change the bytes the device signed.
+
+    Refused without calling HalvorAI, in HalvorAI's error format: 400 `invalid_request` for a body
+    that is not `{"events":[{"event":…,"sig":…}]}`, 413 `payload_too_large` above `max_envelopes`
+    or `max_bytes` (the app splits the batch), 422 `owner_mismatch` when an event's `owner.id` is
+    not `expected_owner_id`, 422 `environment_mismatch` when its `environment` is not the key's.
+    502 when HalvorAI cannot be reached. Answer 401 or 403 yourself for a session that is not valid.
+    """
+    config, raw, refused = _prepare(body, expected_owner_id, api_key, endpoint, max_envelopes, max_bytes)
+    if refused is not None:
+        return refused
+    try:
+        with httpx.Client(timeout=30.0, transport=transport) as http:
+            res = http.post(f"{config.endpoint}/v1/events", content=raw, headers=_headers(config))
+    except httpx.HTTPError as exc:
+        return _unreachable(exc)
+    return _result(res)
+
+
+async def forward_envelopes_async(
+    body: bytes | str,
+    *,
+    expected_owner_id: str,
+    api_key: str | None = None,
+    endpoint: str | None = None,
+    max_envelopes: int = DEFAULT_MAX_ENVELOPES,
+    max_bytes: int = DEFAULT_MAX_BYTES,
+    transport: httpx.AsyncBaseTransport | None = None,
+) -> ForwardResult:
+    """`forward_envelopes` for async frameworks."""
+    config, raw, refused = _prepare(body, expected_owner_id, api_key, endpoint, max_envelopes, max_bytes)
+    if refused is not None:
+        return refused
+    try:
+        async with httpx.AsyncClient(timeout=30.0, transport=transport) as http:
+            res = await http.post(f"{config.endpoint}/v1/events", content=raw, headers=_headers(config))
+    except httpx.HTTPError as exc:
+        return _unreachable(exc)
+    return _result(res)
+
+
+__all__ = [
+    "DeviceRequestError",
+    "ForwardResult",
+    "forward_envelopes",
+    "forward_envelopes_async",
+    "public_device_jwk",
+    "register_device_source",
+    "register_device_source_async",
+]

halvorai_ingest/init.py

@@ -13,6 +13,7 @@
 import httpx
 
 from .client import API_KEY_PATTERN, HalvorClient, default_endpoint
+from .sources import post_source
 from .sign import generate_source_key
 
 DASHBOARD_URL = "https://app.halvorai.com"
@@ -67,22 +68,6 @@ def upsert_env_file(path: Path, values: dict[str, str]) -> None:
     path.write_text(content, encoding="utf-8")
 
 
-def _register_source(
-    http: httpx.Client, endpoint: str, api_key: str, name: str, public_jwk: dict[str, str]
-) -> str:
-    res = http.post(
-        f"{endpoint}/v1/sources",
-        json={"name": name, "adapter": "custom", "public_jwk": public_jwk},
-        headers={"Authorization": f"Bearer {api_key}", "Accept": "application/json"},
-    )
-    if res.status_code != 201:
-        raise RuntimeError(f"POST /v1/sources failed ({res.status_code}): {res.text or res.reason_phrase}")
-    source_id = res.json().get("id")
-    if not isinstance(source_id, str) or not source_id.startswith("src_"):
-        raise RuntimeError("POST /v1/sources returned no source id")
-    return source_id
-
-
 def run_init(
     api_key: str,
     *,
@@ -110,7 +95,7 @@ def run_init(
 
     log(f"Registering source at {base}…")
     with httpx.Client(timeout=30.0, transport=transport) as http:
-        source_id = _register_source(http, base, api_key, source_name, public_jwk)
+        source_id = post_source(http, base, api_key, source_name, public_jwk)
     private_jwk["kid"] = source_id
 
     env_path = Path(cwd).resolve() / ".env.local"