diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml new file mode 100644 index 0000000..f53b178 --- /dev/null +++ b/.github/workflows/publish.yml @@ -0,0 +1,43 @@ +name: publish + +# Push a tag vX.Y.Z (matching src/hibira/_version.py) to publish to PyPI with +# trusted publishing (PyPI project "hibira", workflow publish.yml, +# environment pypi). +on: + push: + tags: ["v*"] + +jobs: + build: + runs-on: ubuntu-latest + permissions: + contents: read + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - name: Check the tag matches the version + run: | + version=$(python -c "exec(open('src/hibira/_version.py').read()); print(__version__)") + test "v$version" = "$GITHUB_REF_NAME" || { echo "tag $GITHUB_REF_NAME does not match version $version"; exit 1; } + - run: pip install -e . pytest fastapi build + - run: pytest -q + - run: python -m build + - uses: actions/upload-artifact@v4 + with: + name: dist + path: dist/ + + publish: + needs: build + runs-on: ubuntu-latest + environment: pypi + permissions: + id-token: write + steps: + - uses: actions/download-artifact@v4 + with: + name: dist + path: dist/ + - uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..7be6c30 --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,48 @@ +name: test + +on: + push: + branches: [main] + pull_request: + +permissions: + contents: read + +jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - run: pip install -e . fastapi ruff mypy + - run: ruff check . + - run: ruff format --check src + - run: mypy + + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python: ["3.9", "3.10", "3.11", "3.12", "3.13"] + steps: + - uses: actions/checkout@v4 + # setup-uv installs any CPython version; setup-python has no 3.9 on newer runner images. + - uses: astral-sh/setup-uv@v6 + with: + python-version: ${{ matrix.python }} + - run: uv pip install -e . pytest fastapi + - run: pytest -q + + build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - run: pip install build twine + - run: python -m build + - run: twine check dist/* diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..86daf29 --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +.venv/ +__pycache__/ +*.pyc +dist/ +build/ +*.egg-info/ +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.env diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..7de2758 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,12 @@ +# Changelog + +## 0.1.0 + +First release. + +- `Client` and `AsyncClient`: `trigger`, `trigger_batch`, `channels`, `channel_info`, `users`, `terminate_user_connections`, settings from `HIBIRA_*` environment variables. +- Channel and user auth (`authenticate_channel`, `authenticate_user`) and webhook verification (`verify_webhook`). +- Hibira Push: `push_to_interests`, `push_to_users`, `push_token`, `delete_push_user`. +- Approvals: `request_approval`, `get_approval`, `wait_for_approval`. +- `stream_to_channel` / `astream_to_channel`: stream LLM output to `useStream()` in `@hibira-io/react` and `@hibira-io/vue`, with `text_of` for OpenAI, Anthropic, OpenAI Agents SDK, LangChain / LangGraph and Gemini chunks. +- `hibira.fastapi`: channel / user auth router and webhook dependency. diff --git a/README.md b/README.md new file mode 100644 index 0000000..71db114 --- /dev/null +++ b/README.md @@ -0,0 +1,266 @@ +# hibira + +Python server library for [Hibira](https://hibira.io), hosted realtime that +speaks the Pusher Channels protocol. Publish events, authorize private and +presence channels, verify webhooks, send push notifications, ask a person for +an approval, and stream LLM output to the browser as it is written. + +```sh +pip install hibira +``` + +Python 3.9+, one dependency ([httpx](https://www.python-httpx.org/)). Both a +regular and an asyncio client. pusher-js, Laravel Echo and the +`@hibira-io/*` front-end packages connect to Hibira unchanged. + +## Set up + +Create an app at [hibira.io/portal](https://hibira.io/portal) (the free plan +needs no card) and put its keys in the environment: + +```sh +HIBIRA_APP_ID=... +HIBIRA_APP_KEY=... +HIBIRA_APP_SECRET=... +``` + +```python +import hibira + +client = hibira.Client.from_env() # or hibira.Client(app_id, key, secret) +client.trigger("orders", "created", {"id": 1042}) +``` + +```python +async with hibira.AsyncClient.from_env() as client: + await client.trigger(["orders", "dashboard"], "created", {"id": 1042}) +``` + +| Setting | Environment | Default | +| --- | --- | --- | +| `app_id`, `key`, `secret` | `HIBIRA_APP_ID`, `HIBIRA_APP_KEY`, `HIBIRA_APP_SECRET` | | +| `host` | `HIBIRA_HOST` | `ws.hibira.io` (or the host shown on your app page) | +| `port`, `use_tls` | `HIBIRA_PORT`, `HIBIRA_USE_TLS` | `443`, `True` | +| `push_instance_id`, `push_secret_key` | `HIBIRA_PUSH_INSTANCE_ID`, `HIBIRA_PUSH_SECRET_KEY` | | +| `push_endpoint` | `HIBIRA_PUSH_ENDPOINT` | `https://` | +| `timeout` | | `10.0` seconds | + +Keyword arguments to `from_env()` win over the environment. Both clients +accept `http_client=` to reuse your own `httpx.Client` / `httpx.AsyncClient`. + +## Stream LLM output to the browser + +`stream_to_channel` sends a stream of text to a channel as it is produced, in +the format `useStream()` in [@hibira-io/react](https://github.com/hibira-io/react) +and [@hibira-io/vue](https://github.com/hibira-io/vue) puts back together. +Tokens are batched (about 10 events a second rather than one per token), each +event stays under the plan's message size, and the last event carries the +whole text when it fits, so people who open the page late catch up. With +channel history on (paid plans), `useStream(channel, { rewind: true })` +replays a stream after a reload. + +```python +from openai import OpenAI +import hibira + +client = hibira.Client.from_env() +openai = OpenAI() + +stream = openai.chat.completions.create(model="gpt-5-mini", messages=messages, stream=True) +result = hibira.stream_to_channel(client, f"private-conversation-{conversation_id}", stream, meta={"role": "assistant"}) +save_message(conversation_id, result.text) +``` + +```jsx +// React +const { text, done } = useStream(`private-conversation-${conversationId}`); +``` + +`text_of` reads the text out of chunks from the common SDKs, so you can pass +their streams as they are: + +| Source | Pass | +| --- | --- | +| OpenAI chat completions | `client.chat.completions.create(..., stream=True)` | +| OpenAI Responses | `client.responses.create(..., stream=True)` | +| Anthropic | `client.messages.create(..., stream=True)` (or the async client) | +| OpenAI Agents SDK | `Runner.run_streamed(agent, input).stream_events()` | +| LangGraph / LangChain | `graph.astream(input, stream_mode="messages")`, `llm.astream(...)` | +| Gemini (google-genai) | `client.models.generate_content_stream(...)` | +| Anything else | an iterable of `str` or `bytes`, or pass `pick=lambda chunk: ...` | + +With asyncio, use `astream_to_channel` (it takes async or plain iterables, +and an `AsyncClient` or a `Client`): + +```python +async def answer(conversation_id: str, question: str) -> str: + stream = graph.astream({"messages": [("user", question)]}, stream_mode="messages") + result = await hibira.astream_to_channel(client, f"private-conversation-{conversation_id}", stream) + return result.text +``` + +To stream only one LangGraph node, filter with `pick`: +`pick=lambda c: hibira.text_of(c) if c[1].get("langgraph_node") == "agent" else ""`. + +| Option | Default | | +| --- | --- | --- | +| `id` | random UUID | stream id, also in the result | +| `event` | `"ai-stream"` | event name | +| `interval` | `0.1` | seconds to gather tokens before sending them | +| `max_bytes` | `3500` | largest event payload (Free plans allow 4 KB per message) | +| `meta` | | JSON sent with the first event (model, message id, author) | +| `pick` | `text_of` | text of one chunk | +| `socket_id` | | leave out this connection (the sender's own tab) | +| `stop` | | a `threading.Event`; set it to end the stream with error `"aborted"` (sync only; cancel the task with asyncio) | + +If the source raises, the final event carries the error (useStream shows it) +and the exception is raised again. Both return a `StreamResult(id, text, events)`. + +## Authorize private and presence channels + +```python +client.authenticate_channel("private-user-42", socket_id) +# {"auth": "key:signature"} + +client.authenticate_channel("presence-room-7", socket_id, {"user_id": "42", "user_info": {"name": "Ann"}}) +# {"auth": "...", "channel_data": "{\"user_id\":\"42\",...}"} + +client.authenticate_user(socket_id, {"id": "42", "name": "Ann"}) # user sign-in +``` + +Return the dict as JSON from the endpoint pusher-js calls +(`channelAuthorization.endpoint`). It receives `socket_id` and +`channel_name` as form fields. These methods need no network and are plain +functions on `AsyncClient` too. + +### FastAPI + +```sh +pip install "hibira[fastapi]" +``` + +```python +from fastapi import Depends, FastAPI +from hibira import AsyncClient +from hibira.fastapi import auth_router, webhook + +client = AsyncClient.from_env() +app = FastAPI() + + +async def authorize(request, channel: str): + user = await current_user(request) + if channel == f"private-user-{user.id}": + return True + if channel.startswith("presence-room-"): + return {"user_id": str(user.id), "user_info": {"name": user.name}} + return False + + +app.include_router(auth_router(client, authorize)) # POST /hibira/auth + + +@app.post("/hibira/webhook") +async def on_webhook(payload: dict = Depends(webhook(client))): + for event in payload["events"]: + ... +``` + +`auth_router(client, authorize, user=...)` also answers user sign-in at +`/hibira/user-auth`. Django support is in +[django-hibira](https://github.com/hibira-io/django). + +## Webhooks + +```python +payload = client.verify_webhook( + request.headers.get("X-Pusher-Key"), + request.headers.get("X-Pusher-Signature"), + request.body, # the raw body, bytes or str +) +``` + +Raises `hibira.InvalidWebhook` when the key or signature does not match. + +## Channels API + +```python +client.trigger_batch([{"channel": "a", "name": "e", "data": {"x": 1}}]) +client.channels(prefix="presence-", info="user_count") +client.channel_info("presence-room-7", info=["user_count", "subscription_count"]) +client.users("presence-room-7") +client.terminate_user_connections("42") +``` + +`trigger` takes one channel or a list of up to 100. Data that is not a `str` +is sent as JSON. `socket_id=` leaves out the sender's own connection. + +## Push notifications + +Turn on Push for the app in the portal and set `HIBIRA_PUSH_INSTANCE_ID` and +`HIBIRA_PUSH_SECRET_KEY`. Bodies use the Beams format. + +```python +client.push_to_interests( + ["news"], + { + "web": {"notification": {"title": "New post", "deep_link": "https://example.com/p/1"}}, + "fcm": {"notification": {"title": "New post"}}, + "apns": {"aps": {"alert": {"title": "New post"}}}, + }, +) +client.push_to_users(["42"], {...}) +client.push_token("42") # {"token": ...} for setUserId in the web, Android and iOS SDKs +client.delete_push_user("42") +``` + +## Approvals + +Ask a person for a decision from a script, worker or agent. Hibira hosts the +answer page. + +```python +approval = client.request_approval( + "Refund order #1042 for $89?", + details="Customer says the parcel never arrived.", + options=["Refund", "Decline"], # the first one approves + push_interests=["ops"], # or push_users=[...], or share approval["url"] + channel="private-hibira-approvals.ops", # shows in the @hibira-io/approvals inbox +) +result = client.wait_for_approval(approval["approval_id"], timeout=3600) +if result["status"] == "approved": + refund(1042) +``` + +`status` is `pending`, `approved`, `rejected` or `expired`. `get_approval(id, wait=25)` +reads it once, waiting up to 25 seconds for an answer. + +## Errors + +Every error is a `hibira.HibiraError`. HTTP errors are `hibira.HTTPError` +(`status`, `message`, `body`), with subclasses `AuthenticationError` (401, +403), `NotFound` (404), `PayloadTooLarge` (413, the event is over the plan's +message size) and `RateLimited` (429). Missing settings raise `ConfigError`. + +## Development + +```sh +python -m venv .venv && .venv/bin/pip install -e . pytest fastapi ruff mypy +.venv/bin/pytest +.venv/bin/ruff check . && .venv/bin/mypy +``` + +`tests/test_e2e.py` runs against a real server when `HIBIRA_E2E=1` and the +`HIBIRA_*` settings point at it (it needs `pip install websockets`). + +## 中文摘要 + +`hibira` 是 Hibira 的 Python 伺服器端套件（Python 3.9 以上，同步與 asyncio 兩種 +client）：發送事件、私人 / presence 頻道授權、驗證 webhook、推播通知、人工核准， +以及用 `stream_to_channel` 把 LLM 的輸出即時串流到瀏覽器（React / Vue 的 +`useStream()` 直接接收）。支援 OpenAI、Anthropic、OpenAI Agents SDK、LangGraph、 +Gemini 的串流格式，並附 FastAPI 授權路由與 webhook 驗證。 + +## License + +MIT diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..34c666b --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,71 @@ +[build-system] +requires = ["hatchling>=1.24"] +build-backend = "hatchling.build" + +[project] +name = "hibira" +dynamic = ["version"] +description = "Hibira for Python: realtime events, channel auth, webhooks, push notifications, approvals and LLM streaming (Pusher-compatible)" +readme = "README.md" +license = "MIT" +license-files = ["LICENSE"] +requires-python = ">=3.9" +authors = [{ name = "Hibira", email = "dev@hibira.io" }] +keywords = ["hibira", "pusher", "websocket", "realtime", "channels", "push notifications", "llm", "streaming", "langgraph", "fastapi"] +classifiers = [ + "Development Status :: 4 - Beta", + "Framework :: AsyncIO", + "Framework :: FastAPI", + "Intended Audience :: Developers", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3 :: Only", + "Topic :: Internet :: WWW/HTTP", + "Topic :: Software Development :: Libraries", + "Typing :: Typed", +] +dependencies = ["httpx>=0.24,<1"] + +[project.optional-dependencies] +fastapi = ["fastapi>=0.100"] + +[project.urls] +Homepage = "https://hibira.io" +Documentation = "https://github.com/hibira-io/python#readme" +Source = "https://github.com/hibira-io/python" +Changelog = "https://github.com/hibira-io/python/blob/main/CHANGELOG.md" +Issues = "https://github.com/hibira-io/python/issues" + +[dependency-groups] +dev = ["pytest>=8", "fastapi>=0.100", "ruff>=0.6", "mypy>=1.10"] + +[tool.hatch.version] +path = "src/hibira/_version.py" + +[tool.hatch.build.targets.wheel] +packages = ["src/hibira"] + +[tool.hatch.build.targets.sdist] +include = ["src", "tests", "README.md", "CHANGELOG.md", "LICENSE"] + +[tool.pytest.ini_options] +testpaths = ["tests"] + +[tool.ruff] +line-length = 120 +target-version = "py39" + +[tool.ruff.lint] +select = ["E", "F", "W", "I", "B", "UP", "BLE"] +ignore = ["UP006", "UP007", "UP035", "UP045"] + +[tool.ruff.lint.flake8-bugbear] +extend-immutable-calls = ["fastapi.Depends"] + +[tool.ruff.lint.per-file-ignores] +"tests/*" = ["B008"] + +[tool.mypy] +python_version = "3.10" +strict = true +files = ["src/hibira"] diff --git a/src/hibira/__init__.py b/src/hibira/__init__.py new file mode 100644 index 0000000..248f55a --- /dev/null +++ b/src/hibira/__init__.py @@ -0,0 +1,41 @@ +"""Hibira for Python: publish realtime events, authorize private and presence +channels, verify webhooks, send push notifications, ask people for approvals +and stream LLM output to browsers. + +Hibira speaks the Pusher Channels protocol, so pusher-js, Laravel Echo and +the @hibira-io/* front-end packages connect to it unchanged. +""" + +from ._client import DEFAULT_HOST, AsyncClient, Client +from ._errors import ( + AuthenticationError, + ConfigError, + HibiraError, + HTTPError, + InvalidWebhook, + NotFound, + PayloadTooLarge, + RateLimited, +) +from ._version import __version__ +from .stream import STREAM_EVENT, StreamResult, astream_to_channel, stream_to_channel, text_of + +__all__ = [ + "DEFAULT_HOST", + "STREAM_EVENT", + "AsyncClient", + "AuthenticationError", + "Client", + "ConfigError", + "HTTPError", + "HibiraError", + "InvalidWebhook", + "NotFound", + "PayloadTooLarge", + "RateLimited", + "StreamResult", + "__version__", + "astream_to_channel", + "stream_to_channel", + "text_of", +] diff --git a/src/hibira/_client.py b/src/hibira/_client.py new file mode 100644 index 0000000..94cea22 --- /dev/null +++ b/src/hibira/_client.py @@ -0,0 +1,472 @@ +from __future__ import annotations + +import os +import time +from dataclasses import dataclass, field +from typing import Any, Iterable, Mapping, Sequence +from urllib.parse import quote, unquote + +import httpx + +from . import _signing +from ._errors import ConfigError, error_for +from ._version import __version__ + +DEFAULT_HOST = "ws.hibira.io" +MAX_TRIGGER_CHANNELS = 100 +MAX_APPROVAL_WAIT = 25 + +_USER_AGENT = f"hibira-python/{__version__}" + + +@dataclass +class _Request: + method: str + url: str + params: dict[str, str] = field(default_factory=dict) + content: bytes | None = None + headers: dict[str, str] = field(default_factory=dict) + + +def _env_bool(value: str | None) -> bool | None: + if value is None or value == "": + return None + return value.strip().lower() not in ("0", "false", "no", "off") + + +def _channels(channels: str | Iterable[str]) -> list[str]: + out = [channels] if isinstance(channels, str) else list(channels) + if not out: + raise ValueError("give at least one channel") + if len(out) > MAX_TRIGGER_CHANNELS: + raise ValueError(f"trigger takes at most {MAX_TRIGGER_CHANNELS} channels") + for ch in out: + _signing.check_channel(ch) + return out + + +def _data(data: Any) -> str: + return data if isinstance(data, str) else _signing.compact(data) + + +def _parse(response: httpx.Response) -> Any: + if response.status_code >= 400: + raise error_for(response.status_code, response.text) + if not response.content: + return {} + try: + return response.json() + except ValueError: + return {} + + +class _Base: + def __init__( + self, + app_id: str, + key: str, + secret: str, + host: str = DEFAULT_HOST, + port: int | None = None, + use_tls: bool = True, + timeout: float = 10.0, + push_instance_id: str | None = None, + push_secret_key: str | None = None, + push_endpoint: str | None = None, + ) -> None: + missing = [n for n, v in (("app_id", app_id), ("key", key), ("secret", secret)) if not v] + if missing: + raise ConfigError(f"missing {', '.join(missing)} (set HIBIRA_APP_ID, HIBIRA_APP_KEY, HIBIRA_APP_SECRET)") + self.app_id = str(app_id) + self.key = key + self.secret = secret + self.host = host or DEFAULT_HOST + self.use_tls = use_tls + self.port = port + self.timeout = timeout + scheme = "https" if use_tls else "http" + default_port = 443 if use_tls else 80 + self.base_url = f"{scheme}://{self.host}" + (f":{port}" if port and port != default_port else "") + self.push_instance_id = push_instance_id + self.push_secret_key = push_secret_key + self.push_endpoint = (push_endpoint or self.base_url).rstrip("/") + + @classmethod + def from_env(cls, **overrides: Any) -> Any: + """Settings from HIBIRA_* environment variables; keyword arguments win.""" + env = os.environ + port = env.get("HIBIRA_PORT") + values: dict[str, Any] = { + "app_id": env.get("HIBIRA_APP_ID", ""), + "key": env.get("HIBIRA_APP_KEY", ""), + "secret": env.get("HIBIRA_APP_SECRET", ""), + "host": env.get("HIBIRA_HOST") or DEFAULT_HOST, + "port": int(port) if port else None, + "use_tls": _env_bool(env.get("HIBIRA_USE_TLS")), + "push_instance_id": env.get("HIBIRA_PUSH_INSTANCE_ID") or None, + "push_secret_key": env.get("HIBIRA_PUSH_SECRET_KEY") or None, + "push_endpoint": env.get("HIBIRA_PUSH_ENDPOINT") or None, + } + if values["use_tls"] is None: + values["use_tls"] = True + values.update(overrides) + return cls(**values) + + def __repr__(self) -> str: + return f"{type(self).__name__}(app_id={self.app_id!r}, host={self.host!r})" + + # Signing (no network) + + def authenticate_channel( + self, channel: str, socket_id: str, user_data: Mapping[str, Any] | None = None + ) -> dict[str, str]: + """Authorization for a private- or presence- channel subscription. + + Return it as JSON from your auth endpoint. Presence channels need + user_data with a user_id (and optional user_info). + """ + return _signing.authenticate_channel(self.key, self.secret, channel, socket_id, user_data) + + def authenticate_user(self, socket_id: str, user_data: Mapping[str, Any]) -> dict[str, str]: + """Response for the client's user sign-in (pusher:signin); user_data needs an id.""" + return _signing.authenticate_user(self.key, self.secret, socket_id, user_data) + + def verify_webhook(self, key: str | None, signature: str | None, body: bytes | str) -> Any: + """Check a webhook's X-Pusher-Key and X-Pusher-Signature headers and return its parsed body. + + Pass the raw request body. Raises InvalidWebhook when they do not match. + """ + return _signing.verify_webhook(self.key, self.secret, key, signature, body) + + def push_token(self, user_id: str, ttl: int = 86400) -> dict[str, str]: + """A Hibira Push user token, for setUserId in the web, Android and iOS SDKs.""" + instance, secret = self._push_settings() + if not user_id or len(str(user_id)) > 164: + raise ValueError("user_id must be 1 to 164 characters") + return {"token": _signing.push_token(instance, secret, str(user_id), ttl)} + + # Request builders + + def _signed(self, method: str, path: str, params: Mapping[str, Any] | None = None, body: Any = None) -> _Request: + content = _signing.compact(body).encode() if body is not None else None + full = f"/apps/{quote(self.app_id, safe='')}{path}" + q = _signing.sign_query(self.key, self.secret, method, unquote(full), params, content) + headers = {"User-Agent": _USER_AGENT} + if content is not None: + headers["Content-Type"] = "application/json" + return _Request(method, self.base_url + full, q, content, headers) + + def _trigger_req(self, channels: str | Iterable[str], event: str, data: Any, socket_id: str | None) -> _Request: + if not event or len(event) > 200: + raise ValueError("event name must be 1 to 200 characters") + body: dict[str, Any] = {"name": event, "data": _data(data), "channels": _channels(channels)} + if socket_id: + _signing.check_socket_id(socket_id) + body["socket_id"] = socket_id + return self._signed("POST", "/events", body=body) + + def _batch_req(self, events: Sequence[Mapping[str, Any]]) -> _Request: + if not events: + raise ValueError("give at least one event") + batch = [] + for e in events: + _signing.check_channel(e["channel"]) + item = {"channel": e["channel"], "name": e["name"], "data": _data(e.get("data", ""))} + if e.get("socket_id"): + _signing.check_socket_id(e["socket_id"]) + item["socket_id"] = e["socket_id"] + if e.get("info"): + item["info"] = e["info"] + batch.append(item) + return self._signed("POST", "/batch_events", body={"batch": batch}) + + def _channels_req(self, prefix: str | None, info: str | Iterable[str] | None) -> _Request: + return self._signed("GET", "/channels", {"filter_by_prefix": prefix, "info": _info(info)}) + + def _channel_req(self, channel: str, info: str | Iterable[str] | None) -> _Request: + _signing.check_channel(channel) + return self._signed("GET", f"/channels/{channel}", {"info": _info(info)}) + + def _users_req(self, channel: str) -> _Request: + _signing.check_channel(channel) + if not channel.startswith("presence-"): + raise ValueError("users() works on presence- channels") + return self._signed("GET", f"/channels/{channel}/users") + + def _terminate_req(self, user_id: str) -> _Request: + if not user_id: + raise ValueError("user_id is required") + return self._signed("POST", f"/users/{quote(str(user_id), safe='')}/terminate_connections", body={}) + + def _approval_req(self, **fields: Any) -> _Request: + body = {k: v for k, v in fields.items() if v is not None} + if not body.get("title"): + raise ValueError("title is required") + return self._signed("POST", "/approvals", body=body) + + def _get_approval_req(self, approval_id: str, wait: int) -> _Request: + if not approval_id: + raise ValueError("approval_id is required") + if wait < 0 or wait > MAX_APPROVAL_WAIT: + raise ValueError(f"wait must be 0 to {MAX_APPROVAL_WAIT} seconds") + params = {"wait": wait} if wait else None + return self._signed("GET", f"/approvals/{quote(approval_id, safe='')}", params) + + def _push_settings(self) -> tuple[str, str]: + if not self.push_instance_id or not self.push_secret_key: + raise ConfigError( + "push needs push_instance_id and push_secret_key (HIBIRA_PUSH_INSTANCE_ID, HIBIRA_PUSH_SECRET_KEY)" + ) + return self.push_instance_id, self.push_secret_key + + def _push_req(self, method: str, path: str, body: Any = None) -> _Request: + instance, secret = self._push_settings() + headers = {"Authorization": f"Bearer {secret}", "User-Agent": _USER_AGENT} + content = None + if body is not None: + content = _signing.compact(body).encode() + headers["Content-Type"] = "application/json" + url = f"{self.push_endpoint}{path.format(instance=quote(instance, safe=''))}" + return _Request(method, url, {}, content, headers) + + def _push_interests_req(self, interests: Iterable[str], body: Mapping[str, Any]) -> _Request: + names = [interests] if isinstance(interests, str) else list(interests) + if not names: + raise ValueError("give at least one interest") + return self._push_req( + "POST", "/publish_api/v1/instances/{instance}/publishes/interests", {**body, "interests": names} + ) + + def _push_users_req(self, user_ids: Iterable[str], body: Mapping[str, Any]) -> _Request: + users = [user_ids] if isinstance(user_ids, str) else [str(u) for u in user_ids] + if not users: + raise ValueError("give at least one user id") + return self._push_req("POST", "/publish_api/v1/instances/{instance}/publishes/users", {**body, "users": users}) + + def _delete_push_user_req(self, user_id: str) -> _Request: + return self._push_req("DELETE", f"/customer_api/v1/instances/{{instance}}/users/{quote(str(user_id), safe='')}") + + +def _info(info: str | Iterable[str] | None) -> str | None: + if info is None or isinstance(info, str): + return info or None + return ",".join(info) or None + + +def _approval_fields( + title: str, + details: str | None, + options: Sequence[str] | None, + channel: str | None, + expires_in_minutes: int | None, + push_interests: Sequence[str] | None, + push_users: Sequence[str] | None, + agent: str | None, +) -> dict[str, Any]: + return { + "title": title, + "details": details, + "options": list(options) if options else None, + "channel": channel, + "expires_in_minutes": expires_in_minutes, + "push_interests": list(push_interests) if push_interests else None, + "push_users": list(push_users) if push_users else None, + "agent": agent, + } + + +class Client(_Base): + """Server client for a Hibira app: publish events, authorize channels, + verify webhooks, send push notifications and ask for approvals. + + >>> hibira = Client.from_env() + >>> hibira.trigger("orders", "created", {"id": 1042}) + """ + + def __init__( + self, app_id: str, key: str, secret: str, *, http_client: httpx.Client | None = None, **kwargs: Any + ) -> None: + super().__init__(app_id, key, secret, **kwargs) + self._own_http = http_client is None + self._http = http_client or httpx.Client(timeout=self.timeout) + + def _send(self, req: _Request, timeout: float | None = None) -> Any: + r = self._http.request( + req.method, + req.url, + params=req.params or None, + content=req.content, + headers=req.headers, + timeout=timeout if timeout is not None else self.timeout, + ) + return _parse(r) + + def close(self) -> None: + if self._own_http: + self._http.close() + + def __enter__(self) -> Client: + return self + + def __exit__(self, *exc: Any) -> None: + self.close() + + def trigger(self, channels: str | Iterable[str], event: str, data: Any, socket_id: str | None = None) -> Any: + """Publish an event to one channel or up to 100. Data that is not a str is sent as JSON. + + socket_id leaves out that connection (the sender's own tab). + """ + return self._send(self._trigger_req(channels, event, data, socket_id)) + + def trigger_batch(self, events: Sequence[Mapping[str, Any]]) -> Any: + """Publish several events in one request (10 at most by default). + + events: [{"channel", "name", "data", "socket_id"?}] + """ + return self._send(self._batch_req(events)) + + def channels(self, prefix: str | None = None, info: str | Iterable[str] | None = None) -> Any: + return self._send(self._channels_req(prefix, info)) + + def channel_info(self, channel: str, info: str | Iterable[str] | None = None) -> Any: + return self._send(self._channel_req(channel, info)) + + def users(self, channel: str) -> Any: + return self._send(self._users_req(channel)) + + def terminate_user_connections(self, user_id: str) -> Any: + return self._send(self._terminate_req(user_id)) + + def push_to_interests(self, interests: str | Iterable[str], body: Mapping[str, Any]) -> Any: + """Send a push notification to devices subscribed to any of interests. + + body is a Beams publish body: {"web": {...}, "fcm": {...}, "apns": {...}}. + """ + return self._send(self._push_interests_req(interests, body)) + + def push_to_users(self, user_ids: str | Iterable[str], body: Mapping[str, Any]) -> Any: + """Send a push notification to the devices of up to 1000 users.""" + return self._send(self._push_users_req(user_ids, body)) + + def delete_push_user(self, user_id: str) -> Any: + return self._send(self._delete_push_user_req(user_id)) + + def request_approval( + self, + title: str, + details: str | None = None, + options: Sequence[str] | None = None, + channel: str | None = None, + expires_in_minutes: int | None = None, + push_interests: Sequence[str] | None = None, + push_users: Sequence[str] | None = None, + agent: str | None = None, + ) -> Any: + """Ask a person for a decision; share the returned "url" with them.""" + fields = _approval_fields( + title, details, options, channel, expires_in_minutes, push_interests, push_users, agent + ) + return self._send(self._approval_req(**fields)) + + def get_approval(self, approval_id: str, wait: int = 0) -> Any: + """An approval's status; wait (up to 25 s) holds the request until it is answered.""" + return self._send(self._get_approval_req(approval_id, wait), timeout=self.timeout + wait) + + def wait_for_approval(self, approval_id: str, timeout: float = 3600) -> Any: + """Poll until the approval is no longer pending (or timeout seconds pass) and return it.""" + deadline = time.monotonic() + timeout + while True: + left = deadline - time.monotonic() + out = self.get_approval(approval_id, wait=max(0, min(MAX_APPROVAL_WAIT, int(left)))) + if out.get("status") != "pending" or time.monotonic() >= deadline: + return out + + +class AsyncClient(_Base): + """asyncio version of Client: the same methods, HTTP ones are awaitable. + + authenticate_channel, authenticate_user, verify_webhook and push_token + need no network and stay synchronous. + """ + + def __init__( + self, app_id: str, key: str, secret: str, *, http_client: httpx.AsyncClient | None = None, **kwargs: Any + ) -> None: + super().__init__(app_id, key, secret, **kwargs) + self._own_http = http_client is None + self._http = http_client or httpx.AsyncClient(timeout=self.timeout) + + async def _send(self, req: _Request, timeout: float | None = None) -> Any: + r = await self._http.request( + req.method, + req.url, + params=req.params or None, + content=req.content, + headers=req.headers, + timeout=timeout if timeout is not None else self.timeout, + ) + return _parse(r) + + async def aclose(self) -> None: + if self._own_http: + await self._http.aclose() + + async def __aenter__(self) -> AsyncClient: + return self + + async def __aexit__(self, *exc: Any) -> None: + await self.aclose() + + async def trigger(self, channels: str | Iterable[str], event: str, data: Any, socket_id: str | None = None) -> Any: + return await self._send(self._trigger_req(channels, event, data, socket_id)) + + async def trigger_batch(self, events: Sequence[Mapping[str, Any]]) -> Any: + return await self._send(self._batch_req(events)) + + async def channels(self, prefix: str | None = None, info: str | Iterable[str] | None = None) -> Any: + return await self._send(self._channels_req(prefix, info)) + + async def channel_info(self, channel: str, info: str | Iterable[str] | None = None) -> Any: + return await self._send(self._channel_req(channel, info)) + + async def users(self, channel: str) -> Any: + return await self._send(self._users_req(channel)) + + async def terminate_user_connections(self, user_id: str) -> Any: + return await self._send(self._terminate_req(user_id)) + + async def push_to_interests(self, interests: str | Iterable[str], body: Mapping[str, Any]) -> Any: + return await self._send(self._push_interests_req(interests, body)) + + async def push_to_users(self, user_ids: str | Iterable[str], body: Mapping[str, Any]) -> Any: + return await self._send(self._push_users_req(user_ids, body)) + + async def delete_push_user(self, user_id: str) -> Any: + return await self._send(self._delete_push_user_req(user_id)) + + async def request_approval( + self, + title: str, + details: str | None = None, + options: Sequence[str] | None = None, + channel: str | None = None, + expires_in_minutes: int | None = None, + push_interests: Sequence[str] | None = None, + push_users: Sequence[str] | None = None, + agent: str | None = None, + ) -> Any: + fields = _approval_fields( + title, details, options, channel, expires_in_minutes, push_interests, push_users, agent + ) + return await self._send(self._approval_req(**fields)) + + async def get_approval(self, approval_id: str, wait: int = 0) -> Any: + return await self._send(self._get_approval_req(approval_id, wait), timeout=self.timeout + wait) + + async def wait_for_approval(self, approval_id: str, timeout: float = 3600) -> Any: + deadline = time.monotonic() + timeout + while True: + left = deadline - time.monotonic() + out = await self.get_approval(approval_id, wait=max(0, min(MAX_APPROVAL_WAIT, int(left)))) + if out.get("status") != "pending" or time.monotonic() >= deadline: + return out diff --git a/src/hibira/_errors.py b/src/hibira/_errors.py new file mode 100644 index 0000000..2725c0c --- /dev/null +++ b/src/hibira/_errors.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import json +from typing import Any + + +class HibiraError(Exception): + """Base class of every error this package raises.""" + + +class ConfigError(HibiraError): + """A setting is missing or invalid (app id, key, secret, push instance...).""" + + +class InvalidWebhook(HibiraError): + """A webhook's key or signature does not match this app.""" + + +class HTTPError(HibiraError): + """The Hibira HTTP API answered with an error status.""" + + def __init__(self, status: int, message: str, body: Any = None) -> None: + super().__init__(f"{status}: {message}") + self.status = status + self.message = message + self.body = body + + +class AuthenticationError(HTTPError): + """401 or 403: bad key, secret or signature, or a suspended account.""" + + +class NotFound(HTTPError): + """404: unknown app, channel, approval or push instance.""" + + +class PayloadTooLarge(HTTPError): + """413: the event is over the plan's message size (4 KB on Free).""" + + +class RateLimited(HTTPError): + """429: too many requests or messages for the plan.""" + + +def error_for(status: int, text: str) -> HTTPError: + body: Any = text + message = text.strip() or "request failed" + try: + body = json.loads(text) + except ValueError: + pass + else: + if isinstance(body, dict): + for k in ("error", "message", "description"): + if isinstance(body.get(k), str): + message = body[k] + break + cls: type[HTTPError] = HTTPError + if status in (401, 403): + cls = AuthenticationError + elif status == 404: + cls = NotFound + elif status == 413: + cls = PayloadTooLarge + elif status == 429: + cls = RateLimited + return cls(status, message, body) diff --git a/src/hibira/_signing.py b/src/hibira/_signing.py new file mode 100644 index 0000000..8083755 --- /dev/null +++ b/src/hibira/_signing.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import re +import time +from typing import Any, Mapping + +from ._errors import InvalidWebhook + +_SOCKET_ID = re.compile(r"\A\d+\.\d+\Z") +_CHANNEL = re.compile(r"\A[-a-zA-Z0-9_=@,.;]+\Z") + + +def compact(value: Any) -> str: + return json.dumps(value, separators=(",", ":"), ensure_ascii=False) + + +def hmac_hex(secret: str, message: str | bytes) -> str: + if isinstance(message, str): + message = message.encode() + return hmac.new(secret.encode(), message, hashlib.sha256).hexdigest() + + +def check_socket_id(socket_id: str) -> None: + if not isinstance(socket_id, str) or not _SOCKET_ID.match(socket_id): + raise ValueError(f"invalid socket_id: {socket_id!r}") + + +def check_channel(channel: str) -> None: + if not isinstance(channel, str) or not _CHANNEL.match(channel) or len(channel) > 200: + raise ValueError(f"invalid channel name: {channel!r}") + + +def sign_query( + key: str, + secret: str, + method: str, + path: str, + params: Mapping[str, Any] | None = None, + body: bytes | None = None, + now: float | None = None, +) -> dict[str, str]: + """Query parameters of a signed HTTP API request (Pusher auth_version 1.0).""" + q = {k.lower(): str(v) for k, v in (params or {}).items() if v is not None} + q["auth_key"] = key + q["auth_timestamp"] = str(int(now if now is not None else time.time())) + q["auth_version"] = "1.0" + if body: + q["body_md5"] = hashlib.md5(body).hexdigest() + string = "&".join(f"{k}={q[k]}" for k in sorted(q)) + q["auth_signature"] = hmac_hex(secret, f"{method.upper()}\n{path}\n{string}") + return q + + +def authenticate_channel( + key: str, secret: str, channel: str, socket_id: str, user_data: Mapping[str, Any] | None = None +) -> dict[str, str]: + check_channel(channel) + check_socket_id(socket_id) + if channel.startswith("private-encrypted-"): + raise ValueError("encrypted channels are not supported") + if channel.startswith("presence-"): + if not user_data or user_data.get("user_id") in (None, ""): + raise ValueError("presence channels need user_data with a user_id") + channel_data = compact(dict(user_data)) + sig = hmac_hex(secret, f"{socket_id}:{channel}:{channel_data}") + return {"auth": f"{key}:{sig}", "channel_data": channel_data} + if not channel.startswith("private-"): + raise ValueError("only private- and presence- channels need authorization") + return {"auth": f"{key}:{hmac_hex(secret, f'{socket_id}:{channel}')}"} + + +def authenticate_user(key: str, secret: str, socket_id: str, user_data: Mapping[str, Any]) -> dict[str, str]: + check_socket_id(socket_id) + if not user_data or user_data.get("id") in (None, ""): + raise ValueError("user_data needs an id") + encoded = compact(dict(user_data)) + return {"auth": f"{key}:{hmac_hex(secret, f'{socket_id}::user::{encoded}')}", "user_data": encoded} + + +def verify_webhook(key: str, secret: str, got_key: str | None, signature: str | None, body: bytes | str) -> Any: + raw = body.encode() if isinstance(body, str) else bytes(body) + if not got_key or not hmac.compare_digest(got_key, key): + raise InvalidWebhook("webhook key does not match this app") + if not signature or not hmac.compare_digest(signature.lower(), hmac_hex(secret, raw)): + raise InvalidWebhook("webhook signature does not match") + try: + return json.loads(raw) + except ValueError as err: + raise InvalidWebhook("webhook body is not JSON") from err + + +def _b64(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode() + + +def push_token(instance_id: str, secret_key: str, user_id: str, ttl: int, now: float | None = None) -> str: + t = int(now if now is not None else time.time()) + header = _b64(compact({"alg": "HS256", "typ": "JWT"}).encode()) + claims = {"sub": user_id, "iss": f"https://{instance_id}.pushnotifications.pusher.com", "iat": t, "exp": t + ttl} + payload = _b64(compact(claims).encode()) + sig = hmac.new(secret_key.encode(), f"{header}.{payload}".encode(), hashlib.sha256).digest() + return f"{header}.{payload}.{_b64(sig)}" diff --git a/src/hibira/_version.py b/src/hibira/_version.py new file mode 100644 index 0000000..3dc1f76 --- /dev/null +++ b/src/hibira/_version.py @@ -0,0 +1 @@ +__version__ = "0.1.0" diff --git a/src/hibira/fastapi.py b/src/hibira/fastapi.py new file mode 100644 index 0000000..97e04cf --- /dev/null +++ b/src/hibira/fastapi.py @@ -0,0 +1,128 @@ +"""FastAPI (and Starlette) helpers: the channel and user auth endpoints +pusher-js calls, and a dependency that verifies webhooks. + + from hibira import AsyncClient + from hibira.fastapi import auth_router, webhook + + hibira = AsyncClient.from_env() + + async def authorize(request, channel): + user = await current_user(request) + if channel == f"private-user-{user.id}": + return True + if channel.startswith("presence-room-"): + return {"user_id": user.id, "user_info": {"name": user.name}} + return False + + app.include_router(auth_router(hibira, authorize)) + + @app.post("/hibira/webhook") + async def on_webhook(payload: dict = Depends(webhook(hibira))): + ... +""" + +from __future__ import annotations + +import inspect +import json +from typing import Any, Awaitable, Callable, Mapping, Optional, Union +from urllib.parse import parse_qs + +try: + from fastapi import APIRouter, HTTPException, Request + from fastapi.responses import JSONResponse +except ImportError as err: # pragma: no cover + raise ImportError('hibira.fastapi needs FastAPI: pip install "hibira[fastapi]"') from err + +from ._client import _Base +from ._errors import InvalidWebhook + +Authorize = Callable[[Request, str], Union[Any, Awaitable[Any]]] +Identify = Callable[[Request], Union[Optional[Mapping[str, Any]], Awaitable[Optional[Mapping[str, Any]]]]] + + +async def _maybe(value: Any) -> Any: + return await value if inspect.isawaitable(value) else value + + +async def _form(request: Request) -> dict[str, str]: + raw = await request.body() + if request.headers.get("content-type", "").startswith("application/json"): + try: + data = json.loads(raw or b"{}") + except ValueError: + data = {} + return {k: str(v) for k, v in data.items()} if isinstance(data, dict) else {} + return {k: v[0] for k, v in parse_qs(raw.decode("utf-8", "replace")).items()} + + +def auth_router( + client: _Base, + authorize: Authorize, + *, + path: str = "/hibira/auth", + user: Optional[Identify] = None, + user_path: str = "/hibira/user-auth", +) -> APIRouter: + """Router with the endpoint pusher-js calls to authorize private- and presence- channels. + + authorize(request, channel) (sync or async) returns False/None to refuse, + True to allow a private channel, or the member's user_data + ({"user_id", "user_info"?}) for a presence channel. Point the client's + channelAuthorization.endpoint at path. + + With user (request -> {"id", ...} or None), the router also answers user + sign-in at user_path (userAuthentication.endpoint). + """ + router = APIRouter() + + @router.post(path, include_in_schema=False) + async def channel_auth(request: Request) -> JSONResponse: + form = await _form(request) + socket_id, channel = form.get("socket_id", ""), form.get("channel_name", "") + if not socket_id or not channel: + raise HTTPException(400, "socket_id and channel_name are required") + verdict = await _maybe(authorize(request, channel)) + if not verdict: + raise HTTPException(403, "forbidden") + user_data = verdict if isinstance(verdict, Mapping) else None + if channel.startswith("presence-") and user_data is None: + raise HTTPException(500, "authorize must return user_data with a user_id for presence channels") + try: + return JSONResponse(client.authenticate_channel(channel, socket_id, user_data)) + except ValueError as err: + raise HTTPException(400, str(err)) from err + + if user is not None: + + @router.post(user_path, include_in_schema=False) + async def user_auth(request: Request) -> JSONResponse: + form = await _form(request) + socket_id = form.get("socket_id", "") + if not socket_id: + raise HTTPException(400, "socket_id is required") + data = await _maybe(user(request)) + if not data: + raise HTTPException(403, "forbidden") + try: + return JSONResponse(client.authenticate_user(socket_id, data)) + except ValueError as err: + raise HTTPException(400, str(err)) from err + + return router + + +def webhook(client: _Base) -> Callable[[Request], Awaitable[Any]]: + """Dependency that verifies a Hibira webhook and returns its parsed body + ({"time_ms", "events": [...]}); a bad signature answers 401.""" + + async def verified(request: Request) -> Any: + body = await request.body() + try: + return client.verify_webhook( + request.headers.get("x-pusher-key"), request.headers.get("x-pusher-signature"), body + ) + except InvalidWebhook as err: + raise HTTPException(401, str(err)) from err + + return verified diff --git a/src/hibira/py.typed b/src/hibira/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/hibira/stream.py b/src/hibira/stream.py new file mode 100644 index 0000000..c284ce4 --- /dev/null +++ b/src/hibira/stream.py @@ -0,0 +1,419 @@ +"""Server side of the Hibira stream protocol: send an LLM's output (or any +stream of text) to a channel as it is produced. The useStream() hook in +@hibira-io/react and composable in @hibira-io/vue put it back together. + +Same wire format as streamToChannel in those packages: event "ai-stream", +payloads {id, seq, delta, meta (first event only)}, then a final +{id, seq, done: true, text?, error?}. Tokens are batched (one event per +interval rather than one per token), each event stays under max_bytes (Free +plans allow 4 KB per message) and events are sent one after another, in order. +""" + +from __future__ import annotations + +import asyncio +import codecs +import inspect +import json +import threading +import uuid +from dataclasses import dataclass +from typing import Any, AsyncIterable, Callable, Iterable, Optional, Union + +STREAM_EVENT = "ai-stream" + +_MISSING = object() + + +@dataclass(frozen=True) +class StreamResult: + id: str + text: str + events: int + + +def _get(obj: Any, name: str) -> Any: + if isinstance(obj, dict): + return obj.get(name) + return getattr(obj, name, None) + + +def _content_text(content: Any) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + out = [] + for block in content: + if isinstance(block, str): + out.append(block) + elif _get(block, "type") in ("text", "text_delta", "output_text") and isinstance(_get(block, "text"), str): + out.append(_get(block, "text")) + return "".join(out) + return "" + + +def text_of(chunk: Any, decoder: Optional[codecs.IncrementalDecoder] = None) -> str: + """Text of one chunk from the common LLM SDK streams, or "" when it has none. + + Handles str and bytes, Anthropic (content_block_delta events), OpenAI chat + completions (choices[0].delta.content) and responses + (response.output_text.delta), the OpenAI Agents SDK (raw_response_event), + LangChain / LangGraph message chunks (stream_mode="messages", also as + (mode, chunk) tuples), Gemini (chunk.text) and dicts of the same shapes. + """ + if chunk is None: + return "" + if isinstance(chunk, str): + return chunk + if isinstance(chunk, (bytes, bytearray, memoryview)): + if decoder is None: + return bytes(chunk).decode("utf-8", errors="replace") + return decoder.decode(bytes(chunk)) + if isinstance(chunk, tuple) and chunk: + if len(chunk) >= 2 and isinstance(chunk[0], str): + return text_of(chunk[-1], decoder) + return text_of(chunk[0], decoder) + kind = _get(chunk, "type") + if kind == "content_block_delta": + delta = _get(chunk, "delta") + return (_get(delta, "text") or "") if _get(delta, "type") == "text_delta" else "" + if kind == "response.output_text.delta": + return _get(chunk, "delta") or "" + if kind == "raw_response_event": + return text_of(_get(chunk, "data"), decoder) + if kind == "text-delta": + for name in ("text", "textDelta", "delta"): + v = _get(chunk, name) + if isinstance(v, str): + return v + return "" + if kind in ("AIMessageChunk", "ai"): + return _content_text(_get(chunk, "content")) + choices = _get(chunk, "choices") + if isinstance(choices, list): + if not choices: + return "" + return _get(_get(choices[0], "delta"), "content") or "" + if kind is None or isinstance(kind, str): + try: + text = _get(chunk, "text") + except Exception: # noqa: BLE001 + return "" + if isinstance(text, str): + return text + return "" + + +def _size(s: str) -> int: + return len(json.dumps(s, ensure_ascii=False).encode()) - 2 + + +def _bytes(s: str) -> int: + return len(s.encode()) + + +def split(text: str, budget: int) -> list[str]: + """Split text so each piece's JSON-escaped UTF-8 size fits budget, never inside a character.""" + out: list[str] = [] + piece: list[str] = [] + size = 0 + for ch in text: + n = _size(ch) + if size + n > budget and piece: + out.append("".join(piece)) + piece, size = [], 0 + piece.append(ch) + size += n + if piece: + out.append("".join(piece)) + return out + + +def _payload_bytes(payload: dict[str, Any]) -> int: + return len(json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode()) + + +class _Encoder: + def __init__(self, id: str, meta: Any, max_bytes: int) -> None: + self.id = id + self.meta = meta + self.max_bytes = max_bytes + envelope = {"id": id, "seq": 1_000_000_000, "delta": ""} + if meta is not _MISSING: + envelope["meta"] = meta + self.budget = max_bytes - _payload_bytes(envelope) + if self.budget < 16: + raise ValueError("stream_to_channel: max_bytes is too small for the id and meta") + self.seq = 0 + + def deltas(self, buffer: str) -> list[dict[str, Any]]: + out = [] + for delta in split(buffer, self.budget): + payload: dict[str, Any] = {"id": self.id, "seq": self.seq, "delta": delta} + if self.seq == 0 and self.meta is not _MISSING: + payload["meta"] = self.meta + self.seq += 1 + out.append(payload) + return out + + def done(self, text: str, error: Optional[str]) -> dict[str, Any]: + payload: dict[str, Any] = {"id": self.id, "seq": self.seq, "done": True} + if self.seq == 0 and self.meta is not _MISSING: + payload["meta"] = self.meta + if error is not None: + payload["error"] = error + if _payload_bytes({**payload, "text": text}) <= self.max_bytes: + payload["text"] = text + self.seq += 1 + return payload + + +def _error_text(err: BaseException) -> str: + if isinstance(err, (asyncio.CancelledError, KeyboardInterrupt)): + return "aborted" + return str(err) or type(err).__name__ + + +def stream_to_channel( + client: Any, + channel: str, + source: Iterable[Any], + *, + id: Optional[str] = None, + event: str = STREAM_EVENT, + interval: float = 0.1, + max_bytes: int = 3500, + meta: Any = _MISSING, + pick: Optional[Callable[[Any], Optional[str]]] = None, + socket_id: Optional[str] = None, + stop: Optional[threading.Event] = None, +) -> StreamResult: + """Send a stream of text to a channel as events useStream() puts back together. + + client is a hibira.Client (anything with trigger(channel, event, data, + socket_id=None) works). source yields text chunks or LLM SDK chunks (see + text_of; pass pick for other shapes). meta is sent with the first event. + Setting stop ends the stream with error "aborted". Returns after the last + event is sent; if source raises, the final event carries the error and it + is raised again. + """ + enc = _Encoder(id or str(uuid.uuid4()), meta, max_bytes) + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") + lock = threading.Lock() + send_lock = threading.Lock() + state: dict[str, Any] = {"buffer": "", "timer": None, "send_error": None} + + def send(payloads: list[dict[str, Any]]) -> None: + with send_lock: + for p in payloads: + if state["send_error"] is not None: + return + try: + client.trigger(channel, event, p, socket_id=socket_id) + except Exception as err: # noqa: BLE001 + state["send_error"] = err + + def flush() -> None: + with lock: + timer = state["timer"] + if timer is not None: + timer.cancel() + state["timer"] = None + buffer, state["buffer"] = state["buffer"], "" + if not buffer: + return + # Taking send_lock before releasing lock keeps events in seq order. + send_lock.acquire() + try: + payloads = enc.deltas(buffer) + except BaseException: + send_lock.release() + raise + try: + for p in payloads: + if state["send_error"] is not None: + break + try: + client.trigger(channel, event, p, socket_id=socket_id) + except Exception as err: # noqa: BLE001 + state["send_error"] = err + finally: + send_lock.release() + + text_parts: list[str] = [] + error: Optional[BaseException] = None + try: + for chunk in source: + if stop is not None and stop.is_set(): + raise _Aborted() + if state["send_error"] is not None: + raise state["send_error"] + delta = (pick(chunk) or "") if pick else text_of(chunk, decoder) + if not delta: + continue + text_parts.append(delta) + with lock: + state["buffer"] += delta + full = _bytes(state["buffer"]) >= enc.budget + if not full and state["timer"] is None: + t = threading.Timer(interval, flush) + t.daemon = True + state["timer"] = t + t.start() + if full: + flush() + tail = decoder.decode(b"", final=True) + if tail: + text_parts.append(tail) + with lock: + state["buffer"] += tail + if stop is not None and stop.is_set(): + raise _Aborted() + except BaseException as err: # noqa: BLE001 + error = err + flush() + + text = "".join(text_parts) + message = None + if error is not None: + message = "aborted" if isinstance(error, _Aborted) else _error_text(error) + send([enc.done(text, message)]) + + if error is not None and not isinstance(error, _Aborted): + raise error + if state["send_error"] is not None: + raise state["send_error"] + return StreamResult(enc.id, text, enc.seq) + + +class _Aborted(Exception): + pass + + +async def _aiter(source: Union[AsyncIterable[Any], Iterable[Any]]) -> Any: + if hasattr(source, "__aiter__"): + async for item in source: + yield item + else: + for item in source: + yield item + + +async def astream_to_channel( + client: Any, + channel: str, + source: Union[AsyncIterable[Any], Iterable[Any]], + *, + id: Optional[str] = None, + event: str = STREAM_EVENT, + interval: float = 0.1, + max_bytes: int = 3500, + meta: Any = _MISSING, + pick: Optional[Callable[[Any], Optional[str]]] = None, + socket_id: Optional[str] = None, +) -> StreamResult: + """asyncio version of stream_to_channel, for a hibira.AsyncClient (or a + sync Client, whose trigger then runs in a thread) and an async iterable + such as an OpenAI / Anthropic async stream or graph.astream(...). + + Cancelling the task sends the final event with error "aborted" and + re-raises CancelledError. + """ + enc = _Encoder(id or str(uuid.uuid4()), meta, max_bytes) + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") + send_lock = asyncio.Lock() + buffer = "" + timer: Optional[asyncio.TimerHandle] = None + pending: set[asyncio.Task[None]] = set() + send_error: Optional[BaseException] = None + loop = asyncio.get_running_loop() + + async def trigger(payload: dict[str, Any]) -> None: + result = client.trigger(channel, event, payload, socket_id=socket_id) + if inspect.isawaitable(result): + await result + + async def send(payloads: list[dict[str, Any]]) -> None: + nonlocal send_error + async with send_lock: + for p in payloads: + if send_error is not None: + return + try: + await trigger(p) + except Exception as err: # noqa: BLE001 + send_error = err + + def flush() -> None: + nonlocal buffer, timer + if timer is not None: + timer.cancel() + timer = None + if not buffer: + return + payloads = enc.deltas(buffer) + buffer = "" + task = loop.create_task(send(payloads)) + pending.add(task) + task.add_done_callback(pending.discard) + + if not _is_async_client(client): + sync_client = client + + class _Threaded: + def trigger(self, *a: Any, **kw: Any) -> Any: + return asyncio.to_thread(sync_client.trigger, *a, **kw) + + client = _Threaded() + + text_parts: list[str] = [] + error: Optional[BaseException] = None + try: + async for chunk in _aiter(source): + if send_error is not None: + raise send_error + delta = (pick(chunk) or "") if pick else text_of(chunk, decoder) + if not delta: + continue + text_parts.append(delta) + buffer += delta + if _bytes(buffer) >= enc.budget: + flush() + elif timer is None: + timer = loop.call_later(interval, flush) + tail = decoder.decode(b"", final=True) + if tail: + text_parts.append(tail) + buffer += tail + except BaseException as err: # noqa: BLE001 + error = err + flush() + + text = "".join(text_parts) + message = _error_text(error) if error is not None else None + final = enc.done(text, message) + + async def finish() -> None: + # Tasks run in creation order, but each waits on send_lock (FIFO), so + # the final event goes last. + await send([final]) + if pending: + await asyncio.gather(*list(pending), return_exceptions=True) + + finisher = asyncio.ensure_future(finish()) + try: + await asyncio.shield(finisher) + except asyncio.CancelledError: + await finisher + raise + + if error is not None: + raise error + if send_error is not None: + raise send_error + return StreamResult(enc.id, text, enc.seq) + + +def _is_async_client(client: Any) -> bool: + trigger = getattr(client, "trigger", None) + return inspect.iscoroutinefunction(trigger) diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..aa2af35 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,184 @@ +import asyncio +import json +from urllib.parse import parse_qs, urlsplit + +import httpx +import pytest + +import hibira +from hibira import _signing + + +def recorder(status=200, body=b"{}"): + seen = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(status, content=body, headers={"content-type": "application/json"}) + + return seen, handler + + +def make(handler, **kw): + return hibira.Client( + "app1", "key1", "secret1", http_client=httpx.Client(transport=httpx.MockTransport(handler)), **kw + ) + + +def check_signature(req: httpx.Request): + q = {k: v[0] for k, v in parse_qs(urlsplit(str(req.url)).query).items()} + sig = q.pop("auth_signature") + signed = _signing.sign_query( + "key1", + "secret1", + req.method, + req.url.path, + {k: v for k, v in q.items() if not k.startswith("auth_") and k != "body_md5"}, + req.content or None, + now=int(q["auth_timestamp"]), + ) + assert signed["auth_signature"] == sig + return q + + +def test_trigger(): + seen, handler = recorder() + make(handler).trigger(["a", "b"], "created", {"id": 1}, socket_id="1.2") + req = seen[0] + assert req.method == "POST" + assert str(req.url).startswith("https://ws.hibira.io/apps/app1/events?") + assert json.loads(req.content) == { + "name": "created", + "data": '{"id":1}', + "channels": ["a", "b"], + "socket_id": "1.2", + } + check_signature(req) + + +def test_trigger_string_data_and_validation(): + seen, handler = recorder() + c = make(handler) + c.trigger("a", "e", "raw") + assert json.loads(seen[0].content)["data"] == "raw" + with pytest.raises(ValueError): + c.trigger([], "e", 1) + with pytest.raises(ValueError): + c.trigger([f"c{i}" for i in range(101)], "e", 1) + with pytest.raises(ValueError): + c.trigger("a", "e", 1, socket_id="bad") + + +def test_batch_and_queries(): + seen, handler = recorder(body=b'{"channels":{}}') + c = make(handler, host="localhost", port=8080, use_tls=False) + c.trigger_batch([{"channel": "a", "name": "e", "data": {"x": 1}}]) + assert json.loads(seen[0].content) == {"batch": [{"channel": "a", "name": "e", "data": '{"x":1}'}]} + assert c.channels(prefix="presence-", info=["user_count"]) == {"channels": {}} + q = check_signature(seen[1]) + assert q["filter_by_prefix"] == "presence-" and q["info"] == "user_count" + assert str(seen[1].url).startswith("http://localhost:8080/apps/app1/channels?") + c.channel_info("presence-x", info="user_count,subscription_count") + assert seen[2].url.path == "/apps/app1/channels/presence-x" + c.users("presence-x") + assert seen[3].url.path == "/apps/app1/channels/presence-x/users" + with pytest.raises(ValueError): + c.users("private-x") + c.terminate_user_connections("u 1") + assert seen[4].url.raw_path.startswith(b"/apps/app1/users/u%201/terminate_connections") + for r in seen: + check_signature(r) + + +@pytest.mark.parametrize( + ("status", "cls"), + [ + (413, hibira.PayloadTooLarge), + (401, hibira.AuthenticationError), + (403, hibira.AuthenticationError), + (404, hibira.NotFound), + (429, hibira.RateLimited), + (500, hibira.HTTPError), + ], +) +def test_errors(status, cls): + _, handler = recorder(status, b'{"error":"nope"}') + with pytest.raises(cls) as info: + make(handler).trigger("a", "e", 1) + assert info.value.status == status + assert info.value.message == "nope" + assert isinstance(info.value, hibira.HibiraError) + + +def test_push(): + seen, handler = recorder(body=b'{"publishId":"p1"}') + c = make(handler, push_instance_id="inst", push_secret_key="sk") + assert c.push_to_interests("news", {"web": {"notification": {"title": "Hi"}}}) == {"publishId": "p1"} + assert seen[0].url == "https://ws.hibira.io/publish_api/v1/instances/inst/publishes/interests" + assert seen[0].headers["authorization"] == "Bearer sk" + assert json.loads(seen[0].content)["interests"] == ["news"] + c.push_to_users(["u1", 2], {"fcm": {}}) + assert json.loads(seen[1].content)["users"] == ["u1", "2"] + c.delete_push_user("u1") + assert seen[2].method == "DELETE" and seen[2].url.path == "/customer_api/v1/instances/inst/users/u1" + with pytest.raises(hibira.ConfigError): + make(handler).push_to_users("u1", {}) + + +def test_approvals(): + answers = iter( + [ + {"approval_id": "apr_1", "status": "pending", "url": "https://hibira.io/approve/apr_1/t"}, + {"approval_id": "apr_1", "status": "pending"}, + {"approval_id": "apr_1", "status": "approved", "choice": "Refund"}, + ] + ) + seen = [] + + def handler(req): + seen.append(req) + return httpx.Response(201 if req.method == "POST" else 200, json=next(answers)) + + c = make(handler) + out = c.request_approval("Refund #1042?", options=["Refund", "Decline"], push_interests=["ops"]) + assert out["url"].startswith("https://hibira.io/approve/") + assert json.loads(seen[0].content) == { + "title": "Refund #1042?", + "options": ["Refund", "Decline"], + "push_interests": ["ops"], + } + assert c.wait_for_approval("apr_1", timeout=60)["choice"] == "Refund" + assert seen[1].url.path == "/apps/app1/approvals/apr_1" + assert check_signature(seen[1])["wait"] == "25" + with pytest.raises(ValueError): + c.get_approval("apr_1", wait=26) + + +def test_from_env(monkeypatch): + monkeypatch.setenv("HIBIRA_APP_ID", "1") + monkeypatch.setenv("HIBIRA_APP_KEY", "k") + monkeypatch.setenv("HIBIRA_APP_SECRET", "s") + monkeypatch.setenv("HIBIRA_HOST", "ws-tyo.hibira.io") + monkeypatch.setenv("HIBIRA_USE_TLS", "false") + monkeypatch.setenv("HIBIRA_PORT", "6001") + c = hibira.Client.from_env(app_id="2") + assert c.app_id == "2" and c.base_url == "http://ws-tyo.hibira.io:6001" + assert c.push_endpoint == "http://ws-tyo.hibira.io:6001" + monkeypatch.delenv("HIBIRA_APP_SECRET") + with pytest.raises(hibira.ConfigError): + hibira.Client.from_env() + + +def test_async_client(): + seen, handler = recorder() + + async def main(): + async with hibira.AsyncClient( + "app1", "key1", "secret1", http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)) + ) as c: + await c.trigger("a", "e", {"x": 1}) + assert c.authenticate_channel("private-a", "1.1")["auth"].startswith("key1:") + + asyncio.run(main()) + assert json.loads(seen[0].content)["data"] == '{"x":1}' + check_signature(seen[0]) diff --git a/tests/test_e2e.py b/tests/test_e2e.py new file mode 100644 index 0000000..bbac7f9 --- /dev/null +++ b/tests/test_e2e.py @@ -0,0 +1,166 @@ +"""End-to-end tests against a running Hibira (pusher-server) node. + + HIBIRA_E2E=1 HIBIRA_HOST=127.0.0.1 HIBIRA_PORT=8080 HIBIRA_USE_TLS=false \ + HIBIRA_APP_ID=... HIBIRA_APP_KEY=... HIBIRA_APP_SECRET=... pytest tests/test_e2e.py + +The app needs channel history on (history: true) for the rewind test. +""" + +import asyncio +import json +import os +import uuid + +import pytest + +import hibira + +pytestmark = pytest.mark.skipif(os.environ.get("HIBIRA_E2E") != "1", reason="set HIBIRA_E2E=1 to run against a server") + +websockets = pytest.importorskip("websockets") + + +class Socket: + def __init__(self, client): + self.client = client + self.events = [] + + async def __aenter__(self): + c = self.client + scheme = "wss" if c.use_tls else "ws" + port = f":{c.port}" if c.port else "" + self.ws = await websockets.connect(f"{scheme}://{c.host}{port}/app/{c.key}?protocol=7&client=python-e2e") + msg = json.loads(await self.ws.recv()) + assert msg["event"] == "pusher:connection_established" + self.socket_id = json.loads(msg["data"])["socket_id"] + self.reader = asyncio.create_task(self._read()) + return self + + async def __aexit__(self, *exc): + self.reader.cancel() + await self.ws.close() + + async def _read(self): + async for raw in self.ws: + msg = json.loads(raw) + data = msg.get("data") + if isinstance(data, str): + try: + data = json.loads(data) + except ValueError: + pass + self.events.append((msg.get("channel"), msg["event"], data)) + + async def send(self, event, data): + await self.ws.send(json.dumps({"event": event, "data": data})) + + async def subscribe(self, channel, user_data=None): + data = {"channel": channel} + if channel.startswith(("private-", "presence-")): + data.update(self.client.authenticate_channel(channel, self.socket_id, user_data)) + await self.send("pusher:subscribe", data) + await self.wait(lambda e: e[0] == channel and e[1].endswith("subscription_succeeded")) + + async def wait(self, pred, timeout=5.0): + for _ in range(int(timeout / 0.02)): + for e in self.events: + if pred(e): + return e + await asyncio.sleep(0.02) + raise AssertionError(f"timed out; got {self.events}") + + +def rebuild(events, stream_id): + parts = sorted( + (e[2] for e in events if e[1] == "ai-stream" and e[2].get("id") == stream_id), key=lambda p: p["seq"] + ) + assert [p["seq"] for p in parts] == list(range(len(parts))) + assert parts[-1]["done"] + return "".join(p.get("delta", "") for p in parts), parts + + +@pytest.fixture +def channel(): + return f"private-py-{uuid.uuid4().hex[:8]}" + + +def test_trigger_and_query(channel): + async def main(): + async with hibira.AsyncClient.from_env() as c, Socket(c) as ws: + await ws.subscribe(channel) + await c.trigger([channel, "public-other"], "hello", {"n": 1}) + assert (await ws.wait(lambda e: e[1] == "hello"))[2] == {"n": 1} + await c.trigger(channel, "skip-me", "x", socket_id=ws.socket_id) + await c.trigger_batch([{"channel": channel, "name": "batched", "data": {"n": 2}}]) + await ws.wait(lambda e: e[1] == "batched") + assert not any(e[1] == "skip-me" for e in ws.events) + info = await c.channel_info(channel, info="subscription_count") + assert info["occupied"] is True + assert channel in (await c.channels(prefix="private-py-"))["channels"] + + asyncio.run(main()) + + +def test_presence_users(): + room = f"presence-py-{uuid.uuid4().hex[:8]}" + + async def main(): + async with hibira.AsyncClient.from_env() as c, Socket(c) as ws: + await ws.subscribe(room, {"user_id": "ann", "user_info": {"name": "Ann"}}) + assert (await c.users(room))["users"] == [{"id": "ann"}] + + asyncio.run(main()) + + +def test_errors(): + c = hibira.Client.from_env() + with pytest.raises(hibira.HTTPError) as info: + c.trigger("public-x", "big", "x" * 200_000) + assert info.value.status in (400, 413) + bad = hibira.Client.from_env(secret="wrong-secret-0123456789") + with pytest.raises(hibira.AuthenticationError): + bad.trigger("public-x", "e", 1) + + +def test_stream_sync_and_rewind(channel): + async def main(): + c = hibira.Client.from_env() + async with Socket(c) as ws: + await ws.subscribe(channel) + words = [f"word{i} " for i in range(40)] + ["日本語 😀"] + out = await asyncio.to_thread( + hibira.stream_to_channel, c, channel, iter(words), meta={"role": "assistant"}, interval=0.01 + ) + await ws.wait(lambda e: e[1] == "ai-stream" and e[2].get("done")) + text, parts = rebuild(ws.events, out.id) + assert text == "".join(words) == out.text + assert parts[0]["meta"] == {"role": "assistant"} and parts[-1]["text"] == text + + async with Socket(c) as late: + await late.subscribe(channel) + await late.send("pusher:rewind", {"channel": channel, "seconds": 60}) + ok = await late.wait(lambda e: e[1] in ("pusher:rewind_succeeded", "pusher:rewind_error")) + if ok[1] == "pusher:rewind_error": + pytest.skip(f"channel history is off for this app: {ok[2]}") + text, _ = rebuild(late.events, out.id) + assert text == out.text + + asyncio.run(main()) + + +def test_stream_async(channel): + async def tokens(): + for i in range(20): + yield {"choices": [{"delta": {"content": f"{i} "}}]} + await asyncio.sleep(0.005) + + async def main(): + async with hibira.AsyncClient.from_env() as c, Socket(c) as ws: + await ws.subscribe(channel) + out = await hibira.astream_to_channel(c, channel, tokens(), interval=0.02, max_bytes=200) + await ws.wait(lambda e: e[1] == "ai-stream" and e[2].get("done")) + text, parts = rebuild(ws.events, out.id) + assert text == out.text == "".join(f"{i} " for i in range(20)) + assert all(len(json.dumps(p, separators=(",", ":")).encode()) <= 200 for p in parts) + + asyncio.run(main()) diff --git a/tests/test_fastapi.py b/tests/test_fastapi.py new file mode 100644 index 0000000..42c0736 --- /dev/null +++ b/tests/test_fastapi.py @@ -0,0 +1,71 @@ +import hashlib +import hmac + +import pytest + +fastapi = pytest.importorskip("fastapi") +from fastapi import Depends, FastAPI # noqa: E402 +from fastapi.testclient import TestClient # noqa: E402 + +import hibira # noqa: E402 +from hibira.fastapi import auth_router, webhook # noqa: E402 + +client = hibira.Client("1", "key", "secret") + + +async def authorize(request, channel): + if request.headers.get("x-user") != "ann": + return False + if channel.startswith("presence-"): + return {"user_id": "ann", "user_info": {"name": "Ann"}} + return channel == "private-ann" + + +app = FastAPI() +app.include_router(auth_router(client, authorize, user=lambda r: {"id": "ann"} if r.headers.get("x-user") else None)) +received = [] + + +@app.post("/hook") +def hook(payload: dict = Depends(webhook(client))): + received.append(payload) + return {"ok": True} + + +http = TestClient(app) + + +def test_private_and_presence_auth(): + r = http.post("/hibira/auth", data={"socket_id": "1.2", "channel_name": "private-ann"}, headers={"x-user": "ann"}) + assert r.status_code == 200 and r.json() == client.authenticate_channel("private-ann", "1.2") + r = http.post("/hibira/auth", json={"socket_id": "1.2", "channel_name": "presence-room"}, headers={"x-user": "ann"}) + assert r.json()["channel_data"] == '{"user_id":"ann","user_info":{"name":"Ann"}}' + assert ( + http.post( + "/hibira/auth", data={"socket_id": "1.2", "channel_name": "private-bob"}, headers={"x-user": "ann"} + ).status_code + == 403 + ) + assert http.post("/hibira/auth", data={"socket_id": "1.2", "channel_name": "private-ann"}).status_code == 403 + assert ( + http.post( + "/hibira/auth", data={"socket_id": "x", "channel_name": "private-ann"}, headers={"x-user": "ann"} + ).status_code + == 400 + ) + assert http.post("/hibira/auth", data={}).status_code == 400 + + +def test_user_auth(): + r = http.post("/hibira/user-auth", data={"socket_id": "1.2"}, headers={"x-user": "ann"}) + assert r.json() == client.authenticate_user("1.2", {"id": "ann"}) + assert http.post("/hibira/user-auth", data={"socket_id": "1.2"}).status_code == 403 + + +def test_webhook(): + body = b'{"time_ms":1,"events":[]}' + sig = hmac.new(b"secret", body, hashlib.sha256).hexdigest() + r = http.post("/hook", content=body, headers={"x-pusher-key": "key", "x-pusher-signature": sig}) + assert r.status_code == 200 and received[-1] == {"time_ms": 1, "events": []} + r = http.post("/hook", content=body, headers={"x-pusher-key": "key", "x-pusher-signature": "0" * 64}) + assert r.status_code == 401 diff --git a/tests/test_signing.py b/tests/test_signing.py new file mode 100644 index 0000000..96b612d --- /dev/null +++ b/tests/test_signing.py @@ -0,0 +1,90 @@ +import base64 +import hashlib +import hmac +import json + +import pytest + +import hibira +from hibira import _signing + +KEY = "278d425bdf160c739803" +SECRET = "7ad3773142a6692b25b8" + + +def test_sign_query_matches_pusher_reference(): + body = b'{"name":"foo","channels":["project-3"],"data":"{\\"some\\":\\"data\\"}"}' + q = _signing.sign_query(KEY, SECRET, "POST", "/apps/3/events", body=body, now=1353088179) + assert q["body_md5"] == "ec365a775a4cd0599faeb73354201b6f" + assert q["auth_signature"] == "da454824c97ba181a32ccc17a72625ba02771f50b50e1e7430e47a1f3f457e6c" + + +def test_private_channel_auth_matches_pusher_reference(): + c = hibira.Client("3", KEY, SECRET) + assert c.authenticate_channel("private-foobar", "1234.1234") == { + "auth": "278d425bdf160c739803:58df8b0c36d6982b82c3ecf6b4662e34fe8c25bba48f5369f135bf843651c3a4" + } + + +def test_presence_channel_auth(): + c = hibira.Client("3", KEY, SECRET) + out = c.authenticate_channel("presence-foobar", "1234.1234", {"user_id": 10, "user_info": {"name": "Mr. Channels"}}) + assert out["channel_data"] == '{"user_id":10,"user_info":{"name":"Mr. Channels"}}' + expected = hmac.new( + SECRET.encode(), f"1234.1234:presence-foobar:{out['channel_data']}".encode(), hashlib.sha256 + ).hexdigest() + assert out["auth"] == f"{KEY}:{expected}" + + +@pytest.mark.parametrize( + ("channel", "socket_id", "user_data"), + [ + ("public", "1.1", None), + ("private-a", "nope", None), + ("private-encrypted-a", "1.1", None), + ("presence-a", "1.1", None), + ("presence-a", "1.1", {"user_info": {}}), + ("private-a b", "1.1", None), + ], +) +def test_channel_auth_rejects(channel, socket_id, user_data): + with pytest.raises(ValueError): + hibira.Client("3", KEY, SECRET).authenticate_channel(channel, socket_id, user_data) + + +def test_user_auth(): + out = hibira.Client("3", KEY, SECRET).authenticate_user("1.2", {"id": "u1", "name": "Ann"}) + assert out["user_data"] == '{"id":"u1","name":"Ann"}' + sig = hmac.new(SECRET.encode(), b'1.2::user::{"id":"u1","name":"Ann"}', hashlib.sha256).hexdigest() + assert out["auth"] == f"{KEY}:{sig}" + with pytest.raises(ValueError): + hibira.Client("3", KEY, SECRET).authenticate_user("1.2", {"name": "x"}) + + +def test_verify_webhook(): + c = hibira.Client("3", KEY, SECRET) + body = b'{"time_ms":1,"events":[{"name":"channel_occupied","channel":"a"}]}' + sig = hmac.new(SECRET.encode(), body, hashlib.sha256).hexdigest() + assert c.verify_webhook(KEY, sig, body)["events"][0]["name"] == "channel_occupied" + assert c.verify_webhook(KEY, sig.upper(), body.decode())["time_ms"] == 1 + for key, s, b in [("other", sig, body), (KEY, "0" * 64, body), (KEY, None, body), (KEY, sig, body + b" ")]: + with pytest.raises(hibira.InvalidWebhook): + c.verify_webhook(key, s, b) + + +def test_push_token(): + c = hibira.Client("3", KEY, SECRET, push_instance_id="inst-1", push_secret_key="push-secret") + token = c.push_token("alice", ttl=60)["token"] + header, payload, sig = token.split(".") + + def dec(s): + return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) + + assert json.loads(dec(header)) == {"alg": "HS256", "typ": "JWT"} + claims = json.loads(dec(payload)) + assert claims["sub"] == "alice" + assert claims["iss"] == "https://inst-1.pushnotifications.pusher.com" + assert claims["exp"] - claims["iat"] == 60 + assert dec(sig) == hmac.new(b"push-secret", f"{header}.{payload}".encode(), hashlib.sha256).digest() + with pytest.raises(hibira.ConfigError): + hibira.Client("3", KEY, SECRET).push_token("alice") diff --git a/tests/test_stream.py b/tests/test_stream.py new file mode 100644 index 0000000..b179f1b --- /dev/null +++ b/tests/test_stream.py @@ -0,0 +1,218 @@ +import asyncio +import json +import threading +import time +from types import SimpleNamespace as NS + +import pytest + +import hibira +from hibira.stream import split, text_of + + +class Recorder: + def __init__(self, fail_at=None, delay=0.0): + self.events = [] + self.fail_at = fail_at + self.delay = delay + self.lock = threading.Lock() + + def trigger(self, channel, event, data, socket_id=None): + if self.delay: + time.sleep(self.delay) + with self.lock: + if self.fail_at is not None and len(self.events) == self.fail_at: + raise hibira.PayloadTooLarge(413, "too large") + self.events.append((channel, event, data, socket_id)) + + +class AsyncRecorder(Recorder): + async def trigger(self, channel, event, data, socket_id=None): + await asyncio.sleep(self.delay) + if self.fail_at is not None and len(self.events) == self.fail_at: + raise hibira.PayloadTooLarge(413, "too large") + self.events.append((channel, event, data, socket_id)) + + +def rebuild(events): + payloads = [e[2] for e in events] + assert [p["seq"] for p in payloads] == list(range(len(payloads))) + return "".join(p.get("delta", "") for p in payloads if not p.get("done")), payloads[-1] + + +def test_text_of_shapes(): + assert text_of("hi") == "hi" + assert text_of(b"hi") == "hi" + assert text_of({"type": "content_block_delta", "delta": {"type": "text_delta", "text": "a"}}) == "a" + assert text_of({"type": "content_block_delta", "delta": {"type": "input_json_delta", "partial_json": "{"}}) == "" + assert text_of(NS(type="content_block_delta", delta=NS(type="text_delta", text="b"))) == "b" + assert text_of({"type": "response.output_text.delta", "delta": "c"}) == "c" + assert text_of(NS(type="raw_response_event", data=NS(type="response.output_text.delta", delta="d"))) == "d" + assert text_of(NS(type="run_item_stream_event", item=None)) == "" + assert text_of({"choices": [{"delta": {"content": "e"}}]}) == "e" + assert text_of(NS(choices=[NS(delta=NS(content=None))])) == "" + assert text_of(NS(choices=[])) == "" + ai = NS(type="AIMessageChunk", content=[{"type": "text", "text": "f"}, {"type": "tool_use", "id": "x"}]) + assert text_of((ai, {"langgraph_node": "agent"})) == "f" + assert text_of(("messages", (NS(type="AIMessageChunk", content="g"), {}))) == "g" + assert text_of(NS(type="tool", content="secret tool output")) == "" + assert text_of(NS(text="h")) == "h" + assert text_of({"type": "text-delta", "textDelta": "i"}) == "i" + assert text_of(None) == "" and text_of(42) == "" + + +def test_split_respects_budget_and_characters(): + text = 'héllo "wörld"\n😀' * 50 + for budget in (16, 20, 33): + pieces = split(text, budget) + assert "".join(pieces) == text + for p in pieces: + assert len(json.dumps(p, ensure_ascii=False).encode()) - 2 <= budget + + +def test_stream_batches_and_finishes(): + rec = Recorder() + + def tokens(): + for word in "the quick brown fox".split(): + yield word + " " + + out = hibira.stream_to_channel(rec, "private-chat", tokens(), id="s1", meta={"role": "assistant"}, socket_id="1.2") + assert out == hibira.StreamResult("s1", "the quick brown fox ", len(rec.events)) + text, last = rebuild(rec.events) + assert text == "the quick brown fox " + assert last == {"id": "s1", "seq": len(rec.events) - 1, "done": True, "text": text} + first = rec.events[0] + assert first[0] == "private-chat" and first[1] == "ai-stream" and first[3] == "1.2" + assert first[2]["meta"] == {"role": "assistant"} + assert len(rec.events) == 2 # fast tokens are batched into one event + assert all("meta" not in e[2] for e in rec.events[1:]) + + +def test_stream_interval_flushes_slow_tokens(): + rec = Recorder() + + def tokens(): + yield "a" + time.sleep(0.15) + assert [e[2]["delta"] for e in rec.events] == ["a"] + yield "b" + + hibira.stream_to_channel(rec, "c", tokens(), interval=0.05) + assert rebuild(rec.events)[0] == "ab" + + +def test_stream_splits_large_text_under_max_bytes(): + rec = Recorder() + big = "字" * 5000 + out = hibira.stream_to_channel(rec, "c", [big], max_bytes=1000, meta={"m": 1}) + text, last = rebuild(rec.events) + assert text == big == out.text + for e in rec.events: + assert len(json.dumps(e[2], separators=(",", ":"), ensure_ascii=False).encode()) <= 1000 + assert "text" not in last + + +def test_empty_stream_sends_done_with_meta(): + rec = Recorder() + hibira.stream_to_channel(rec, "c", [], id="x", meta=None) + assert [e[2] for e in rec.events] == [{"id": "x", "seq": 0, "done": True, "meta": None, "text": ""}] + + +def test_source_error_is_sent_and_raised(): + rec = Recorder() + + def tokens(): + yield "partial" + raise RuntimeError("model overloaded") + + with pytest.raises(RuntimeError): + hibira.stream_to_channel(rec, "c", tokens()) + text, last = rebuild(rec.events) + assert text == "partial" and last["error"] == "model overloaded" and last["done"] + + +def test_stop_event_aborts(): + rec = Recorder() + stop = threading.Event() + + def tokens(): + yield "a" + stop.set() + yield "b" + + out = hibira.stream_to_channel(rec, "c", tokens(), stop=stop) + assert out.text == "a" + assert rec.events[-1][2]["error"] == "aborted" + + +def test_send_error_stops_and_raises(): + rec = Recorder(fail_at=0) + with pytest.raises(hibira.PayloadTooLarge): + hibira.stream_to_channel(rec, "c", iter(["a"] * 3), interval=10) + assert rec.events == [] + + +def test_utf8_bytes_split_across_chunks(): + rec = Recorder() + data = "日本語".encode() + out = hibira.stream_to_channel(rec, "c", [data[:1], data[1:4], data[4:]]) + assert out.text == "日本語" + + +def test_max_bytes_too_small(): + with pytest.raises(ValueError): + hibira.stream_to_channel(Recorder(), "c", ["a"], max_bytes=40, meta={"x": "y" * 20}) + + +def test_async_stream_in_order(): + rec = AsyncRecorder(delay=0.01) + + async def tokens(): + for i in range(30): + yield f"{i}," + await asyncio.sleep(0.004) + + out = asyncio.run(hibira.astream_to_channel(rec, "c", tokens(), interval=0.02, meta={"a": 1})) + text, last = rebuild(rec.events) + assert text == out.text == "".join(f"{i}," for i in range(30)) + assert last["done"] and last["text"] == text + assert 2 < len(rec.events) < 30 + + +def test_async_stream_with_sync_client_and_plain_iterable(): + rec = Recorder() + out = asyncio.run(hibira.astream_to_channel(rec, "c", ["x", "y"])) + assert out.text == "xy" and rebuild(rec.events)[0] == "xy" + + +def test_async_cancel_sends_aborted(): + rec = AsyncRecorder() + + async def tokens(): + yield "a" + await asyncio.sleep(10) + yield "b" + + async def main(): + task = asyncio.create_task(hibira.astream_to_channel(rec, "c", tokens(), interval=0.01)) + await asyncio.sleep(0.1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + asyncio.run(main()) + text, last = rebuild(rec.events) + assert text == "a" and last["error"] == "aborted" + + +def test_async_source_error(): + rec = AsyncRecorder() + + async def tokens(): + yield "a" + raise ValueError("boom") + + with pytest.raises(ValueError): + asyncio.run(hibira.astream_to_channel(rec, "c", tokens())) + assert rec.events[-1][2]["error"] == "boom"