diff --git a/CHANGELOG.md b/CHANGELOG.md index 3d12dac..3cfe4f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,26 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/), and this project adheres to [Semantic Versioning](https://semver.org/). +## [Unreleased] + +### Added + +- Pine Platform API support for enterprise integrations. `AsyncPineAI` and + `PineAI` accept a tenant secret key as `api_key`; + `client.platform.managed_users.create()` and `.get()` return typed + `ManagedUser` values and raise `PlatformError` without upstream bodies. +- Managed-user mode: `api_key` plus `managed_user` (the tenant's + `external_id`) acts as that user. REST requests send the key and a + `Pine-Managed-User` header, the Socket.IO handshake sends + `{"token", "managed_user"}`, and `connect()` resolves the Pine user ID through + `auth.me()`. Injected HTTP clients cannot add or override that header. +- `client_name` adds a `Pine-Client` header to every REST request and to the + Socket.IO handshake. +- `API_KEY_PREFIXES` (`pine_sk_live_`, `pine_sk_test_`), `validate_external_id`, + `MANAGED_USER_HEADER` and `CLIENT_HEADER` are public. The managed user is + sent only with the API key it was configured with, and identity headers are + applied after caller headers, so neither can be overridden per request. + ## [0.5.0rc1] - 2026-09-18 Internal release candidate for 0.5.0. See [MIGRATION.md](MIGRATION.md) before diff --git a/README.md b/README.md index b4ca6ea..ed81bee 100644 --- a/README.md +++ b/README.md @@ -106,6 +106,79 @@ never retry or replay form submissions. A transport ACK is not a delivery receipt. The legacy synchronous `send_form_response()` is deprecated because it cannot verify the original form request. +## Integrating as an enterprise (Platform API) + +A Pine Platform integration (a *tenant*) holds a secret key, `pine_sk_live_...` +in production or `pine_sk_test_...` elsewhere. Keep it on your servers: read it +from an environment variable or secret store, and never ship it in browser, +mobile or other client-side code, prompts, or logs. Each end user of your +product becomes a *managed user*, identified by your own `external_id` +(1-128 of letters, digits, `.`, `_`, `:`, `|`, `-`, not starting with `.`; no +personal data). + +Create the managed user once. The call is idempotent per `external_id`: a +repeat returns the stored user unchanged, even if the profile you send differs. +`phone` is optional, in E.164 form, and must be a number you have verified. + +```python +import os + +from pine_assistant import AsyncPineAI + +api_key = os.environ["PINE_API_KEY"] + +async with AsyncPineAI(api_key=api_key) as tenant: + user = await tenant.platform.managed_users.create( + "usr_01J9Z3K8QF", + email="ada@example.com", + name="Ada Lovelace", + phone="+14155550100", + ) + same_user = await tenant.platform.managed_users.get("usr_01J9Z3K8QF") +``` + +Then act as that user with `managed_user`. Every user-scoped REST request sends +the key plus a `Pine-Managed-User` header, and the Socket.IO handshake sends +both; `connect()` looks up the user's Pine user ID once through `auth.me()`. +Two restrictions apply. Every error below subclasses `PineAIError` and carries +`code` and `status_code`: + +- `client.auth.mint_ticket()` always fails with 403 + `platform_route_not_allowed`, raised as `AuthError`. Managed users cannot mint + login tickets. +- While the backend's managed-user access switch is off, every user-scoped call + fails with 403 `platform_managed_user_access_disabled`. The exception type is + the resource's usual one: `SessionError` from `sessions.*`, `AuthError` from + `auth.*` and from `connect()`, which reads `auth.me()` first. + +Billing and admission can also reject sessions and chat for a managed user: +402 `platform_spend_limit_reached`, 403 `platform_access_suspended` or +`platform_managed_user_unavailable`, 409 `platform_session_state_conflict`, 409 +`platform_session_limit_reached` from `sessions.start` when the running-task +limit is reached, and 429 when the tenant quota is exhausted. + +```python +async with AsyncPineAI( + api_key=api_key, managed_user="usr_01J9Z3K8QF", +) as client: + session = await client.sessions.create() + await client.connect() + await client.join_session(session["id"]) + message = "Call my dentist to move Tuesday's appointment" + async for event in client.chat(session["id"], message): + print(event.type, event.data) +``` + +`PineAI` accepts the same `api_key` and `managed_user` for synchronous REST. +`client_name="..."` sends a `Pine-Client` header on REST requests and on the +Socket.IO handshake. The backend records it as the source only for requests made +with a tenant key, and only the value `mcp`; it changes no behavior. Platform +API failures raise `PlatformError` with a stable `code` and `status_code` (for +example 401 for a bad key, 403 for a suspended integration, 404 +`platform_managed_user_not_found`, 409 `platform_managed_user_limit_reached`); +like every SDK error, it never reproduces the upstream body, the key or the +`external_id`. + ## Quick Start (Sync REST) `PineAI` is a synchronous REST client. It returns values directly for auth and diff --git a/src/pine_assistant/__init__.py b/src/pine_assistant/__init__.py index 46ace45..9e47a68 100644 --- a/src/pine_assistant/__init__.py +++ b/src/pine_assistant/__init__.py @@ -12,7 +12,7 @@ from pine_assistant.auth import Auth, SyncAuth from pine_assistant.chat import ChatEvent from pine_assistant.client import AsyncPineAI, PineAI -from pine_assistant.errors import AuthError, ConnectionError, PineAIError, SessionError +from pine_assistant.errors import AuthError, ConnectionError, PineAIError, PlatformError, SessionError from pine_assistant.models.auth import AuthIdentity, AuthTicket, RedeemedTicket from pine_assistant.models.events import ( SUPPORTED_EVENTS, @@ -20,6 +20,7 @@ S2CEvent, is_supported_event, ) +from pine_assistant.models.platform import ManagedUser from pine_assistant.models.session import ( SessionInfo, SessionListResponse, @@ -28,7 +29,16 @@ SessionOutcomeRating, SessionOutcomesPage, ) +from pine_assistant.platform_api import ( + API_KEY_PREFIXES, + ManagedUsersAPI, + PlatformAPI, + SyncManagedUsersAPI, + SyncPlatformAPI, + validate_external_id, +) from pine_assistant.sessions import SessionsAPI, SyncSessionsAPI +from pine_assistant.transport.http import CLIENT_HEADER, MANAGED_USER_HEADER __version__ = "0.5.0rc1" __all__ = [ @@ -38,6 +48,15 @@ "SyncAuth", "SessionsAPI", "SyncSessionsAPI", + "PlatformAPI", + "SyncPlatformAPI", + "ManagedUsersAPI", + "SyncManagedUsersAPI", + "ManagedUser", + "API_KEY_PREFIXES", + "MANAGED_USER_HEADER", + "CLIENT_HEADER", + "validate_external_id", "AuthIdentity", "AuthTicket", "RedeemedTicket", @@ -51,6 +70,7 @@ "PineAIError", "AuthError", "SessionError", + "PlatformError", "ConnectionError", "C2SEvent", "S2CEvent", diff --git a/src/pine_assistant/client.py b/src/pine_assistant/client.py index dddfdb9..1a81a66 100644 --- a/src/pine_assistant/client.py +++ b/src/pine_assistant/client.py @@ -5,6 +5,7 @@ import asyncio import logging import os +import re import uuid from collections.abc import AsyncGenerator, Callable from pathlib import Path @@ -17,6 +18,7 @@ from pine_assistant.errors import ConnectionError from pine_assistant.models.events import C2SEvent from pine_assistant.models.form import FormSubmissionResult, FormToUserData, encode_form_answers +from pine_assistant.platform_api import API_KEY_PREFIXES, PlatformAPI, SyncPlatformAPI, validate_external_id from pine_assistant.sessions import SessionsAPI, SyncSessionsAPI from pine_assistant.transport.http import DEFAULT_API_BASE_PATH, DEFAULT_BASE_URL, HttpClient, SyncHttpClient from pine_assistant.transport.socketio import SocketIOManager @@ -31,9 +33,35 @@ FORM_RESPONSE_TIMEOUT_S = 10.0 FORM_RESPONSE_MAX_TIMEOUT_S = 30.0 +_CLIENT_NAME = re.compile(r"[A-Za-z0-9._-]{1,64}") + _logger = logging.getLogger(__name__) +def _platform_identity( + access_token: str | None, + user_id: str | None, + api_key: str | None, + managed_user: str | None, + client_name: str | None, +) -> str | None: + """Validate the credential mode and return the bearer the client sends. + + Error messages never include the key or the external ID. + """ + if client_name is not None and not (isinstance(client_name, str) and _CLIENT_NAME.fullmatch(client_name)): + raise ValueError("client_name must be 1-64 letters, digits, '.', '_' or '-'") + if api_key is None and managed_user is None: + return access_token + if access_token is not None or user_id is not None: + raise ValueError("api_key cannot be combined with access_token or user_id") + if not isinstance(api_key, str) or not api_key.startswith(API_KEY_PREFIXES): + raise ValueError("api_key must be a Pine Platform secret key starting with pine_sk_live_ or pine_sk_test_") + if managed_user is not None: + validate_external_id(managed_user, "managed_user") + return api_key + + def _get_or_create_device_id(provided: str | None = None) -> str: """Resolve a stable device_id. @@ -69,6 +97,16 @@ class AsyncPineAI: """Async Pine AI client (primary). A client tracks one session. Concurrent sessions need one client each. + + Authenticate either as a Pine user with ``access_token`` (and ``user_id`` + for Socket.IO), or as a Pine Platform tenant with its secret ``api_key``. + A tenant client calls the Platform API through ``platform``; adding + ``managed_user`` (the tenant's ``external_id``) makes it act as that + managed user: every user-scoped REST request carries the key and a + ``Pine-Managed-User`` header, the Socket.IO handshake carries both, and + ``connect()`` resolves the managed user's Pine user ID itself. + ``client_name`` adds a ``Pine-Client`` header to every REST request and + to the Socket.IO handshake. """ def __init__( @@ -83,10 +121,16 @@ def __init__( api_base_path: str = DEFAULT_API_BASE_PATH, http_client: httpx.AsyncClient | None = None, http_transport: httpx.AsyncBaseTransport | None = None, + api_key: str | None = None, + managed_user: str | None = None, + client_name: str | None = None, ): + access_token = _platform_identity(access_token, user_id, api_key, managed_user, client_name) self._base_url = base_url self._access_token = access_token self._user_id = user_id + self._managed_user = managed_user + self._client_name = client_name self._device_id = _get_or_create_device_id(device_id) self._transports = transports self._ready_timeout = ready_timeout @@ -97,9 +141,12 @@ def __init__( api_base_path=api_base_path, client=http_client, transport=http_transport, + managed_user=managed_user, + client_name=client_name, ) self.auth = Auth(self.http) self.sessions = SessionsAPI(self.http) + self.platform = PlatformAPI(self.http) self._sio: SocketIOManager | None = None self._chat: ChatEngine | None = None @@ -115,6 +162,17 @@ async def __aexit__(self, *_exc_info: object) -> None: await self.aclose() async def connect(self, access_token: str | None = None, user_id: str | None = None) -> None: + """Open Socket.IO and wait until the backend reports it ready. + + With ``api_key``/``managed_user`` the identity is fixed at construction; + the managed user's Pine user ID, which envelopes carry, is read once + from ``auth.me()`` and reused for the client's lifetime. + """ + if self._managed_user is not None: + if access_token is not None or user_id is not None: + raise ValueError("a managed-user client cannot connect with another access_token or user_id") + if self._user_id is None: + self._user_id = (await self.auth.me()).user_id token = access_token or self._access_token uid = user_id or self._user_id if not token or not uid: @@ -128,6 +186,8 @@ async def connect(self, access_token: str | None = None, user_id: str | None = N device_id=self._device_id, transports=self._transports, ready_timeout=self._ready_timeout, + managed_user=self._managed_user, + client_name=self._client_name, ) self._chat = ChatEngine(self._sio, check_session_state=self._session_state) await self._sio.connect() @@ -433,6 +493,8 @@ class PineAI: REST resources return concrete values through ``httpx.Client``. Real-time Socket.IO requires an event loop and is intentionally available only on :class:`AsyncPineAI`; this client does not create a hidden loop or thread. + ``api_key``, ``managed_user`` and ``client_name`` behave as on + :class:`AsyncPineAI`. """ def __init__( @@ -443,16 +505,22 @@ def __init__( api_base_path: str = DEFAULT_API_BASE_PATH, http_client: httpx.Client | None = None, http_transport: httpx.BaseTransport | None = None, + api_key: str | None = None, + managed_user: str | None = None, + client_name: str | None = None, ) -> None: self.http = SyncHttpClient( base_url=base_url, - token=access_token, + token=_platform_identity(access_token, None, api_key, managed_user, client_name), api_base_path=api_base_path, client=http_client, transport=http_transport, + managed_user=managed_user, + client_name=client_name, ) self.auth = SyncAuth(self.http) self.sessions = SyncSessionsAPI(self.http) + self.platform = SyncPlatformAPI(self.http) def __enter__(self) -> "PineAI": return self diff --git a/src/pine_assistant/errors.py b/src/pine_assistant/errors.py index d8ac9fb..d6adcee 100644 --- a/src/pine_assistant/errors.py +++ b/src/pine_assistant/errors.py @@ -37,6 +37,13 @@ def __init__( super().__init__(code, message, details, status_code=status_code) +class PlatformError(PineAIError): + """A Platform API failure: stable ``code`` and ``status_code``, never the upstream body.""" + + def __init__(self, message: str, code: str = "platform_error", *, status_code: int | None = None): + super().__init__(code, message, status_code=status_code) + + class ConnectionError(PineAIError): def __init__(self, message: str): super().__init__("connection_error", message) diff --git a/src/pine_assistant/models/platform.py b/src/pine_assistant/models/platform.py new file mode 100644 index 0000000..d8ce276 --- /dev/null +++ b/src/pine_assistant/models/platform.py @@ -0,0 +1,27 @@ +"""Typed responses for the Pine Platform API.""" + +from pydantic import BaseModel, ConfigDict, field_validator + +from pine_assistant.models.session import SessionInfo + + +class ManagedUser(BaseModel): + """One end user of a Platform integration, addressed by its ``external_id``. + + ``id`` is the Pine user ID that ``/api/v2`` and Socket.IO payloads carry; + the Platform API never accepts it as input. + """ + + model_config = ConfigDict(extra="allow") + + id: str + external_id: str + email: str + name: str + phone: str | None + created_at: str + + @field_validator("id", mode="before") + @classmethod + def normalize_id(cls, value: object) -> str: + return SessionInfo.normalize_id(value) diff --git a/src/pine_assistant/platform_api.py b/src/pine_assistant/platform_api.py new file mode 100644 index 0000000..7b8fe31 --- /dev/null +++ b/src/pine_assistant/platform_api.py @@ -0,0 +1,114 @@ +""" +Pine Platform API — enterprise (tenant) resources authenticated by a secret key. + +A tenant's secret key (``pine_sk_live_...`` / ``pine_sk_test_...``) belongs on +its servers only. These requests act as the tenant itself and never carry a +``Pine-Managed-User`` header. +""" + +from __future__ import annotations + +import re +from urllib.parse import quote + +from pydantic import ValidationError + +from pine_assistant.errors import PineAIError, PlatformError +from pine_assistant.models.platform import ManagedUser +from pine_assistant.transport.http import HttpClient, SyncHttpClient + +API_KEY_PREFIXES = ("pine_sk_live_", "pine_sk_test_") +_EXTERNAL_ID = re.compile(r"[A-Za-z0-9_:|-][A-Za-z0-9._:|-]{0,127}") +_MANAGED_USERS = "/platform/v1/managed-users" + + +def validate_external_id(value: str, field: str = "external_id") -> str: + """Check the backend's external_id rule; the value is never echoed on failure.""" + if not isinstance(value, str) or not _EXTERNAL_ID.fullmatch(value): + raise ValueError( + f"{field} must be 1-128 letters, digits, '.', '_', ':', '|' or '-' and must not start with '.'" + ) + return value + + +def _create_body(external_id: str, email: str, name: str, phone: str | None) -> dict[str, str]: + body = {"external_id": validate_external_id(external_id), "email": email, "name": name} + if phone is not None: + body["phone"] = phone + return body + + +def _path(external_id: str) -> str: + return f"{_MANAGED_USERS}/{quote(validate_external_id(external_id), safe='')}" + + +def _platform_failure(error: PineAIError) -> PlatformError: + return PlatformError("Pine Platform API request failed", error.code, status_code=error.status_code) + + +def _parse_managed_user(payload: object) -> ManagedUser: + try: + return ManagedUser.model_validate(payload) + except ValidationError: + raise PlatformError("Pine Platform API returned an invalid managed user", "invalid_response") from None + + +class ManagedUsersAPI: + def __init__(self, http: HttpClient): + self._http = http + + async def create(self, external_id: str, *, email: str, name: str, phone: str | None = None) -> ManagedUser: + """Create the managed user for ``external_id``, or return the existing one unchanged. + + Idempotent per integration and ``external_id``: a repeated call returns + the stored user even when ``email``, ``name`` or ``phone`` differ. + ``phone`` is an E.164 number you have verified. + """ + body = _create_body(external_id, email, name, phone) + try: + payload = await self._http.post(_MANAGED_USERS, body, as_tenant=True) + except PineAIError as exc: + raise _platform_failure(exc) from None + return _parse_managed_user(payload) + + async def get(self, external_id: str) -> ManagedUser: + """Return the managed user for ``external_id``; another integration's user is a 404.""" + path = _path(external_id) + try: + payload = await self._http.get(path, as_tenant=True) + except PineAIError as exc: + raise _platform_failure(exc) from None + return _parse_managed_user(payload) + + +class SyncManagedUsersAPI: + def __init__(self, http: SyncHttpClient): + self._http = http + + def create(self, external_id: str, *, email: str, name: str, phone: str | None = None) -> ManagedUser: + body = _create_body(external_id, email, name, phone) + try: + payload = self._http.post(_MANAGED_USERS, body, as_tenant=True) + except PineAIError as exc: + raise _platform_failure(exc) from None + return _parse_managed_user(payload) + + def get(self, external_id: str) -> ManagedUser: + path = _path(external_id) + try: + payload = self._http.get(path, as_tenant=True) + except PineAIError as exc: + raise _platform_failure(exc) from None + return _parse_managed_user(payload) + + +class PlatformAPI: + """Platform API resources for a client built with ``api_key``.""" + + def __init__(self, http: HttpClient): + self.managed_users = ManagedUsersAPI(http) + + +class SyncPlatformAPI: + def __init__(self, http: SyncHttpClient): + self.managed_users = SyncManagedUsersAPI(http) diff --git a/src/pine_assistant/transport/http.py b/src/pine_assistant/transport/http.py index f09b67f..2341ba3 100644 --- a/src/pine_assistant/transport/http.py +++ b/src/pine_assistant/transport/http.py @@ -11,6 +11,9 @@ DEFAULT_BASE_URL = "https://www.19pine.ai" DEFAULT_API_BASE_PATH = "/api" _USER_AGENT = "pine-assistant-sdk" +MANAGED_USER_HEADER = "Pine-Managed-User" +CLIENT_HEADER = "Pine-Client" +_IDENTITY_HEADERS = ("Authorization", MANAGED_USER_HEADER, CLIENT_HEADER) # Public HTTPX timeout inputs accepted by both sync and async clients. HTTPX's # internal ``TimeoutTypes`` alias is not exported in every supported release. @@ -87,7 +90,47 @@ def _json_data(response: httpx.Response) -> Any: return payload -class HttpClient: +class _RequestIdentity: + """The credential, managed user and client name shared by the sync and async transports.""" + + def __init__(self, token: str | None, managed_user: str | None, client_name: str | None) -> None: + self._token = token + self._managed_user = managed_user + # A managed user is bound to the API key it was configured with and is + # never asserted with any other token, e.g. one installed by set_token(). + self._managed_user_key = token if managed_user else None + self._client_name = client_name + + def set_token(self, token: str | None) -> None: + self._token = token + + def _identity(self, authenticated: bool, token: str | None, as_tenant: bool) -> dict[str, str]: + """The identity headers a request carries; Platform requests the tenant makes as itself omit the user.""" + identity: dict[str, str] = {} + actual_token = self._token if token is None else token + if authenticated and actual_token: + identity["Authorization"] = f"Bearer {actual_token}" + if self._managed_user and actual_token == self._managed_user_key and not as_tenant: + identity[MANAGED_USER_HEADER] = self._managed_user + if self._client_name: + identity[CLIENT_HEADER] = self._client_name + return identity + + @staticmethod + def _apply_identity(request: httpx.Request, identity: dict[str, str]) -> None: + """Make the identity headers exactly the computed ones. + + Applied last, so neither an injected client's defaults nor caller + headers can authenticate a request or choose who it acts as. + """ + for name in _IDENTITY_HEADERS: + if name in identity: + request.headers[name] = identity[name] + else: + request.headers.pop(name, None) + + +class HttpClient(_RequestIdentity): """Async HTTP client; injected clients remain caller-owned.""" def __init__( @@ -99,11 +142,13 @@ def __init__( client: httpx.AsyncClient | None = None, transport: httpx.AsyncBaseTransport | None = None, timeout: TimeoutConfig = 30.0, + managed_user: str | None = None, + client_name: str | None = None, ): if client is not None and transport is not None: raise ValueError("pass either client or transport, not both") + super().__init__(token, managed_user, client_name) self._api_base_url = _api_url(base_url, api_base_path) - self._token = token self._owns_client = client is None self._client = client or httpx.AsyncClient( headers={"User-Agent": _USER_AGENT, "Accept": "application/json"}, @@ -112,15 +157,6 @@ def __init__( follow_redirects=False, ) - def set_token(self, token: str | None) -> None: - self._token = token - - def _auth_headers(self, authenticated: bool, token: str | None = None) -> dict[str, str]: - headers: dict[str, str] = {"Content-Type": "application/json"} - actual_token = self._token if token is None else token - if authenticated and actual_token: - headers["Authorization"] = f"Bearer {actual_token}" - return headers def _url(self, path: str) -> str: # Absolute paths keep an injected client's base_url from changing the @@ -138,11 +174,10 @@ async def _request( token: str | None = None, files: Any = None, extra_headers: dict[str, str] | None = None, + as_tenant: bool = False, ) -> Any: - actual_token = self._token if token is None else token - headers = self._auth_headers(authenticated, token) - if extra_headers: - headers.update(extra_headers) + identity = self._identity(authenticated, token, as_tenant) + headers = {"Content-Type": "application/json", **(extra_headers or {})} if files is not None: # httpx must generate the multipart boundary itself. headers.pop("Content-Type", None) @@ -150,12 +185,7 @@ async def _request( request = self._client.build_request( method, self._url(path), json=body, params=params, headers=headers, files=files, ) - # An injected client may carry default authentication. Never let it - # authenticate an anonymous request or a request for another user. - if authenticated and actual_token: - request.headers["Authorization"] = f"Bearer {actual_token}" - else: - request.headers.pop("Authorization", None) + self._apply_identity(request, identity) response = await self._client.send(request, follow_redirects=False, auth=None) except httpx.TimeoutException as exc: raise PineAIError("timeout", "Pine API request timed out") from exc @@ -164,13 +194,17 @@ async def _request( return _json_data(response) async def get(self, path: str, authenticated: bool = True, *, token: str | None = None, - params: dict[str, str | int] | None = None) -> Any: - return await self._request("GET", path, authenticated=authenticated, token=token, params=params) + params: dict[str, str | int] | None = None, as_tenant: bool = False) -> Any: + return await self._request( + "GET", path, authenticated=authenticated, token=token, params=params, as_tenant=as_tenant, + ) async def post(self, path: str, body: dict[str, Any] | None = None, authenticated: bool = True, - *, token: str | None = None, headers: dict[str, str] | None = None) -> Any: + *, token: str | None = None, headers: dict[str, str] | None = None, + as_tenant: bool = False) -> Any: return await self._request( "POST", path, body=body, authenticated=authenticated, token=token, extra_headers=headers, + as_tenant=as_tenant, ) async def put(self, path: str, body: dict[str, Any] | None = None, authenticated: bool = True, @@ -195,7 +229,7 @@ async def close(self) -> None: await self._client.aclose() -class SyncHttpClient: +class SyncHttpClient(_RequestIdentity): """Synchronous REST transport with the same serialization and error rules.""" def __init__( @@ -207,11 +241,13 @@ def __init__( client: httpx.Client | None = None, transport: httpx.BaseTransport | None = None, timeout: TimeoutConfig = 30.0, + managed_user: str | None = None, + client_name: str | None = None, ) -> None: if client is not None and transport is not None: raise ValueError("pass either client or transport, not both") + super().__init__(token, managed_user, client_name) self._api_base_url = _api_url(base_url, api_base_path) - self._token = token self._owns_client = client is None self._client = client or httpx.Client( headers={"User-Agent": _USER_AGENT, "Accept": "application/json"}, @@ -220,15 +256,6 @@ def __init__( follow_redirects=False, ) - def set_token(self, token: str | None) -> None: - self._token = token - - def _auth_headers(self, authenticated: bool, token: str | None = None) -> dict[str, str]: - headers = {"Content-Type": "application/json"} - actual_token = self._token if token is None else token - if authenticated and actual_token: - headers["Authorization"] = f"Bearer {actual_token}" - return headers def _url(self, path: str) -> str: return f"{self._api_base_url}/{path.lstrip('/')}" @@ -244,21 +271,17 @@ def _request( token: str | None = None, files: Any = None, extra_headers: dict[str, str] | None = None, + as_tenant: bool = False, ) -> Any: - actual_token = self._token if token is None else token - headers = self._auth_headers(authenticated, token) - if extra_headers: - headers.update(extra_headers) + identity = self._identity(authenticated, token, as_tenant) + headers = {"Content-Type": "application/json", **(extra_headers or {})} if files is not None: headers.pop("Content-Type", None) try: request = self._client.build_request( method, self._url(path), json=body, params=params, headers=headers, files=files, ) - if authenticated and actual_token: - request.headers["Authorization"] = f"Bearer {actual_token}" - else: - request.headers.pop("Authorization", None) + self._apply_identity(request, identity) response = self._client.send(request, follow_redirects=False, auth=None) except httpx.TimeoutException as exc: raise PineAIError("timeout", "Pine API request timed out") from exc @@ -267,13 +290,17 @@ def _request( return _json_data(response) def get(self, path: str, authenticated: bool = True, *, token: str | None = None, - params: dict[str, str | int] | None = None) -> Any: - return self._request("GET", path, authenticated=authenticated, token=token, params=params) + params: dict[str, str | int] | None = None, as_tenant: bool = False) -> Any: + return self._request( + "GET", path, authenticated=authenticated, token=token, params=params, as_tenant=as_tenant, + ) def post(self, path: str, body: dict[str, Any] | None = None, authenticated: bool = True, - *, token: str | None = None, headers: dict[str, str] | None = None) -> Any: + *, token: str | None = None, headers: dict[str, str] | None = None, + as_tenant: bool = False) -> Any: return self._request( "POST", path, body=body, authenticated=authenticated, token=token, extra_headers=headers, + as_tenant=as_tenant, ) def put(self, path: str, body: dict[str, Any] | None = None, authenticated: bool = True, diff --git a/src/pine_assistant/transport/socketio.py b/src/pine_assistant/transport/socketio.py index 33ccf57..9aaa1b0 100644 --- a/src/pine_assistant/transport/socketio.py +++ b/src/pine_assistant/transport/socketio.py @@ -1,7 +1,8 @@ """ Socket.IO connection manager — spec sections 3.1, 5.1.2. -Connection: wss://{baseUrl}/api/v2/socket.io/ with auth={token}. +Connection: wss://{baseUrl}/api/v2/socket.io/ with auth={token}, plus +managed_user when a Platform API key acts for a managed user. Waits for `ready` event before resolving connect(). """ @@ -15,6 +16,7 @@ from pine_assistant.errors import ConnectionError as PineConnectionError from pine_assistant.models.form import FormSubmissionResult +from pine_assistant.transport.http import CLIENT_HEADER SOCKETIO_PATH = "/api/v2/socket.io/" @@ -41,9 +43,13 @@ def __init__( device_id: str | None = None, transports: list[str] | None = None, ready_timeout: float = 15.0, + managed_user: str | None = None, + client_name: str | None = None, ): self._base_url = base_url self._token = token + self._managed_user = managed_user + self._client_name = client_name self._user_id = user_id self._device_id = device_id or str(uuid.uuid4()) self._transports = transports or ["websocket"] @@ -144,7 +150,8 @@ async def disconnect(_reason: str = "") -> None: try: await self._sio.connect( self._base_url, - auth={"token": self._token}, + auth=self._handshake_auth(), + headers={CLIENT_HEADER: self._client_name} if self._client_name else {}, transports=self._transports, socketio_path=SOCKETIO_PATH, wait_timeout=self._ready_timeout, @@ -190,6 +197,12 @@ async def disconnect(_reason: str = "") -> None: if key in self._form_submission_pending_keys } + def _handshake_auth(self) -> dict[str, str]: + auth = {"token": self._token} + if self._managed_user: + auth["managed_user"] = self._managed_user + return auth + def _track_membership(self, event_type: str, session_id: str | None) -> None: """Remember which sessions to re-join after a reconnect. diff --git a/tests/test_platform.py b/tests/test_platform.py new file mode 100644 index 0000000..dcc823f --- /dev/null +++ b/tests/test_platform.py @@ -0,0 +1,415 @@ +"""Platform API and managed-user mode against a fake backend on 127.0.0.1.""" + +import json +from dataclasses import dataclass, field +from typing import Any + +import httpx +import pytest +import socketio +from aiohttp import web + +from pine_assistant import ( + API_KEY_PREFIXES, + MANAGED_USER_HEADER, + AsyncPineAI, + AuthError, + ManagedUser, + PineAI, + PineAIError, + PlatformError, + SessionError, + validate_external_id, +) + +API_KEY = "pine_sk_test_c2VjcmV0LWtleS1mb3ItdGVzdHM" +EXTERNAL_ID = "acme:user|42-x" +ENCODED_EXTERNAL_ID = "acme%3Auser%7C42-x" +PINE_USER_ID = "4242" + + +def success(data): + return {"status": "success", "data": data} + + +def managed_user(**overrides): + return { + "id": PINE_USER_ID, + "external_id": EXTERNAL_ID, + "email": "ada@example.com", + "name": "Ada Lovelace", + "phone": None, + "created_at": "2026-10-07T00:00:00Z", + **overrides, + } + + +STORED_USER = managed_user(email="stored@example.com", name="Stored Name", phone="+14155550199") + + +@dataclass +class FakeBackend: + base_url: str = "" + requests: list[dict[str, Any]] = field(default_factory=list) + handshakes: list[Any] = field(default_factory=list) + handshake_clients: list[str | None] = field(default_factory=list) + envelopes: list[dict[str, Any]] = field(default_factory=list) + error_status: int | None = None + error_code: str = "platform_failure" + existing: bool = False + + def app(self) -> web.Application: + sio = socketio.AsyncServer(async_mode="aiohttp") + app = web.Application() + sio.attach(app, socketio_path="/api/v2/socket.io/") + + @sio.event + async def connect(sid, environ, auth): + self.handshakes.append(auth) + self.handshake_clients.append(environ.get("HTTP_PINE_CLIENT")) + await sio.emit("ready", {}, to=sid) + + @sio.on("session:history") + async def history(sid, envelope): + self.envelopes.append(envelope) + await sio.emit( + "session:history", + { + "metadata": {"request_id": envelope["metadata"]["request_id"]}, + "payload": {"session_id": envelope["payload"]["session_id"], "data": {"messages": []}}, + }, + to=sid, + ) + + async def record(request: web.Request) -> web.Response | None: + body = await request.text() + self.requests.append({ + "method": request.method, + "path": request.raw_path, + "headers": dict(request.headers), + "body": json.loads(body) if body else None, + }) + if self.error_status is not None: + # A hostile body: none of it may surface through the SDK. + return web.json_response( + {"status": "error", "error": {"code": self.error_code, "message": f"{API_KEY} {EXTERNAL_ID}"}}, + status=self.error_status, + ) + return None + + async def create_user(request): + failure = await record(request) + if failure: + return failure + if self.existing: + # The stored user is returned unchanged, whatever the request carried. + return web.json_response(success(STORED_USER), status=200) + return web.json_response(success(managed_user(**self.requests[-1]["body"])), status=201) + + async def get_user(request): + return await record(request) or web.json_response(success(managed_user())) + + async def auth_me(request): + return await record(request) or web.json_response(success({"user_id": PINE_USER_ID})) + + async def sessions(request): + return await record(request) or web.json_response( + success({"sessions": [{"id": "7"}], "total": 1, "limit": 30, "offset": 0}) + ) + + app.router.add_post("/api/platform/v1/managed-users", create_user) + app.router.add_get("/api/platform/v1/managed-users/{external_id}", get_user) + app.router.add_get("/api/v2/auth/me", auth_me) + app.router.add_get("/api/v2/sessions", sessions) + app.router.add_post("/api/v2/sessions", sessions) + app.router.add_post("/api/v2/auth/tickets", sessions) + return app + + +@pytest.fixture +async def backend(): + fake = FakeBackend() + runner = web.AppRunner(fake.app()) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + host, port = runner.addresses[0][:2] + fake.base_url = f"http://{host}:{port}" + try: + yield fake + finally: + await runner.cleanup() + + +def _assert_no_secrets(text: str) -> None: + assert API_KEY not in text + assert EXTERNAL_ID not in text + + +@pytest.mark.asyncio +async def test_managed_user_rest_requests_carry_key_header_and_client_name(backend): + async with AsyncPineAI( + api_key=API_KEY, managed_user=EXTERNAL_ID, client_name="mcp", base_url=backend.base_url, + ) as client: + listed = await client.sessions.list() + + assert listed.sessions[0].id == "7" + [request] = backend.requests + assert request["headers"]["Authorization"] == f"Bearer {API_KEY}" + assert request["headers"]["Pine-Managed-User"] == EXTERNAL_ID + assert request["headers"]["Pine-Client"] == "mcp" + + +@pytest.mark.asyncio +async def test_managed_user_connect_resolves_user_once_and_sends_handshake(backend): + async with AsyncPineAI(api_key=API_KEY, managed_user=EXTERNAL_ID, base_url=backend.base_url) as client: + await client.connect() + await client.get_history("7") + await client.disconnect() + await client.connect() + + assert [request["path"] for request in backend.requests] == ["/api/v2/auth/me"] + assert backend.requests[0]["headers"]["Pine-Managed-User"] == EXTERNAL_ID + assert "Pine-Client" not in backend.requests[0]["headers"] + assert backend.handshakes == [{"token": API_KEY, "managed_user": EXTERNAL_ID}] * 2 + assert backend.handshake_clients == [None, None] + assert backend.envelopes[0]["metadata"]["source"]["user_id"] == PINE_USER_ID + + +@pytest.mark.asyncio +async def test_client_name_is_sent_on_the_socketio_handshake(backend): + async with AsyncPineAI( + api_key=API_KEY, managed_user=EXTERNAL_ID, client_name="mcp", base_url=backend.base_url, + ) as client: + await client.connect() + assert backend.requests[0]["headers"]["Pine-Client"] == "mcp" + assert backend.handshake_clients == ["mcp"] + + +@pytest.mark.asyncio +async def test_user_token_handshake_is_unchanged(backend): + async with AsyncPineAI(access_token="user-token", user_id="17", base_url=backend.base_url) as client: + await client.connect() + assert backend.handshakes == [{"token": "user-token"}] + assert backend.requests == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("existing", "phone"), [(False, "+14155550100"), (True, None)]) +async def test_create_managed_user_is_typed_and_acts_as_the_tenant(backend, existing, phone): + backend.existing = existing + async with AsyncPineAI(api_key=API_KEY, managed_user="someone-else", base_url=backend.base_url) as client: + user = await client.platform.managed_users.create( + EXTERNAL_ID, email="ada@example.com", name="Ada Lovelace", phone=phone, + ) + + assert isinstance(user, ManagedUser) + assert user.id == PINE_USER_ID + assert user.external_id == EXTERNAL_ID + if existing: + assert user == ManagedUser.model_validate(STORED_USER) + else: + assert (user.email, user.name, user.phone) == ("ada@example.com", "Ada Lovelace", phone) + [request] = backend.requests + assert request["method"] == "POST" + assert request["path"] == "/api/platform/v1/managed-users" + expected = {"external_id": EXTERNAL_ID, "email": "ada@example.com", "name": "Ada Lovelace"} + assert request["body"] == (expected | {"phone": phone} if phone else expected) + assert request["headers"]["Authorization"] == f"Bearer {API_KEY}" + assert "Pine-Managed-User" not in request["headers"] + + +@pytest.mark.asyncio +async def test_get_managed_user_percent_encodes_the_external_id(backend): + async with AsyncPineAI(api_key=API_KEY, base_url=backend.base_url) as client: + user = await client.platform.managed_users.get(EXTERNAL_ID) + + assert user.id == PINE_USER_ID + assert backend.requests[0]["path"] == f"/api/platform/v1/managed-users/{ENCODED_EXTERNAL_ID}" + assert "Pine-Managed-User" not in backend.requests[0]["headers"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", [400, 401, 403, 404, 409, 429, 500]) +async def test_platform_errors_keep_code_and_status_without_upstream_body(backend, status): + backend.error_status = status + async with AsyncPineAI(api_key=API_KEY, base_url=backend.base_url) as client: + with pytest.raises(PlatformError) as excinfo: + await client.platform.managed_users.get(EXTERNAL_ID) + + assert excinfo.value.status_code == status + assert excinfo.value.code == "platform_failure" + assert excinfo.value.__cause__ is None + _assert_no_secrets(str(excinfo.value)) + _assert_no_secrets(repr(excinfo.value)) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", [401, 403, 404]) +async def test_managed_user_resolution_failure_is_typed_and_does_not_connect(backend, status): + backend.error_status = status + async with AsyncPineAI(api_key=API_KEY, managed_user=EXTERNAL_ID, base_url=backend.base_url) as client: + with pytest.raises(PineAIError) as excinfo: + await client.connect() + assert excinfo.value.status_code == status + assert backend.handshakes == [] + _assert_no_secrets(str(excinfo.value)) + + +@pytest.mark.parametrize( + "kwargs", + [ + {"api_key": API_KEY, "managed_user": EXTERNAL_ID, "access_token": "user-token"}, + {"api_key": API_KEY, "managed_user": EXTERNAL_ID, "user_id": "17"}, + {"managed_user": EXTERNAL_ID}, + {"api_key": "not-a-pine-key", "managed_user": EXTERNAL_ID}, + {"api_key": "pine_sk_prod_c2VjcmV0", "managed_user": EXTERNAL_ID}, + {"api_key": "pine_sk_c2VjcmV0"}, + {"api_key": API_KEY, "managed_user": ".leading-dot"}, + {"api_key": API_KEY, "managed_user": "x" * 129}, + {"api_key": API_KEY, "managed_user": "has space"}, + {"api_key": API_KEY, "client_name": "bad name"}, + ], +) +def test_invalid_platform_configuration_is_rejected_without_echoing_values(kwargs): + client_types = (AsyncPineAI,) if "user_id" in kwargs else (AsyncPineAI, PineAI) + for client_type in client_types: + with pytest.raises(ValueError) as excinfo: + client_type(base_url="https://pine.test", **kwargs) + _assert_no_secrets(str(excinfo.value)) + for value in kwargs.values(): + assert value not in str(excinfo.value) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("external_id", ["", "a/b", "../x", ".hidden", "x" * 129]) +async def test_invalid_external_id_is_rejected_before_any_request(backend, external_id): + async with AsyncPineAI(api_key=API_KEY, base_url=backend.base_url) as client: + with pytest.raises(ValueError): + await client.platform.managed_users.get(external_id) + with pytest.raises(ValueError): + await client.platform.managed_users.create(external_id, email="a@example.com", name="A") + assert backend.requests == [] + + +@pytest.mark.asyncio +async def test_injected_client_cannot_add_or_override_the_managed_user(): + seen = [] + + async def handler(request): + seen.append(request) + if request.url.path.startswith("/api/platform/"): + return httpx.Response(200, json=success(managed_user())) + return httpx.Response(200, json=success({"user_id": "1"})) + + injected = httpx.AsyncClient( + headers={"Pine-Managed-User": "injected"}, transport=httpx.MockTransport(handler), + ) + async with AsyncPineAI(access_token="user-token", base_url="https://pine.test", http_client=injected) as client: + await client.auth.me() + async with AsyncPineAI( + api_key=API_KEY, managed_user=EXTERNAL_ID, base_url="https://pine.test", http_client=injected, + ) as client: + await client.auth.me() + await client.platform.managed_users.get(EXTERNAL_ID) + await injected.aclose() + + assert "Pine-Managed-User" not in seen[0].headers + assert seen[1].headers["Pine-Managed-User"] == EXTERNAL_ID + assert "Pine-Managed-User" not in seen[2].headers + + +def test_sync_client_supports_platform_and_managed_user_requests(): + seen = [] + + def handler(request): + seen.append(request) + if request.url.path.startswith("/api/platform/"): + return httpx.Response(201, json=success(managed_user())) + return httpx.Response(200, json=success({"sessions": [], "total": 0, "limit": 30, "offset": 0})) + + with PineAI( + api_key=API_KEY, managed_user=EXTERNAL_ID, client_name="mcp", base_url="https://pine.test", + http_transport=httpx.MockTransport(handler), + ) as client: + created = client.platform.managed_users.create(EXTERNAL_ID, email="ada@example.com", name="Ada Lovelace") + client.sessions.list() + + assert created.id == PINE_USER_ID + assert seen[0].url.raw_path == b"/api/platform/v1/managed-users" + assert "Pine-Managed-User" not in seen[0].headers + assert seen[1].headers["Pine-Managed-User"] == EXTERNAL_ID + assert all(request.headers["Authorization"] == f"Bearer {API_KEY}" for request in seen) + assert all(request.headers["Pine-Client"] == "mcp" for request in seen) + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("status", "code"), [(403, "platform_route_not_allowed"), (429, "rate_limited")]) +async def test_user_scoped_failures_keep_code_and_status_without_upstream_body(backend, status, code): + backend.error_status, backend.error_code = status, code + async with AsyncPineAI(api_key=API_KEY, managed_user=EXTERNAL_ID, base_url=backend.base_url) as client: + with pytest.raises(SessionError) as excinfo: + await client.sessions.list() + assert (excinfo.value.code, excinfo.value.status_code) == (code, status) + _assert_no_secrets(str(excinfo.value)) + + +@pytest.mark.asyncio +async def test_managed_user_is_never_sent_with_a_token_other_than_the_key(backend): + async with AsyncPineAI(api_key=API_KEY, managed_user=EXTERNAL_ID, base_url=backend.base_url) as client: + client.http.set_token("user-jwt") + await client.sessions.list() + await client.http.get("/v2/sessions", token=API_KEY) + await client.http.get("/v2/sessions", token="other-jwt") + + first, second, third = (request["headers"] for request in backend.requests) + assert first["Authorization"] == "Bearer user-jwt" and MANAGED_USER_HEADER not in first + assert second["Authorization"] == f"Bearer {API_KEY}" and second[MANAGED_USER_HEADER] == EXTERNAL_ID + assert third["Authorization"] == "Bearer other-jwt" and MANAGED_USER_HEADER not in third + + +@pytest.mark.asyncio +async def test_caller_headers_cannot_override_identity(backend): + forged = {"Authorization": "Bearer forged", MANAGED_USER_HEADER: "someone-else", "Pine-Client": "forged"} + async with AsyncPineAI(api_key=API_KEY, managed_user=EXTERNAL_ID, base_url=backend.base_url) as client: + await client.http.post("/v2/sessions", {}, headers=forged) + async with AsyncPineAI(access_token="user-token", base_url=backend.base_url) as client: + await client.http.post("/v2/sessions", {}, headers=forged) + + managed, user = (request["headers"] for request in backend.requests) + assert managed["Authorization"] == f"Bearer {API_KEY}" + assert managed[MANAGED_USER_HEADER] == EXTERNAL_ID + assert "Pine-Client" not in managed + assert user["Authorization"] == "Bearer user-token" + assert MANAGED_USER_HEADER not in user and "Pine-Client" not in user + + +def test_public_platform_names(): + assert API_KEY_PREFIXES == ("pine_sk_live_", "pine_sk_test_") + assert MANAGED_USER_HEADER == "Pine-Managed-User" + assert validate_external_id(EXTERNAL_ID) == EXTERNAL_ID + + +AUTH_ATTEMPT_ID = "a" * 22 +CODE_CHALLENGE = "c" * 43 + + +@pytest.mark.asyncio +async def test_documented_managed_user_restrictions_raise_the_resource_error_type(backend): + async with AsyncPineAI(api_key=API_KEY, managed_user=EXTERNAL_ID, base_url=backend.base_url) as client: + backend.error_status, backend.error_code = 403, "platform_route_not_allowed" + with pytest.raises(AuthError) as minted: + await client.auth.mint_ticket( + purpose="mcp", client_id="client", auth_attempt_id=AUTH_ATTEMPT_ID, code_challenge=CODE_CHALLENGE, + ) + backend.error_code = "platform_managed_user_access_disabled" + with pytest.raises(SessionError) as listed: + await client.sessions.list() + with pytest.raises(AuthError) as connected: + await client.connect() + + assert (minted.value.code, minted.value.status_code) == ("platform_route_not_allowed", 403) + for excinfo in (listed, connected): + assert (excinfo.value.code, excinfo.value.status_code) == ("platform_managed_user_access_disabled", 403) + _assert_no_secrets(str(excinfo.value)) + assert backend.handshakes == []