From 07974a7d47ff594b2e7ee14abb512c37e6c22e90 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Thu, 16 Jul 2026 22:42:48 +0400 Subject: [PATCH 01/36] feat: add Python OBO sample for Work IQ Adds python/obo/ -- a FastAPI middle-tier service that accepts a frontend's access token, exchanges it for a Work IQ token via the On-Behalf-Of flow, and calls the Copilot Chat REST API through the Work IQ Gateway. Every existing sample is a public client that signs a user in directly. This is the first middle-tier sample: the user signs in to your frontend, and the service brokers the Work IQ call on their behalf. A frontend token cannot be forwarded as-is -- its `aud` is your own API, so the Gateway rejects it with 401. Audience is signed into the JWT, so the token must be exchanged, not re-pointed at another resource. Auth (python/obo/app/auth.py): - Inbound tokens validated against Entra's JWKS (signature, aud, iss, scp). - OBO exchange via azure-identity's OnBehalfOfCredential, which is MSAL-backed (it wraps msal.ConfidentialClientApplication underneath). - Managed identity support, for enterprise security postures that disallow deployed secrets: with AZURE_CLIENT_SECRET unset, the service authenticates using a workload-identity federated credential -- DefaultAzureCredential fetches a token for api://AzureADTokenExchange and supplies it as the client assertion, so no secret ever ships. A client secret remains supported for local development. Note that DefaultAzureCredential alone cannot perform the exchange: its chain (managed identity, environment service principal, Azure CLI) issues app-only or developer identities, and WorkIQAgent.Ask is delegated-only. It proves the app's identity; the user assertion supplies the user's. Both are required. Contents: - python/obo/app/config.py -- env config, fails fast on missing values - python/obo/app/auth.py -- JWT validation + OBO exchange - python/obo/app/workiq.py -- async Gateway client (sync + SSE streaming) - python/obo/app/main.py -- FastAPI routes (/api/chat, /api/chat/stream) - python/obo/smoke_test.py -- Gateway faked via httpx.MockTransport - README.md: python/obo/ row added to the sample table; Python 3.10+ added to the toolchain list; callout that scripts/admin-setup.sh does not cover this sample -- it creates a public client, and OBO needs a confidential one that also exposes its own API. Validation: - smoke_test.py passes with no credentials and no network: conversation creation, citation parsing, cumulative-to-delta stream conversion, the 403 path surfacing request-id, and the auth gate (missing and malformed tokens both 401 before body validation). - azure-identity 1.25.3 confirmed to expose client_assertion_func on the async OnBehalfOfCredential. - NOT yet validated end-to-end against a live tenant. That requires a confidential app registration with admin-consented WorkIQAgent.Ask and a Copilot-licensed user; the wire contract was matched against dotnet/rest/ rather than observed from the Gateway. --- README.md | 6 + python/obo/.env.example | 18 +++ python/obo/.gitignore | 1 + python/obo/README.md | 211 ++++++++++++++++++++++++++++++++++++ python/obo/app/__init__.py | 1 + python/obo/app/auth.py | 122 +++++++++++++++++++++ python/obo/app/config.py | 81 ++++++++++++++ python/obo/app/main.py | 164 ++++++++++++++++++++++++++++ python/obo/app/workiq.py | 171 +++++++++++++++++++++++++++++ python/obo/requirements.txt | 5 + python/obo/smoke_test.py | 124 +++++++++++++++++++++ 11 files changed, 904 insertions(+) create mode 100644 python/obo/.env.example create mode 100644 python/obo/.gitignore create mode 100644 python/obo/README.md create mode 100644 python/obo/app/__init__.py create mode 100644 python/obo/app/auth.py create mode 100644 python/obo/app/config.py create mode 100644 python/obo/app/main.py create mode 100644 python/obo/app/workiq.py create mode 100644 python/obo/requirements.txt create mode 100644 python/obo/smoke_test.py diff --git a/README.md b/README.md index ef8383f..a83bb16 100644 --- a/README.md +++ b/README.md @@ -7,9 +7,12 @@ Sample clients for the [Work IQ](https://learn.microsoft.com/en-us/microsoft-365 | [**dotnet/a2a/**](dotnet/a2a/) | C# | Windows, macOS, Linux | [A2A](https://a2a-protocol.org) | Interactive agent session using the A2A protocol over JSON-RPC | | [**dotnet/a2a-raw/**](dotnet/a2a-raw/) | C# | Windows, macOS, Linux | [A2A](https://a2a-protocol.org) | Same, but with raw `HttpClient` + JSON (no A2A SDK) | | [**dotnet/rest/**](dotnet/rest/) | C# | Windows, macOS, Linux | REST | Interactive chat using the [Copilot Chat API](https://learn.microsoft.com/en-us/microsoft-365-copilot/extensibility/api/ai-services/chat/overview) | +| [**python/obo/**](python/obo/) | Python | Windows, macOS, Linux | REST | Backend service brokering Work IQ for a frontend via the [On-Behalf-Of flow](https://learn.microsoft.com/en-us/entra/identity-platform/v2-oauth2-on-behalf-of-flow) | | [**rust/a2a/**](rust/a2a/) | Rust | Windows, macOS, Linux | [A2A](https://a2a-protocol.org) | Interactive agent session with MSAL auth and token caching | | [**swift/a2a/**](swift/a2a/) | Swift | iOS/iPadOS (macOS to build) | [A2A](https://a2a-protocol.org) | SwiftUI chat app for Work IQ | +Every sample is a **public client** that signs a user in directly, except [**python/obo/**](python/obo/) — that one is a **middle tier**: the user signs in to your frontend, and the service exchanges their token for a Work IQ token. It needs a different app registration; see its [README](python/obo/README.md#app-registration). + --- ## Gateway @@ -24,6 +27,7 @@ All samples target the **Work IQ Gateway** (`workiq.svc.cloud.microsoft`) — th 2. **Entra app registration** configured in your tenant — this is a one-time setup per tenant. Details below. 3. **Your language toolchain**: - **dotnet/** samples: [.NET 8.0 SDK](https://dotnet.microsoft.com/download/dotnet/8.0) or later + - **python/** samples: [Python 3.10+](https://www.python.org/downloads/) - **rust/** samples: [Rust toolchain](https://rustup.rs/) (stable) - **swift/** samples: [Xcode 26+](https://developer.apple.com/xcode/) (macOS only) @@ -46,6 +50,8 @@ You (or your tenant admin) must create an Entra app registration with specific p After setup you'll have two values: `APP_ID` and `TENANT_ID`. Pass them to any sample via `--appid` and `--tenant`. +> **[`python/obo/`](python/obo/) is the exception.** The script above creates a *public* client; the On-Behalf-Of flow requires a *confidential* client that also exposes its own API. See [its README](python/obo/README.md#app-registration) for the separate setup. + --- ## Authentication methods diff --git a/python/obo/.env.example b/python/obo/.env.example new file mode 100644 index 0000000..2a1313d --- /dev/null +++ b/python/obo/.env.example @@ -0,0 +1,18 @@ +# Backend app registration (confidential client). +AZURE_TENANT_ID=00000000-0000-0000-0000-000000000000 +AZURE_CLIENT_ID=00000000-0000-0000-0000-000000000000 + +# Application ID URI this backend's tokens are issued for. The frontend must +# request `${API_AUDIENCE}/access_as_user`. +API_AUDIENCE=api://00000000-0000-0000-0000-000000000000 + +# Scope the inbound token must carry. Defaults to access_as_user. +# REQUIRED_SCOPE=access_as_user + +# Leave AZURE_CLIENT_SECRET unset in Azure: the app then authenticates with a +# managed identity federated credential via DefaultAzureCredential (no secret). +# Set it only for local development. +# AZURE_CLIENT_SECRET= + +# Override the gateway host for testing. Defaults to the production gateway. +# WORKIQ_HOST=https://workiq.svc.cloud.microsoft diff --git a/python/obo/.gitignore b/python/obo/.gitignore new file mode 100644 index 0000000..21d0b89 --- /dev/null +++ b/python/obo/.gitignore @@ -0,0 +1 @@ +.venv/ diff --git a/python/obo/README.md b/python/obo/README.md new file mode 100644 index 0000000..3ba4238 --- /dev/null +++ b/python/obo/README.md @@ -0,0 +1,211 @@ +# Work IQ Python OBO Sample + +A FastAPI **middle-tier** service that accepts your frontend's access token, exchanges it for a Work IQ token via the [On-Behalf-Of flow](https://learn.microsoft.com/en-us/entra/identity-platform/v2-oauth2-on-behalf-of-flow), and calls the [Copilot Chat REST API](https://learn.microsoft.com/en-us/microsoft-365-copilot/extensibility/api/ai-services/chat/overview) through the **Work IQ Gateway**. + +Every other sample in this repo is a **public client** that signs a user in directly. This one is different: the user signs in to *your* frontend, and this service brokers the Work IQ call for them. + +``` +Frontend ──token(aud=your API)──▶ This service ──token(aud=Work IQ)──▶ Gateway + │ + └── validate ──▶ OBO exchange +``` + +Supports both **synchronous** and **streaming** (SSE) modes. + +## API reference + +| Operation | Method | Path | Docs | +|-----------|--------|------|------| +| Create conversation | `POST` | `/rest/beta/conversations` | [Docs](https://learn.microsoft.com/en-us/microsoft-365-copilot/extensibility/api/ai-services/chat/copilotroot-post-conversations) | +| Chat (sync) | `POST` | `/rest/beta/conversations/{id}/chat` | [Docs](https://learn.microsoft.com/en-us/microsoft-365-copilot/extensibility/api/ai-services/chat/copilotconversation-chat) | +| Chat (stream) | `POST` | `/rest/beta/conversations/{id}/chatOverStream` | [Docs](https://learn.microsoft.com/en-us/microsoft-365-copilot/extensibility/api/ai-services/chat/copilotconversation-chatoverstream) | + +## Why you can't just forward the frontend's token + +A token's audience is baked into the signed JWT. The token your frontend holds has +`aud` set to *your* API — the Gateway rejects it with `401`. You cannot re-point it at +another resource; you exchange it for a new one. That exchange is OBO. + +**`DefaultAzureCredential` cannot do this exchange.** Its chain (managed identity, +environment service principal, Azure CLI) issues tokens for the *app's own* identity or +a *developer's* identity. None of them accept an inbound user token and re-issue it for +a different resource. And `WorkIQAgent.Ask` is delegated-only — an app-only token gets +you nothing, because responses depend on the signed-in user's Copilot license and their +own data access. + +`azure.identity.aio.OnBehalfOfCredential` is the credential that does this. + +`DefaultAzureCredential` still has a real job here: proving the *app's* identity so no +client secret is ever deployed. Leave `AZURE_CLIENT_SECRET` unset and the service uses a +managed identity federated credential as its client assertion — see [`app/auth.py`](app/auth.py): + +```python +def assertion() -> str: + return credential.get_token("api://AzureADTokenExchange/.default").token + +OnBehalfOfCredential( + tenant_id=..., client_id=..., + client_assertion_func=assertion, # app identity — DefaultAzureCredential + user_assertion=inbound_token, # user identity — from your frontend +) +``` + +Both identities are required: the app proves it is allowed to ask, the user token +determines what comes back. + +## Prerequisites + +1. **Microsoft 365 Copilot license** on your test user. +2. **Two Entra app registrations.** `scripts/admin-setup.sh` at the repo root does **not** + cover this sample — it creates a *public* client for the CLI samples, and OBO requires + a *confidential* client. See [App registration](#app-registration) below. +3. **Python 3.10+**. + +## App registration + +**Frontend app** (SPA / public client) — requests `api:///access_as_user`. + +**Backend app** (this service — confidential client): + +| Blade | Setting | +|-------|---------| +| Expose an API | Application ID URI `api://`, scope `access_as_user` | +| API permissions | `Work IQ` → delegated `WorkIQAgent.Ask` → **Grant admin consent** | +| Certificates & secrets | In Azure, prefer a **federated credential** bound to your managed identity over a client secret | + +```bash +# Ensure the Work IQ service principal exists in your tenant +az ad sp create --id fdcc1f02-fc51-4226-8753-f668596af7f7 + +# Grant this service the delegated Work IQ permission, then consent +az ad app permission add --id \ + --api fdcc1f02-fc51-4226-8753-f668596af7f7 \ + --api-permissions "0b1715fd-f4bf-4c63-b16d-5be31f9847c2=Scope" +az ad app permission admin-consent --id +``` + +## Quick start + +```bash +python -m venv .venv && source .venv/bin/activate +pip install -r requirements.txt + +cp .env.example .env # fill in tenant, client id, audience +set -a && source .env && set +a + +uvicorn app.main:app --reload +``` + +```bash +curl -X POST localhost:8000/api/chat \ + -H "Authorization: Bearer /access_as_user>" \ + -H "Content-Type: application/json" \ + -d '{"message": "What meetings do I have tomorrow?"}' +``` + +```json +{ + "conversation_id": "conv-123", + "text": "You have 3 meetings scheduled...", + "citations": [ + { "type": "citation", "source": "sharepoint", "provider": "Q3.docx", "url": "https://..." } + ] +} +``` + +Pass `conversation_id` back on later turns to continue the same conversation with full context. + +### Verify without credentials + +```bash +python smoke_test.py +``` + +Fakes the Gateway with `httpx.MockTransport` and checks conversation creation, citation +parsing, streaming deltas, the `403` error path, and the auth gate. No tenant, no license, +no network. + +## Endpoints + +| Endpoint | Mode | Response | +|----------|------|----------| +| `POST /api/chat` | Synchronous | JSON — `conversation_id`, `text`, `citations` | +| `POST /api/chat/stream` | SSE | `event: conversation` then `data: {"text": ""}` frames | + +Request body for both: `{"message": "...", "conversation_id": "..."}` (`conversation_id` optional). + +## Layout + +| File | Purpose | +|------|---------| +| [`app/config.py`](app/config.py) | Environment config; fails fast on missing values | +| [`app/auth.py`](app/auth.py) | Inbound JWT validation + the OBO exchange | +| [`app/workiq.py`](app/workiq.py) | Async Gateway client (sync + streaming) | +| [`app/main.py`](app/main.py) | FastAPI routes | +| [`smoke_test.py`](smoke_test.py) | Gateway faked via `httpx.MockTransport` | + +## How it works + +``` +Frontend This service Gateway + | | | + |-- POST /api/chat --->| | + | Bearer | | + | |-- validate signature/aud/scp | + | | against Entra JWKS | + | | | + | |-- OBO exchange ---▶ Entra | + | |◀-- token(aud=Work IQ) -- | + | | | + | |-- POST .../conversations ----->| + | |<-- 201 { "id": "conv-123" } ---| + | |-- POST .../conv-123/chat ----->| + | |<-- 200 { "messages": [...] } --| + |<-- 200 JSON ---------| | +``` + +Each SSE event from the Gateway contains the **full conversation state** (cumulative, not +incremental). [`app/workiq.py`](app/workiq.py) diffs against the previous event and yields +only new text, so `/api/chat/stream` emits true deltas. + +Auth is resolved before body validation, so an unauthenticated caller gets `401` and +learns nothing about the request schema. + +## Dependencies + +| Package | Purpose | +|---------|---------| +| `azure-identity` | `OnBehalfOfCredential` for the exchange; `DefaultAzureCredential` for the app assertion | +| `pyjwt[crypto]` | Validating the inbound token against Entra's JWKS | +| `httpx` | Async HTTP + SSE against the Gateway | +| `fastapi` / `uvicorn` | The service itself | + +No MSAL wrapper needed — `azure-identity` builds on MSAL underneath. + +## Sample-specific troubleshooting + +| Symptom | Fix | +|---------|-----| +| `401` from this service | Inbound token failed validation. Check `aud` matches `API_AUDIENCE` and `scp` includes `access_as_user`. | +| `403 Unable to obtain Work IQ access` | The OBO exchange failed. Usually missing admin consent on `WorkIQAgent.Ask`, or the user lacks a Copilot license. | +| `401` from the Gateway | The *outbound* token's `aud` is wrong — must be `api://workiq.svc.cloud.microsoft`, not your API. | +| `AADSTS50013: Assertion failed signature validation` | The federated credential subject/issuer doesn't match your managed identity. | +| Slow first response per turn | Every request does a fresh OBO round trip. See [Notes before production](#notes-before-production). | + +See the [root README](../../README.md#troubleshooting) for the full matrix (Copilot license, consent, audience mismatch). + +## Notes before production + +- **Cache the OBO result.** Each request currently exchanges the token again. Cache on a + hash of the inbound token, honoring `expires_on`, to save a round trip per turn. +- **The `request-id` response header** is what Microsoft support asks for. + [`app/workiq.py`](app/workiq.py) logs it but keeps it out of client-facing errors. +- **Never log tokens.** Errors here are logged with the user's `oid`, not the assertion. + +## Resources + +- [On-Behalf-Of flow](https://learn.microsoft.com/en-us/entra/identity-platform/v2-oauth2-on-behalf-of-flow) +- [Workload identity federation](https://learn.microsoft.com/en-us/entra/workload-id/workload-identity-federation) +- [`azure-identity` for Python](https://learn.microsoft.com/en-us/python/api/overview/azure/identity-readme) +- [Chat API Overview](https://learn.microsoft.com/en-us/microsoft-365-copilot/extensibility/api/ai-services/chat/overview) +- [Work IQ Overview](https://learn.microsoft.com/en-us/microsoft-365/copilot/extensibility/workiq-overview) diff --git a/python/obo/app/__init__.py b/python/obo/app/__init__.py new file mode 100644 index 0000000..3231871 --- /dev/null +++ b/python/obo/app/__init__.py @@ -0,0 +1 @@ +"""Work IQ On-Behalf-Of backend sample.""" diff --git a/python/obo/app/auth.py b/python/obo/app/auth.py new file mode 100644 index 0000000..73f7029 --- /dev/null +++ b/python/obo/app/auth.py @@ -0,0 +1,122 @@ +"""Validate the frontend's token, then exchange it for a Work IQ token via OBO. + +Two distinct tokens are involved and they are not interchangeable: + + 1. Inbound — aud = this backend's Application ID URI (API_AUDIENCE). + Issued to the frontend. Work IQ rejects it (401, audience mismatch). + 2. Outbound — aud = api://workiq.svc.cloud.microsoft. Minted here by the + On-Behalf-Of flow, carrying the same user identity. +""" + +from __future__ import annotations + +import asyncio +from typing import Any, Callable + +import jwt +from azure.identity import DefaultAzureCredential as SyncDefaultAzureCredential +from azure.identity.aio import OnBehalfOfCredential +from jwt import PyJWKClient + +from .config import TOKEN_EXCHANGE_SCOPE, WORKIQ_SCOPE, Settings + + +class InvalidToken(Exception): + """The inbound token failed signature, audience, issuer, or scope checks.""" + + +class TokenValidator: + """Validates inbound tokens against Entra's published signing keys. + + PyJWKClient caches keys after the first fetch, so the blocking call it makes + is offloaded to a worker thread rather than stalling the event loop. + """ + + def __init__(self, settings: Settings) -> None: + self._settings = settings + self._jwks_client = PyJWKClient(settings.jwks_uri, cache_keys=True) + + async def validate(self, token: str) -> dict[str, Any]: + try: + signing_key = await asyncio.to_thread( + self._jwks_client.get_signing_key_from_jwt, token + ) + claims: dict[str, Any] = jwt.decode( + token, + signing_key.key, + algorithms=["RS256"], + audience=self._settings.api_audience, + issuer=self._settings.issuer, + options={"require": ["exp", "aud", "iss"]}, + ) + except jwt.PyJWTError as exc: + raise InvalidToken(f"token rejected: {exc}") from exc + + self._require_scope(claims) + return claims + + def _require_scope(self, claims: dict[str, Any]) -> None: + granted = set(str(claims.get("scp", "")).split()) + if self._settings.required_scope not in granted: + raise InvalidToken( + f"token is missing the required scope '{self._settings.required_scope}'" + ) + + +def _federated_assertion(credential: SyncDefaultAzureCredential) -> Callable[[], str]: + """Build a client assertion from a managed identity token. + + This is where DefaultAzureCredential belongs in an OBO backend: it proves the + *app's* identity so no client secret is needed. It cannot perform the OBO + exchange itself — that needs the user's assertion, which only OnBehalfOfCredential + accepts. The callable is sync because azure-identity invokes it synchronously. + """ + + def assertion() -> str: + return credential.get_token(TOKEN_EXCHANGE_SCOPE).token + + return assertion + + +class WorkIQTokenExchange: + """Mints Work IQ access tokens on behalf of the calling user.""" + + def __init__(self, settings: Settings) -> None: + self._settings = settings + self._mi_credential = ( + SyncDefaultAzureCredential() if settings.uses_managed_identity else None + ) + + async def token_for(self, user_assertion: str) -> str: + """Exchange a validated inbound token for a Work IQ access token. + + A fresh credential per request keeps user identities isolated. The trade-off + is a round trip to Entra on every call — add a cache keyed by a hash of the + assertion if that latency matters. + """ + credential = self._build_credential(user_assertion) + async with credential: + access_token = await credential.get_token(WORKIQ_SCOPE) + return access_token.token + + def _build_credential(self, user_assertion: str) -> OnBehalfOfCredential: + common = { + "tenant_id": self._settings.tenant_id, + "client_id": self._settings.client_id, + "user_assertion": user_assertion, + } + + if self._mi_credential is not None: + return OnBehalfOfCredential( + client_assertion_func=_federated_assertion(self._mi_credential), + **common, + ) + + return OnBehalfOfCredential( + client_secret=self._settings.client_secret, + **common, + ) + + async def close(self) -> None: + if self._mi_credential is not None: + self._mi_credential.close() diff --git a/python/obo/app/config.py b/python/obo/app/config.py new file mode 100644 index 0000000..1014025 --- /dev/null +++ b/python/obo/app/config.py @@ -0,0 +1,81 @@ +"""Configuration for the Work IQ OBO backend, loaded from the environment. + +Missing required values fail at import time rather than on the first request. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from functools import lru_cache + +# The Work IQ Gateway multiplexes REST/A2A/MCP on one host behind a path prefix +# (/rest, /a2a, /tools). Only the REST surface is used here. +WORKIQ_DEFAULT_HOST = "https://workiq.svc.cloud.microsoft" +WORKIQ_PATH = "/rest/beta" + +# Work IQ resource. WorkIQAgent.Ask is granted by admin consent on the app +# registration, so `.default` resolves to it without naming the scope here. +WORKIQ_SCOPE = "api://workiq.svc.cloud.microsoft/.default" + +# Scope the frontend must present on its token to this backend. +DEFAULT_REQUIRED_SCOPE = "access_as_user" + +ENTRA_AUTHORITY = "https://login.microsoftonline.com" + +# Audience used when exchanging a managed identity token for a client assertion +# (workload identity federation). Only used in the secretless configuration. +TOKEN_EXCHANGE_SCOPE = "api://AzureADTokenExchange/.default" + + +class ConfigError(RuntimeError): + """A required environment variable is missing or malformed.""" + + +@dataclass(frozen=True) +class Settings: + """Immutable runtime configuration.""" + + tenant_id: str + client_id: str + api_audience: str + required_scope: str + workiq_host: str + client_secret: str | None + + @property + def issuer(self) -> str: + return f"{ENTRA_AUTHORITY}/{self.tenant_id}/v2.0" + + @property + def jwks_uri(self) -> str: + return f"{ENTRA_AUTHORITY}/{self.tenant_id}/discovery/v2.0/keys" + + @property + def workiq_base(self) -> str: + return f"{self.workiq_host.rstrip('/')}{WORKIQ_PATH}" + + @property + def uses_managed_identity(self) -> bool: + """True when no secret is configured, so a federated assertion is used.""" + return self.client_secret is None + + +def _require(name: str) -> str: + value = os.environ.get(name, "").strip() + if not value: + raise ConfigError(f"{name} must be set") + return value + + +@lru_cache(maxsize=1) +def get_settings() -> Settings: + secret = os.environ.get("AZURE_CLIENT_SECRET", "").strip() + return Settings( + tenant_id=_require("AZURE_TENANT_ID"), + client_id=_require("AZURE_CLIENT_ID"), + api_audience=_require("API_AUDIENCE"), + required_scope=os.environ.get("REQUIRED_SCOPE", DEFAULT_REQUIRED_SCOPE), + workiq_host=os.environ.get("WORKIQ_HOST", WORKIQ_DEFAULT_HOST), + client_secret=secret or None, + ) diff --git a/python/obo/app/main.py b/python/obo/app/main.py new file mode 100644 index 0000000..8abda5e --- /dev/null +++ b/python/obo/app/main.py @@ -0,0 +1,164 @@ +"""FastAPI backend that brokers Work IQ calls on behalf of the signed-in user. + +Flow per request: + frontend token -> validate -> OBO exchange -> Work IQ token -> call gateway +""" + +from __future__ import annotations + +import json +import logging +from contextlib import asynccontextmanager +from typing import Annotated, AsyncIterator + +from fastapi import Depends, FastAPI, Header, HTTPException, status +from fastapi.responses import StreamingResponse +from pydantic import BaseModel, Field + +from .auth import InvalidToken, TokenValidator, WorkIQTokenExchange +from .config import Settings, get_settings +from .workiq import WorkIQClient, WorkIQError + +logger = logging.getLogger(__name__) + +BEARER_PREFIX = "Bearer " + + +class ChatRequest(BaseModel): + message: str = Field(min_length=1, max_length=8000) + conversation_id: str | None = None + + +class CitationModel(BaseModel): + type: str + source: str + provider: str + url: str + + +class ChatResponse(BaseModel): + conversation_id: str + text: str + citations: list[CitationModel] + + +@asynccontextmanager +async def lifespan(app: FastAPI) -> AsyncIterator[None]: + settings = get_settings() + app.state.settings = settings + app.state.validator = TokenValidator(settings) + app.state.exchange = WorkIQTokenExchange(settings) + logger.info( + "ready — auth=%s, workiq=%s", + "managed-identity" if settings.uses_managed_identity else "client-secret", + settings.workiq_base, + ) + try: + yield + finally: + await app.state.exchange.close() + + +app = FastAPI(title="Work IQ OBO Backend", lifespan=lifespan) + + +def _bearer_token(authorization: Annotated[str | None, Header()] = None) -> str: + if not authorization or not authorization.startswith(BEARER_PREFIX): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Missing bearer token", + headers={"WWW-Authenticate": "Bearer"}, + ) + return authorization[len(BEARER_PREFIX) :].strip() + + +async def workiq_token( + inbound_token: Annotated[str, Depends(_bearer_token)], +) -> str: + """Validate the caller's token and exchange it for a Work IQ token.""" + validator: TokenValidator = app.state.validator + exchange: WorkIQTokenExchange = app.state.exchange + + try: + claims = await validator.validate(inbound_token) + except InvalidToken as exc: + logger.warning("rejected inbound token: %s", exc) + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid token", + headers={"WWW-Authenticate": "Bearer"}, + ) from exc + + try: + return await exchange.token_for(inbound_token) + except Exception as exc: + # A failure here is usually consent, licensing, or a misconfigured app + # registration — log it with the user for triage, but do not leak it. + logger.error("OBO exchange failed for %s: %s", claims.get("oid"), exc) + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Unable to obtain Work IQ access for this user", + ) from exc + + +def _client(token: str, settings: Settings) -> WorkIQClient: + return WorkIQClient(access_token=token, base_url=settings.workiq_base) + + +@app.post("/api/chat", response_model=ChatResponse) +async def chat( + request: ChatRequest, + token: Annotated[str, Depends(workiq_token)], +) -> ChatResponse: + settings: Settings = app.state.settings + + try: + async with _client(token, settings) as client: + conversation_id = request.conversation_id or await client.create_conversation() + reply = await client.chat(conversation_id, request.message) + except WorkIQError as exc: + logger.error("work iq call failed: %s", exc) + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, detail="Work IQ request failed" + ) from exc + + return ChatResponse( + conversation_id=conversation_id, + text=reply.text, + citations=[ + CitationModel( + type=c.attribution_type, + source=c.attribution_source, + provider=c.provider_display_name, + url=c.see_more_web_url, + ) + for c in reply.citations + ], + ) + + +@app.post("/api/chat/stream") +async def chat_stream( + request: ChatRequest, + token: Annotated[str, Depends(workiq_token)], +) -> StreamingResponse: + settings: Settings = app.state.settings + + async def events() -> AsyncIterator[str]: + try: + async with _client(token, settings) as client: + conversation_id = ( + request.conversation_id or await client.create_conversation() + ) + yield f"event: conversation\ndata: {conversation_id}\n\n" + + # JSON-encode each delta: raw newlines in the text would + # otherwise terminate the SSE frame early. + async for delta in client.chat_stream(conversation_id, request.message): + yield f"data: {json.dumps({'text': delta})}\n\n" + except WorkIQError as exc: + # The response has already started, so the error rides the stream. + logger.error("work iq stream failed: %s", exc) + yield "event: error\ndata: Work IQ request failed\n\n" + + return StreamingResponse(events(), media_type="text/event-stream") diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py new file mode 100644 index 0000000..3b769b7 --- /dev/null +++ b/python/obo/app/workiq.py @@ -0,0 +1,171 @@ +"""Async client for the Work IQ Gateway REST surface. + +Mirrors the wire contract exercised by dotnet/rest/Program.cs: + POST /rest/beta/conversations -> {"id": ...} + POST /rest/beta/conversations/{id}/chat -> {"messages": [...]} + POST /rest/beta/conversations/{id}/chatOverStream -> SSE, cumulative text +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, AsyncIterator + +import httpx + +REQUEST_TIMEOUT_SECONDS = 300.0 +SSE_DATA_PREFIX = "data: " + + +class WorkIQError(Exception): + """Work IQ returned an error or an unparseable response.""" + + +@dataclass(frozen=True) +class Citation: + attribution_type: str + attribution_source: str + provider_display_name: str + see_more_web_url: str + + +@dataclass(frozen=True) +class ChatReply: + text: str + citations: tuple[Citation, ...] = field(default=()) + + +def _local_timezone() -> str: + """Work IQ requires an IANA timezone (e.g. America/Los_Angeles).""" + tz = datetime.now().astimezone().tzinfo + name = getattr(tz, "key", None) or str(tz) + # A bare UTC offset like "+05:30" is not an IANA id; fall back rather than 400. + return name if "/" in name or name == "UTC" else "UTC" + + +def _chat_body(message: str) -> dict[str, Any]: + return { + "message": {"text": message}, + "locationHint": {"timeZone": _local_timezone()}, + } + + +def _parse_citations(message: dict[str, Any]) -> tuple[Citation, ...]: + return tuple( + Citation( + attribution_type=a.get("attributionType", ""), + attribution_source=a.get("attributionSource", ""), + provider_display_name=a.get("providerDisplayName", ""), + see_more_web_url=a.get("seeMoreWebUrl", ""), + ) + for a in message.get("attributions", []) + ) + + +def _last_text_message(payload: dict[str, Any]) -> dict[str, Any] | None: + """The assistant's reply is the last message carrying a `text` field.""" + candidates = [m for m in payload.get("messages", []) if "text" in m] + return candidates[-1] if candidates else None + + +class WorkIQClient: + """One instance per user request — it is bound to that user's token.""" + + def __init__( + self, + access_token: str, + base_url: str, + transport: httpx.AsyncBaseTransport | None = None, + ) -> None: + self._client = httpx.AsyncClient( + base_url=base_url, + headers={"Authorization": f"Bearer {access_token}"}, + timeout=REQUEST_TIMEOUT_SECONDS, + transport=transport, + ) + + async def __aenter__(self) -> "WorkIQClient": + return self + + async def __aexit__(self, *_: Any) -> None: + await self._client.aclose() + + async def create_conversation(self) -> str: + response = await self._client.post("/conversations", json={}) + self._raise_for_status(response, "create conversation") + + conversation_id = response.json().get("id") + if not conversation_id: + raise WorkIQError("no conversation id in response") + return conversation_id + + async def chat(self, conversation_id: str, message: str) -> ChatReply: + response = await self._client.post( + f"/conversations/{conversation_id}/chat", json=_chat_body(message) + ) + self._raise_for_status(response, "chat") + + reply = _last_text_message(response.json()) + if reply is None: + raise WorkIQError("no assistant message in response") + + return ChatReply(text=reply["text"], citations=_parse_citations(reply)) + + async def chat_stream( + self, conversation_id: str, message: str + ) -> AsyncIterator[str]: + """Yield text deltas as they arrive. + + The gateway streams cumulative, append-only text, so each event is diffed + against the previous one. A non-prefix update would re-emit the full text; + that does not happen under the current append-only contract. + """ + request = self._client.build_request( + "POST", + f"/conversations/{conversation_id}/chatOverStream", + json=_chat_body(message), + ) + response = await self._client.send(request, stream=True) + try: + if response.status_code >= 400: + await response.aread() + self._raise_for_status(response, "chat stream") + + previous = "" + async for line in response.aiter_lines(): + if not line.startswith(SSE_DATA_PREFIX): + continue + + event = line[len(SSE_DATA_PREFIX) :].strip() + if not event: + continue + + try: + reply = _last_text_message(json.loads(event)) + except json.JSONDecodeError: + continue # Skip malformed events rather than kill the stream. + + if reply is None: + continue + + text = reply.get("text", "") + delta = text[len(previous) :] if text.startswith(previous) else text + previous = text + if delta: + yield delta + finally: + await response.aclose() + + @staticmethod + def _raise_for_status(response: httpx.Response, action: str) -> None: + if response.status_code < 400: + return + + # request-id is what Microsoft support asks for; keep it out of client replies. + request_id = response.headers.get("request-id", "unknown") + raise WorkIQError( + f"{action} failed: {response.status_code} " + f"(request-id={request_id}) {response.text[:500]}" + ) diff --git a/python/obo/requirements.txt b/python/obo/requirements.txt new file mode 100644 index 0000000..cb674f1 --- /dev/null +++ b/python/obo/requirements.txt @@ -0,0 +1,5 @@ +fastapi>=0.115 +uvicorn[standard]>=0.32 +httpx>=0.27 +azure-identity>=1.19 +pyjwt[crypto]>=2.9 diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py new file mode 100644 index 0000000..9127259 --- /dev/null +++ b/python/obo/smoke_test.py @@ -0,0 +1,124 @@ +"""Smoke test: fakes the Work IQ gateway with httpx.MockTransport.""" + +import asyncio +import json +import os +import sys +from pathlib import Path + +os.environ.update( + AZURE_TENANT_ID="11111111-1111-1111-1111-111111111111", + AZURE_CLIENT_ID="22222222-2222-2222-2222-222222222222", + API_AUDIENCE="api://22222222-2222-2222-2222-222222222222", + AZURE_CLIENT_SECRET="local-dev-secret", +) + +sys.path.insert(0, str(Path(__file__).parent)) + +import httpx +from fastapi.testclient import TestClient + +from app.workiq import WorkIQClient, WorkIQError, _last_text_message + +BASE = "https://workiq.test/rest/beta" + + +def handler(request: httpx.Request) -> httpx.Response: + path = request.url.path + if path.endswith("/conversations"): + return httpx.Response(200, json={"id": "conv-42"}) + if path.endswith("/chat"): + return httpx.Response( + 200, + json={ + "messages": [ + {"text": "ignored earlier turn"}, + { + "text": "Your Q3 report is ready.", + "attributions": [ + { + "attributionType": "citation", + "attributionSource": "sharepoint", + "providerDisplayName": "Q3.docx", + "seeMoreWebUrl": "https://contoso.example/q3", + } + ], + }, + ] + }, + ) + if path.endswith("/chatOverStream"): + # Cumulative, append-only text, exactly as the gateway streams it. + frames = ["Hello", "Hello there", "Hello there world"] + body = "".join( + f"data: {json.dumps({'messages': [{'text': f}]})}\n" for f in frames + ) + return httpx.Response(200, text=body) + return httpx.Response(500, json={"error": "unexpected path"}) + + +def error_handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(403, json={"error": "no copilot license"}, headers={"request-id": "abc-123"}) + + +async def main() -> None: + transport = httpx.MockTransport(handler) + + async with WorkIQClient("fake-token", BASE, transport=transport) as client: + conv = await client.create_conversation() + assert conv == "conv-42", conv + print("create_conversation ->", conv) + + reply = await client.chat(conv, "what's up") + assert reply.text == "Your Q3 report is ready.", reply.text + assert len(reply.citations) == 1 + assert reply.citations[0].provider_display_name == "Q3.docx" + print("chat (last text message) ->", reply.text) + print("citations ->", reply.citations[0].provider_display_name) + + deltas = [d async for d in client.chat_stream(conv, "stream please")] + assert deltas == ["Hello", " there", " world"], deltas + assert "".join(deltas) == "Hello there world" + print("stream deltas ->", deltas) + + # Error path: 403 must surface request-id and not raise something unrelated. + async with WorkIQClient("t", BASE, transport=httpx.MockTransport(error_handler)) as c: + try: + await c.create_conversation() + raise AssertionError("expected WorkIQError") + except WorkIQError as exc: + assert "request-id=abc-123" in str(exc), exc + print("403 error path ->", str(exc)[:60]) + + # Pure helper + assert _last_text_message({"messages": []}) is None + assert _last_text_message({"messages": [{"role": "user"}]}) is None + print("helper edge cases -> ok") + + # Auth gate: no token and malformed token must both 401 before any Work IQ call. + from app.main import app + + with TestClient(app) as tc: + r = tc.post("/api/chat", json={"message": "hi"}) + assert r.status_code == 401, r.status_code + print("no bearer token ->", r.status_code, r.json()["detail"]) + + r = tc.post( + "/api/chat", + json={"message": "hi"}, + headers={"Authorization": "Bearer not-a-jwt"}, + ) + assert r.status_code == 401, r.status_code + print("malformed token ->", r.status_code, r.json()["detail"]) + + # Auth runs before body validation, so a bad token wins over a bad body. + # That ordering is deliberate: unauthenticated callers learn nothing + # about the request schema. + r = tc.post("/api/chat", json={"message": ""}, headers={"Authorization": "Bearer x"}) + assert r.status_code == 401, r.status_code + print("auth precedes validation ->", r.status_code) + + print("\nAll checks passed.") + + +asyncio.run(main()) From 2c7350820738a48adbadb569db3df877ad1b01ee Mon Sep 17 00:00:00 2001 From: Shakir Fattani Date: Thu, 16 Jul 2026 23:24:32 +0400 Subject: [PATCH 02/36] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- python/obo/app/workiq.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 3b769b7..e4da0dd 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -124,7 +124,7 @@ async def chat_stream( """ request = self._client.build_request( "POST", - f"/conversations/{conversation_id}/chatOverStream", + f"conversations/{conversation_id}/chatOverStream", json=_chat_body(message), ) response = await self._client.send(request, stream=True) From 54473b42bc9d6257db70d387cf5ae4ea035241a1 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Thu, 16 Jul 2026 23:35:24 +0400 Subject: [PATCH 03/36] fix: resolve remaining Copilot review findings on Python OBO sample Address all 6 review comments from Copilot on microsoft/work-iq-samples#23: - Use relative URL paths (no leading /) consistently across all WorkIQClient methods, not just chatOverStream - Add trailing slash to workiq_base so relative paths resolve correctly - Fix config.py docstring: settings fail at get_settings() call, not import - Strip whitespace from REQUIRED_SCOPE and WORKIQ_HOST env vars - Make Bearer scheme check case-insensitive per RFC 9110 - Update smoke_test BASE to match trailing-slash convention Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/config.py | 9 +++++---- python/obo/app/main.py | 6 +++--- python/obo/app/workiq.py | 4 ++-- python/obo/smoke_test.py | 2 +- 4 files changed, 11 insertions(+), 10 deletions(-) diff --git a/python/obo/app/config.py b/python/obo/app/config.py index 1014025..81d7890 100644 --- a/python/obo/app/config.py +++ b/python/obo/app/config.py @@ -1,6 +1,7 @@ """Configuration for the Work IQ OBO backend, loaded from the environment. -Missing required values fail at import time rather than on the first request. +Missing required values fail at startup (when ``get_settings()`` is first called) +rather than on the first request. """ from __future__ import annotations @@ -53,7 +54,7 @@ def jwks_uri(self) -> str: @property def workiq_base(self) -> str: - return f"{self.workiq_host.rstrip('/')}{WORKIQ_PATH}" + return f"{self.workiq_host.rstrip('/')}{WORKIQ_PATH}/" @property def uses_managed_identity(self) -> bool: @@ -75,7 +76,7 @@ def get_settings() -> Settings: tenant_id=_require("AZURE_TENANT_ID"), client_id=_require("AZURE_CLIENT_ID"), api_audience=_require("API_AUDIENCE"), - required_scope=os.environ.get("REQUIRED_SCOPE", DEFAULT_REQUIRED_SCOPE), - workiq_host=os.environ.get("WORKIQ_HOST", WORKIQ_DEFAULT_HOST), + required_scope=os.environ.get("REQUIRED_SCOPE", DEFAULT_REQUIRED_SCOPE).strip(), + workiq_host=os.environ.get("WORKIQ_HOST", WORKIQ_DEFAULT_HOST).strip(), client_secret=secret or None, ) diff --git a/python/obo/app/main.py b/python/obo/app/main.py index 8abda5e..7ee40f7 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -21,7 +21,7 @@ logger = logging.getLogger(__name__) -BEARER_PREFIX = "Bearer " +BEARER_SCHEME = "bearer" class ChatRequest(BaseModel): @@ -63,13 +63,13 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: def _bearer_token(authorization: Annotated[str | None, Header()] = None) -> str: - if not authorization or not authorization.startswith(BEARER_PREFIX): + if not authorization or not authorization.lower().startswith(BEARER_SCHEME): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token", headers={"WWW-Authenticate": "Bearer"}, ) - return authorization[len(BEARER_PREFIX) :].strip() + return authorization[len(BEARER_SCHEME) :].strip() async def workiq_token( diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index e4da0dd..6673bac 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -93,7 +93,7 @@ async def __aexit__(self, *_: Any) -> None: await self._client.aclose() async def create_conversation(self) -> str: - response = await self._client.post("/conversations", json={}) + response = await self._client.post("conversations", json={}) self._raise_for_status(response, "create conversation") conversation_id = response.json().get("id") @@ -103,7 +103,7 @@ async def create_conversation(self) -> str: async def chat(self, conversation_id: str, message: str) -> ChatReply: response = await self._client.post( - f"/conversations/{conversation_id}/chat", json=_chat_body(message) + f"conversations/{conversation_id}/chat", json=_chat_body(message) ) self._raise_for_status(response, "chat") diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py index 9127259..893cae2 100644 --- a/python/obo/smoke_test.py +++ b/python/obo/smoke_test.py @@ -20,7 +20,7 @@ from app.workiq import WorkIQClient, WorkIQError, _last_text_message -BASE = "https://workiq.test/rest/beta" +BASE = "https://workiq.test/rest/beta/" def handler(request: httpx.Request) -> httpx.Response: From ed8f911ed6e0a2f3bac24948166823e3661311e0 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Thu, 16 Jul 2026 23:49:47 +0400 Subject: [PATCH 04/36] fix: address second-round Copilot review findings - Bearer token parsing: split on whitespace and require exactly scheme + token, rejecting "BearerXYZ" (no space) and bare "Bearer" with no value - Config env vars: treat whitespace-only REQUIRED_SCOPE and WORKIQ_HOST as unset and fall back to defaults instead of passing empty strings - Error messages: remove response body from WorkIQError to avoid leaking potentially sensitive data into logs; keep status + request-id only Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/config.py | 4 ++-- python/obo/app/main.py | 11 +++++++++-- python/obo/app/workiq.py | 2 +- 3 files changed, 12 insertions(+), 5 deletions(-) diff --git a/python/obo/app/config.py b/python/obo/app/config.py index 81d7890..cb8dbe4 100644 --- a/python/obo/app/config.py +++ b/python/obo/app/config.py @@ -76,7 +76,7 @@ def get_settings() -> Settings: tenant_id=_require("AZURE_TENANT_ID"), client_id=_require("AZURE_CLIENT_ID"), api_audience=_require("API_AUDIENCE"), - required_scope=os.environ.get("REQUIRED_SCOPE", DEFAULT_REQUIRED_SCOPE).strip(), - workiq_host=os.environ.get("WORKIQ_HOST", WORKIQ_DEFAULT_HOST).strip(), + required_scope=os.environ.get("REQUIRED_SCOPE", "").strip() or DEFAULT_REQUIRED_SCOPE, + workiq_host=os.environ.get("WORKIQ_HOST", "").strip() or WORKIQ_DEFAULT_HOST, client_secret=secret or None, ) diff --git a/python/obo/app/main.py b/python/obo/app/main.py index 7ee40f7..560e1df 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -63,13 +63,20 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: def _bearer_token(authorization: Annotated[str | None, Header()] = None) -> str: - if not authorization or not authorization.lower().startswith(BEARER_SCHEME): + if not authorization: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token", headers={"WWW-Authenticate": "Bearer"}, ) - return authorization[len(BEARER_SCHEME) :].strip() + parts = authorization.split(None, 1) + if len(parts) != 2 or parts[0].lower() != BEARER_SCHEME: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Missing bearer token", + headers={"WWW-Authenticate": "Bearer"}, + ) + return parts[1] async def workiq_token( diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 6673bac..4a127d5 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -167,5 +167,5 @@ def _raise_for_status(response: httpx.Response, action: str) -> None: request_id = response.headers.get("request-id", "unknown") raise WorkIQError( f"{action} failed: {response.status_code} " - f"(request-id={request_id}) {response.text[:500]}" + f"(request-id={request_id})" ) From 72df7c2e3b54b21d0c4b5e546037c6f426e56f66 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Thu, 16 Jul 2026 23:54:47 +0400 Subject: [PATCH 05/36] chore: strip trailing whitespace from bearer token, consolidate error string Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/main.py | 2 +- python/obo/app/workiq.py | 3 +-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/python/obo/app/main.py b/python/obo/app/main.py index 560e1df..3cb6950 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -76,7 +76,7 @@ def _bearer_token(authorization: Annotated[str | None, Header()] = None) -> str: detail="Missing bearer token", headers={"WWW-Authenticate": "Bearer"}, ) - return parts[1] + return parts[1].strip() async def workiq_token( diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 4a127d5..5e28445 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -166,6 +166,5 @@ def _raise_for_status(response: httpx.Response, action: str) -> None: # request-id is what Microsoft support asks for; keep it out of client replies. request_id = response.headers.get("request-id", "unknown") raise WorkIQError( - f"{action} failed: {response.status_code} " - f"(request-id={request_id})" + f"{action} failed: {response.status_code} (request-id={request_id})" ) From d2b94091b7b66f4b64d45f094c76ea56572ef9c8 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:00:54 +0400 Subject: [PATCH 06/36] fix: guard smoke_test with __name__ check, fix README request-id docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Wrap asyncio.run(main()) in if __name__ == "__main__" to prevent side effects on import - README: clarify that request-id is surfaced in WorkIQError exception (workiq.py), not logged there — logging happens in main.py Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/README.md | 3 ++- python/obo/smoke_test.py | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/python/obo/README.md b/python/obo/README.md index 3ba4238..36ec848 100644 --- a/python/obo/README.md +++ b/python/obo/README.md @@ -199,7 +199,8 @@ See the [root README](../../README.md#troubleshooting) for the full matrix (Copi - **Cache the OBO result.** Each request currently exchanges the token again. Cache on a hash of the inbound token, honoring `expires_on`, to save a round trip per turn. - **The `request-id` response header** is what Microsoft support asks for. - [`app/workiq.py`](app/workiq.py) logs it but keeps it out of client-facing errors. + [`app/workiq.py`](app/workiq.py) includes it in the `WorkIQError` exception; + [`app/main.py`](app/main.py) logs it when handling errors. - **Never log tokens.** Errors here are logged with the user's `oid`, not the assertion. ## Resources diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py index 893cae2..3ea2a79 100644 --- a/python/obo/smoke_test.py +++ b/python/obo/smoke_test.py @@ -121,4 +121,5 @@ async def main() -> None: print("\nAll checks passed.") -asyncio.run(main()) +if __name__ == "__main__": + asyncio.run(main()) From b3bdd53dce6ec495b642e21f22b43ef469aa2e10 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:17:25 +0400 Subject: [PATCH 07/36] fix: wrap httpx transport errors in WorkIQError for proper 502 handling All three WorkIQClient methods (create_conversation, chat, chat_stream) now catch httpx.HTTPError and re-raise as WorkIQError so that main.py's existing except-WorkIQError handler returns 502 instead of an unhandled 500. Also catch ValueError from response.json() for non-JSON gateway responses. Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/workiq.py | 31 ++++++++++++++++++++++++------- 1 file changed, 24 insertions(+), 7 deletions(-) diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 5e28445..99d3543 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -93,21 +93,35 @@ async def __aexit__(self, *_: Any) -> None: await self._client.aclose() async def create_conversation(self) -> str: - response = await self._client.post("conversations", json={}) + try: + response = await self._client.post("conversations", json={}) + except httpx.HTTPError as exc: + raise WorkIQError(f"create conversation failed: {exc}") from exc self._raise_for_status(response, "create conversation") - conversation_id = response.json().get("id") + try: + conversation_id = response.json().get("id") + except ValueError as exc: + raise WorkIQError("create conversation: invalid JSON response") from exc if not conversation_id: raise WorkIQError("no conversation id in response") return conversation_id async def chat(self, conversation_id: str, message: str) -> ChatReply: - response = await self._client.post( - f"conversations/{conversation_id}/chat", json=_chat_body(message) - ) + try: + response = await self._client.post( + f"conversations/{conversation_id}/chat", json=_chat_body(message) + ) + except httpx.HTTPError as exc: + raise WorkIQError(f"chat failed: {exc}") from exc self._raise_for_status(response, "chat") - reply = _last_text_message(response.json()) + try: + payload = response.json() + except ValueError as exc: + raise WorkIQError("chat: invalid JSON response") from exc + + reply = _last_text_message(payload) if reply is None: raise WorkIQError("no assistant message in response") @@ -127,7 +141,10 @@ async def chat_stream( f"conversations/{conversation_id}/chatOverStream", json=_chat_body(message), ) - response = await self._client.send(request, stream=True) + try: + response = await self._client.send(request, stream=True) + except httpx.HTTPError as exc: + raise WorkIQError(f"chat stream failed: {exc}") from exc try: if response.status_code >= 400: await response.aread() From d51e18dd2d81a3e9d75f0280d4158fb4a4678103 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:19:31 +0400 Subject: [PATCH 08/36] fix: catch httpx errors during SSE stream iteration Mid-stream transport failures (connection drops, read timeouts) during aiter_lines() were not wrapped in WorkIQError, causing them to bypass main.py's error handler and surface as unhandled 500s. Now caught and re-raised as WorkIQError for proper 502 handling. Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/workiq.py | 45 +++++++++++++++++++++------------------- 1 file changed, 24 insertions(+), 21 deletions(-) diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 99d3543..8537706 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -151,27 +151,30 @@ async def chat_stream( self._raise_for_status(response, "chat stream") previous = "" - async for line in response.aiter_lines(): - if not line.startswith(SSE_DATA_PREFIX): - continue - - event = line[len(SSE_DATA_PREFIX) :].strip() - if not event: - continue - - try: - reply = _last_text_message(json.loads(event)) - except json.JSONDecodeError: - continue # Skip malformed events rather than kill the stream. - - if reply is None: - continue - - text = reply.get("text", "") - delta = text[len(previous) :] if text.startswith(previous) else text - previous = text - if delta: - yield delta + try: + async for line in response.aiter_lines(): + if not line.startswith(SSE_DATA_PREFIX): + continue + + event = line[len(SSE_DATA_PREFIX) :].strip() + if not event: + continue + + try: + reply = _last_text_message(json.loads(event)) + except json.JSONDecodeError: + continue # Skip malformed events rather than kill the stream. + + if reply is None: + continue + + text = reply.get("text", "") + delta = text[len(previous) :] if text.startswith(previous) else text + previous = text + if delta: + yield delta + except httpx.HTTPError as exc: + raise WorkIQError(f"chat stream interrupted: {exc}") from exc finally: await response.aclose() From c8ff0f4614e074e2a5dbd81c74ca6fd0a3915ccc Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:21:24 +0400 Subject: [PATCH 09/36] refactor: simplify chat_stream error handling with single except block Consolidate the nested try/except for httpx.HTTPError into one block covering both response.aread() (error path) and aiter_lines() (streaming). Previously aread() failures would escape as unhandled httpx errors. Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/workiq.py | 47 ++++++++++++++++++++-------------------- 1 file changed, 23 insertions(+), 24 deletions(-) diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 8537706..8b07184 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -151,30 +151,29 @@ async def chat_stream( self._raise_for_status(response, "chat stream") previous = "" - try: - async for line in response.aiter_lines(): - if not line.startswith(SSE_DATA_PREFIX): - continue - - event = line[len(SSE_DATA_PREFIX) :].strip() - if not event: - continue - - try: - reply = _last_text_message(json.loads(event)) - except json.JSONDecodeError: - continue # Skip malformed events rather than kill the stream. - - if reply is None: - continue - - text = reply.get("text", "") - delta = text[len(previous) :] if text.startswith(previous) else text - previous = text - if delta: - yield delta - except httpx.HTTPError as exc: - raise WorkIQError(f"chat stream interrupted: {exc}") from exc + async for line in response.aiter_lines(): + if not line.startswith(SSE_DATA_PREFIX): + continue + + event = line[len(SSE_DATA_PREFIX) :].strip() + if not event: + continue + + try: + reply = _last_text_message(json.loads(event)) + except json.JSONDecodeError: + continue # Skip malformed events rather than kill the stream. + + if reply is None: + continue + + text = reply.get("text", "") + delta = text[len(previous) :] if text.startswith(previous) else text + previous = text + if delta: + yield delta + except httpx.HTTPError as exc: + raise WorkIQError(f"chat stream failed: {exc}") from exc finally: await response.aclose() From 00008777e48704671416fbd763333a8d58219106 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:22:39 +0400 Subject: [PATCH 10/36] test: add transport error coverage to smoke test Verify that httpx.ConnectError (simulating DNS/network failures) is properly wrapped in WorkIQError for all three client methods: create_conversation, chat, and chat_stream. Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/smoke_test.py | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py index 3ea2a79..cbcd436 100644 --- a/python/obo/smoke_test.py +++ b/python/obo/smoke_test.py @@ -90,6 +90,30 @@ async def main() -> None: assert "request-id=abc-123" in str(exc), exc print("403 error path ->", str(exc)[:60]) + # Transport error path: network failures must surface as WorkIQError. + def transport_error_handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("simulated DNS failure") + + async with WorkIQClient("t", BASE, transport=httpx.MockTransport(transport_error_handler)) as c: + try: + await c.create_conversation() + raise AssertionError("expected WorkIQError") + except WorkIQError as exc: + assert "simulated DNS failure" in str(exc), exc + print("transport error (create) ->", str(exc)[:60]) + + try: + await c.chat("conv-1", "hi") + raise AssertionError("expected WorkIQError") + except WorkIQError as exc: + print("transport error (chat) ->", str(exc)[:60]) + + try: + _ = [d async for d in c.chat_stream("conv-1", "hi")] + raise AssertionError("expected WorkIQError") + except WorkIQError as exc: + print("transport error (stream) ->", str(exc)[:60]) + # Pure helper assert _last_text_message({"messages": []}) is None assert _last_text_message({"messages": [{"role": "user"}]}) is None From a31abfd7fdded49daaed6f5b294b25559c37d384 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:23:41 +0400 Subject: [PATCH 11/36] docs: add 502 error to troubleshooting table Document the 502 Work IQ request failed error path so users know to check connectivity and request-id in logs when Gateway calls fail. Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/python/obo/README.md b/python/obo/README.md index 36ec848..8e27863 100644 --- a/python/obo/README.md +++ b/python/obo/README.md @@ -190,6 +190,7 @@ No MSAL wrapper needed — `azure-identity` builds on MSAL underneath. | `403 Unable to obtain Work IQ access` | The OBO exchange failed. Usually missing admin consent on `WorkIQAgent.Ask`, or the user lacks a Copilot license. | | `401` from the Gateway | The *outbound* token's `aud` is wrong — must be `api://workiq.svc.cloud.microsoft`, not your API. | | `AADSTS50013: Assertion failed signature validation` | The federated credential subject/issuer doesn't match your managed identity. | +| `502 Work IQ request failed` | The Gateway call failed — network error, timeout, or the Gateway returned a server error. Check connectivity and the `request-id` in logs. | | Slow first response per turn | Every request does a fresh OBO round trip. See [Notes before production](#notes-before-production). | See the [root README](../../README.md#troubleshooting) for the full matrix (Copilot license, consent, audience mismatch). From 75fecf7c43bc2550d7773dcc252734d1221c4808 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:24:41 +0400 Subject: [PATCH 12/36] docs: update WorkIQError docstring to cover transport failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The exception now wraps transport errors and bad responses, not just HTTP error codes — update the docstring to reflect this. Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/workiq.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 8b07184..5700b6b 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -20,7 +20,7 @@ class WorkIQError(Exception): - """Work IQ returned an error or an unparseable response.""" + """A Work IQ Gateway call failed (transport error, HTTP error, or bad response).""" @dataclass(frozen=True) From 32063160dae26b064a41ecfbf1159b6e1b3faa66 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:26:34 +0400 Subject: [PATCH 13/36] fix: avoid blocking event loop on credential close, guard response in finally - auth.py: wrap sync DefaultAzureCredential.close() in asyncio.to_thread to avoid blocking the event loop during lifespan shutdown - workiq.py: initialize response=None and guard the finally block so response.aclose() is only called when send() succeeded, preventing potential UnboundLocalError during async generator cleanup Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/auth.py | 2 +- python/obo/app/workiq.py | 7 +++---- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/python/obo/app/auth.py b/python/obo/app/auth.py index 73f7029..888497c 100644 --- a/python/obo/app/auth.py +++ b/python/obo/app/auth.py @@ -119,4 +119,4 @@ def _build_credential(self, user_assertion: str) -> OnBehalfOfCredential: async def close(self) -> None: if self._mi_credential is not None: - self._mi_credential.close() + await asyncio.to_thread(self._mi_credential.close) diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 5700b6b..5b3f56a 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -141,11 +141,9 @@ async def chat_stream( f"conversations/{conversation_id}/chatOverStream", json=_chat_body(message), ) + response = None try: response = await self._client.send(request, stream=True) - except httpx.HTTPError as exc: - raise WorkIQError(f"chat stream failed: {exc}") from exc - try: if response.status_code >= 400: await response.aread() self._raise_for_status(response, "chat stream") @@ -175,7 +173,8 @@ async def chat_stream( except httpx.HTTPError as exc: raise WorkIQError(f"chat stream failed: {exc}") from exc finally: - await response.aclose() + if response is not None: + await response.aclose() @staticmethod def _raise_for_status(response: httpx.Response, action: str) -> None: From b6f529e4d878050c93f3cbf6fe0a440755caad8e Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:31:02 +0400 Subject: [PATCH 14/36] fix: narrow OBO exception catch, validate conversation_id, catch DecodingError MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - main.py: narrow bare except Exception to (ClientAuthenticationError, HttpResponseError) so programming errors propagate as 500 instead of being silently swallowed as 403 - main.py: validate conversation_id with regex pattern to prevent path traversal via caller-supplied IDs injected into URL paths - workiq.py: catch httpx.DecodingError alongside ValueError when parsing response JSON — DecodingError is not a ValueError subclass and would escape as an unhandled 500 - smoke_test.py: clear get_settings lru_cache to avoid stale env vars when running in multi-test processes Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/main.py | 10 ++++++---- python/obo/app/workiq.py | 4 ++-- python/obo/smoke_test.py | 4 ++++ 3 files changed, 12 insertions(+), 6 deletions(-) diff --git a/python/obo/app/main.py b/python/obo/app/main.py index 3cb6950..17bb453 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -11,6 +11,7 @@ from contextlib import asynccontextmanager from typing import Annotated, AsyncIterator +from azure.core.exceptions import ClientAuthenticationError, HttpResponseError from fastapi import Depends, FastAPI, Header, HTTPException, status from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field @@ -26,7 +27,9 @@ class ChatRequest(BaseModel): message: str = Field(min_length=1, max_length=8000) - conversation_id: str | None = None + conversation_id: str | None = Field( + default=None, pattern=r"^[a-zA-Z0-9\-_]+$", max_length=128 + ) class CitationModel(BaseModel): @@ -98,9 +101,8 @@ async def workiq_token( try: return await exchange.token_for(inbound_token) - except Exception as exc: - # A failure here is usually consent, licensing, or a misconfigured app - # registration — log it with the user for triage, but do not leak it. + except (ClientAuthenticationError, HttpResponseError) as exc: + # Consent, licensing, or misconfigured app registration — log for triage. logger.error("OBO exchange failed for %s: %s", claims.get("oid"), exc) raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 5b3f56a..9fc5d3d 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -101,7 +101,7 @@ async def create_conversation(self) -> str: try: conversation_id = response.json().get("id") - except ValueError as exc: + except (ValueError, httpx.DecodingError) as exc: raise WorkIQError("create conversation: invalid JSON response") from exc if not conversation_id: raise WorkIQError("no conversation id in response") @@ -118,7 +118,7 @@ async def chat(self, conversation_id: str, message: str) -> ChatReply: try: payload = response.json() - except ValueError as exc: + except (ValueError, httpx.DecodingError) as exc: raise WorkIQError("chat: invalid JSON response") from exc reply = _last_text_message(payload) diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py index cbcd436..d064b84 100644 --- a/python/obo/smoke_test.py +++ b/python/obo/smoke_test.py @@ -18,8 +18,12 @@ import httpx from fastapi.testclient import TestClient +from app.config import get_settings from app.workiq import WorkIQClient, WorkIQError, _last_text_message +# Ensure env vars set above are picked up, even if config was imported earlier. +get_settings.cache_clear() + BASE = "https://workiq.test/rest/beta/" From caa8acdda889c03ffa98c97a748c6a84c56b3502 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:32:16 +0400 Subject: [PATCH 15/36] chore: add httpx2 dependency to silence starlette deprecation warning Starlette 1.3+ requires httpx2 for TestClient. Adding it alongside httpx (still needed by the app itself) eliminates the deprecation warning during smoke tests. Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/requirements.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/python/obo/requirements.txt b/python/obo/requirements.txt index cb674f1..3e97fbe 100644 --- a/python/obo/requirements.txt +++ b/python/obo/requirements.txt @@ -1,5 +1,6 @@ fastapi>=0.115 uvicorn[standard]>=0.32 httpx>=0.27 +httpx2>=2.7 azure-identity>=1.19 pyjwt[crypto]>=2.9 From 72c25c97a0e0baa120bc95a8ab3e0dfde5dda02f Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:36:46 +0400 Subject: [PATCH 16/36] fix: validate conversation_id type, add per-request timezone support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - workiq.py: assert conversation_id from gateway is a string, not just truthy — a non-string id would cause 422 on continuation turns - workiq.py + main.py: accept optional time_zone parameter so frontends can supply the user's IANA timezone instead of always using the server's locale (which is typically UTC in containers) - ChatRequest gains time_zone field with IANA pattern validation Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/main.py | 11 +++++++++-- python/obo/app/workiq.py | 17 ++++++++++------- 2 files changed, 19 insertions(+), 9 deletions(-) diff --git a/python/obo/app/main.py b/python/obo/app/main.py index 17bb453..354d476 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -30,6 +30,9 @@ class ChatRequest(BaseModel): conversation_id: str | None = Field( default=None, pattern=r"^[a-zA-Z0-9\-_]+$", max_length=128 ) + time_zone: str | None = Field( + default=None, pattern=r"^[A-Za-z_]+/[A-Za-z_/]+$", max_length=64 + ) class CitationModel(BaseModel): @@ -124,7 +127,9 @@ async def chat( try: async with _client(token, settings) as client: conversation_id = request.conversation_id or await client.create_conversation() - reply = await client.chat(conversation_id, request.message) + reply = await client.chat( + conversation_id, request.message, time_zone=request.time_zone + ) except WorkIQError as exc: logger.error("work iq call failed: %s", exc) raise HTTPException( @@ -163,7 +168,9 @@ async def events() -> AsyncIterator[str]: # JSON-encode each delta: raw newlines in the text would # otherwise terminate the SSE frame early. - async for delta in client.chat_stream(conversation_id, request.message): + async for delta in client.chat_stream( + conversation_id, request.message, time_zone=request.time_zone + ): yield f"data: {json.dumps({'text': delta})}\n\n" except WorkIQError as exc: # The response has already started, so the error rides the stream. diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 9fc5d3d..0f9a122 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -45,10 +45,10 @@ def _local_timezone() -> str: return name if "/" in name or name == "UTC" else "UTC" -def _chat_body(message: str) -> dict[str, Any]: +def _chat_body(message: str, time_zone: str | None = None) -> dict[str, Any]: return { "message": {"text": message}, - "locationHint": {"timeZone": _local_timezone()}, + "locationHint": {"timeZone": time_zone or _local_timezone()}, } @@ -103,14 +103,17 @@ async def create_conversation(self) -> str: conversation_id = response.json().get("id") except (ValueError, httpx.DecodingError) as exc: raise WorkIQError("create conversation: invalid JSON response") from exc - if not conversation_id: + if not isinstance(conversation_id, str) or not conversation_id: raise WorkIQError("no conversation id in response") return conversation_id - async def chat(self, conversation_id: str, message: str) -> ChatReply: + async def chat( + self, conversation_id: str, message: str, *, time_zone: str | None = None + ) -> ChatReply: try: response = await self._client.post( - f"conversations/{conversation_id}/chat", json=_chat_body(message) + f"conversations/{conversation_id}/chat", + json=_chat_body(message, time_zone), ) except httpx.HTTPError as exc: raise WorkIQError(f"chat failed: {exc}") from exc @@ -128,7 +131,7 @@ async def chat(self, conversation_id: str, message: str) -> ChatReply: return ChatReply(text=reply["text"], citations=_parse_citations(reply)) async def chat_stream( - self, conversation_id: str, message: str + self, conversation_id: str, message: str, *, time_zone: str | None = None ) -> AsyncIterator[str]: """Yield text deltas as they arrive. @@ -139,7 +142,7 @@ async def chat_stream( request = self._client.build_request( "POST", f"conversations/{conversation_id}/chatOverStream", - json=_chat_body(message), + json=_chat_body(message, time_zone), ) response = None try: From 44d84126745c715c61804ac3f5b04e1b10833846 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:38:13 +0400 Subject: [PATCH 17/36] fix: use request.app.state in workiq_token instead of module-level app Inject FastAPI Request and read state from request.app.state so the dependency always references the live app instance, improving test isolation if the app object is ever replaced. Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/main.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/python/obo/app/main.py b/python/obo/app/main.py index 354d476..2d5a5c4 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -12,7 +12,7 @@ from typing import Annotated, AsyncIterator from azure.core.exceptions import ClientAuthenticationError, HttpResponseError -from fastapi import Depends, FastAPI, Header, HTTPException, status +from fastapi import Depends, FastAPI, Header, HTTPException, Request, status from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field @@ -86,11 +86,12 @@ def _bearer_token(authorization: Annotated[str | None, Header()] = None) -> str: async def workiq_token( + request: Request, inbound_token: Annotated[str, Depends(_bearer_token)], ) -> str: """Validate the caller's token and exchange it for a Work IQ token.""" - validator: TokenValidator = app.state.validator - exchange: WorkIQTokenExchange = app.state.exchange + validator: TokenValidator = request.app.state.validator + exchange: WorkIQTokenExchange = request.app.state.exchange try: claims = await validator.validate(inbound_token) From 7fa859814f6913cba86af1cfe723a917c893f70f Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:43:16 +0400 Subject: [PATCH 18/36] feat: emit SSE done sentinel on success, document streaming contract - Emit "event: done" after all deltas so clients can distinguish clean completion from a dropped connection - Document that clients must handle "event: error" frames since the HTTP 200 is already committed when mid-stream errors occur - Update README with time_zone request field and streaming contract Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/README.md | 6 ++++-- python/obo/app/main.py | 4 +++- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/python/obo/README.md b/python/obo/README.md index 8e27863..eb6f752 100644 --- a/python/obo/README.md +++ b/python/obo/README.md @@ -130,9 +130,11 @@ no network. | Endpoint | Mode | Response | |----------|------|----------| | `POST /api/chat` | Synchronous | JSON — `conversation_id`, `text`, `citations` | -| `POST /api/chat/stream` | SSE | `event: conversation` then `data: {"text": ""}` frames | +| `POST /api/chat/stream` | SSE | `event: conversation`, `data: {"text": ""}` frames, `event: done` on success, `event: error` on failure | -Request body for both: `{"message": "...", "conversation_id": "..."}` (`conversation_id` optional). +Request body for both: `{"message": "...", "conversation_id": "...", "time_zone": "..."}` (`conversation_id` and `time_zone` optional; `time_zone` is an IANA identifier like `America/New_York`). + +**Streaming contract:** once the HTTP 200 is committed, errors cannot change the status code. Clients **must** listen for `event: error` frames to detect mid-stream failures. A `event: done` frame signals clean completion; its absence (with no `error`) indicates a dropped connection. ## Layout diff --git a/python/obo/app/main.py b/python/obo/app/main.py index 2d5a5c4..84cbe3c 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -173,8 +173,10 @@ async def events() -> AsyncIterator[str]: conversation_id, request.message, time_zone=request.time_zone ): yield f"data: {json.dumps({'text': delta})}\n\n" + yield "event: done\ndata: \n\n" except WorkIQError as exc: - # The response has already started, so the error rides the stream. + # The HTTP 200 is already committed, so the error rides the stream. + # Clients must handle "error" events to detect mid-stream failures. logger.error("work iq stream failed: %s", exc) yield "event: error\ndata: Work IQ request failed\n\n" From ed5cf72188d53ae65a9c669bbe45e9886929694f Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:49:25 +0400 Subject: [PATCH 19/36] fix: relax time_zone regex to accept UTC, GMT, and Etc/* identifiers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous pattern required a slash, rejecting valid IANA IDs like UTC and Etc/GMT+5. Now allows alphanumeric, underscores, hyphens, plus signs, and slashes — still rejects path traversal and spaces. Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/obo/app/main.py b/python/obo/app/main.py index 84cbe3c..eee78bf 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -31,7 +31,7 @@ class ChatRequest(BaseModel): default=None, pattern=r"^[a-zA-Z0-9\-_]+$", max_length=128 ) time_zone: str | None = Field( - default=None, pattern=r"^[A-Za-z_]+/[A-Za-z_/]+$", max_length=64 + default=None, pattern=r"^[A-Za-z0-9_+\-/]+$", max_length=64 ) From 6bf5f89149b16070fada66dd88dd3050d42de749 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:52:09 +0400 Subject: [PATCH 20/36] fix: handle non-dict JSON responses from gateway without crashing If the gateway returns a JSON array or scalar instead of an object, response.json().get() raises AttributeError. Now check isinstance before accessing dict methods. Also guard _last_text_message against non-dict payloads and non-dict message entries. Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/workiq.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 0f9a122..35dff79 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -64,9 +64,11 @@ def _parse_citations(message: dict[str, Any]) -> tuple[Citation, ...]: ) -def _last_text_message(payload: dict[str, Any]) -> dict[str, Any] | None: +def _last_text_message(payload: Any) -> dict[str, Any] | None: """The assistant's reply is the last message carrying a `text` field.""" - candidates = [m for m in payload.get("messages", []) if "text" in m] + if not isinstance(payload, dict): + return None + candidates = [m for m in payload.get("messages", []) if isinstance(m, dict) and "text" in m] return candidates[-1] if candidates else None @@ -100,7 +102,8 @@ async def create_conversation(self) -> str: self._raise_for_status(response, "create conversation") try: - conversation_id = response.json().get("id") + body = response.json() + conversation_id = body.get("id") if isinstance(body, dict) else None except (ValueError, httpx.DecodingError) as exc: raise WorkIQError("create conversation: invalid JSON response") from exc if not isinstance(conversation_id, str) or not conversation_id: From 84bcf0dcbf45c86c3b6a17caeb1c2a184d21393a Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 00:55:27 +0400 Subject: [PATCH 21/36] test: add edge case coverage for non-dict payloads in _last_text_message Verify that non-dict JSON payloads (arrays, None, non-dict message entries) are handled gracefully and return None instead of crashing with AttributeError or TypeError. Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/smoke_test.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py index d064b84..1e57353 100644 --- a/python/obo/smoke_test.py +++ b/python/obo/smoke_test.py @@ -118,9 +118,12 @@ def transport_error_handler(request: httpx.Request) -> httpx.Response: except WorkIQError as exc: print("transport error (stream) ->", str(exc)[:60]) - # Pure helper + # Pure helper — edge cases including non-dict payloads assert _last_text_message({"messages": []}) is None assert _last_text_message({"messages": [{"role": "user"}]}) is None + assert _last_text_message([1, 2, 3]) is None # non-dict payload + assert _last_text_message(None) is None # type: ignore[arg-type] + assert _last_text_message({"messages": ["not-a-dict"]}) is None print("helper edge cases -> ok") # Auth gate: no token and malformed token must both 401 before any Work IQ call. From e48c42e57712c600b7292a26cf5445dd56dd035d Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 01:04:10 +0400 Subject: [PATCH 22/36] fix: JSON-encode SSE conversation event, clarify httpx2 as test dep - main.py: emit conversation event as JSON ({"conversation_id": "..."}) instead of a raw string, making the stream contract consistent - README: document the exact SSE event sequence with JSON shapes - requirements.txt: move httpx2 below a comment clarifying it is a test dependency required by starlette.testclient Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/README.md | 8 +++++++- python/obo/app/main.py | 2 +- python/obo/requirements.txt | 4 +++- 3 files changed, 11 insertions(+), 3 deletions(-) diff --git a/python/obo/README.md b/python/obo/README.md index eb6f752..ec35bda 100644 --- a/python/obo/README.md +++ b/python/obo/README.md @@ -130,7 +130,13 @@ no network. | Endpoint | Mode | Response | |----------|------|----------| | `POST /api/chat` | Synchronous | JSON — `conversation_id`, `text`, `citations` | -| `POST /api/chat/stream` | SSE | `event: conversation`, `data: {"text": ""}` frames, `event: done` on success, `event: error` on failure | +| `POST /api/chat/stream` | SSE | See streaming contract below | + +**Stream event sequence:** + +1. `event: conversation` — `data: {"conversation_id": "..."}` (first frame) +2. `data: {"text": ""}` — one per text chunk (default event type) +3. `event: done` — signals clean completion, or `event: error` — `data: Work IQ request failed` on failure Request body for both: `{"message": "...", "conversation_id": "...", "time_zone": "..."}` (`conversation_id` and `time_zone` optional; `time_zone` is an IANA identifier like `America/New_York`). diff --git a/python/obo/app/main.py b/python/obo/app/main.py index eee78bf..5eea022 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -165,7 +165,7 @@ async def events() -> AsyncIterator[str]: conversation_id = ( request.conversation_id or await client.create_conversation() ) - yield f"event: conversation\ndata: {conversation_id}\n\n" + yield f"event: conversation\ndata: {json.dumps({'conversation_id': conversation_id})}\n\n" # JSON-encode each delta: raw newlines in the text would # otherwise terminate the SSE frame early. diff --git a/python/obo/requirements.txt b/python/obo/requirements.txt index 3e97fbe..89b4390 100644 --- a/python/obo/requirements.txt +++ b/python/obo/requirements.txt @@ -1,6 +1,8 @@ fastapi>=0.115 uvicorn[standard]>=0.32 httpx>=0.27 -httpx2>=2.7 azure-identity>=1.19 pyjwt[crypto]>=2.9 + +# Test dependency: starlette.testclient requires httpx2 at runtime. +httpx2>=2.7 From efd1d957417080b705a49d82b9e33b0e8c538fd8 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 01:16:33 +0400 Subject: [PATCH 23/36] fix: guard messages:null in _last_text_message, use request.app.state in routes - workiq.py: handle messages:null from gateway without TypeError - main.py: replace module-level app.state with _get_settings dependency injected via request.app.state, consistent with workiq_token - smoke_test.py: add test case for messages:null payload Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/main.py | 9 ++++++--- python/obo/app/workiq.py | 3 ++- python/obo/smoke_test.py | 1 + 3 files changed, 9 insertions(+), 4 deletions(-) diff --git a/python/obo/app/main.py b/python/obo/app/main.py index 5eea022..f7dcafe 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -114,6 +114,10 @@ async def workiq_token( ) from exc +def _get_settings(request: Request) -> Settings: + return request.app.state.settings + + def _client(token: str, settings: Settings) -> WorkIQClient: return WorkIQClient(access_token=token, base_url=settings.workiq_base) @@ -122,9 +126,8 @@ def _client(token: str, settings: Settings) -> WorkIQClient: async def chat( request: ChatRequest, token: Annotated[str, Depends(workiq_token)], + settings: Annotated[Settings, Depends(_get_settings)], ) -> ChatResponse: - settings: Settings = app.state.settings - try: async with _client(token, settings) as client: conversation_id = request.conversation_id or await client.create_conversation() @@ -156,8 +159,8 @@ async def chat( async def chat_stream( request: ChatRequest, token: Annotated[str, Depends(workiq_token)], + settings: Annotated[Settings, Depends(_get_settings)], ) -> StreamingResponse: - settings: Settings = app.state.settings async def events() -> AsyncIterator[str]: try: diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 35dff79..e5eb2c8 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -68,7 +68,8 @@ def _last_text_message(payload: Any) -> dict[str, Any] | None: """The assistant's reply is the last message carrying a `text` field.""" if not isinstance(payload, dict): return None - candidates = [m for m in payload.get("messages", []) if isinstance(m, dict) and "text" in m] + messages = payload.get("messages") or [] + candidates = [m for m in messages if isinstance(m, dict) and "text" in m] return candidates[-1] if candidates else None diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py index 1e57353..c15fd12 100644 --- a/python/obo/smoke_test.py +++ b/python/obo/smoke_test.py @@ -124,6 +124,7 @@ def transport_error_handler(request: httpx.Request) -> httpx.Response: assert _last_text_message([1, 2, 3]) is None # non-dict payload assert _last_text_message(None) is None # type: ignore[arg-type] assert _last_text_message({"messages": ["not-a-dict"]}) is None + assert _last_text_message({"messages": None}) is None # messages: null print("helper edge cases -> ok") # Auth gate: no token and malformed token must both 401 before any Work IQ call. From c1bb1c9d2da3dc9c97f2c8999334d080d2061ba7 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 01:18:34 +0400 Subject: [PATCH 24/36] fix: validate conversation_id at client level, cache server timezone MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Validate conversation_id against safe regex in WorkIQClient.chat and chat_stream before URL interpolation — prevents path manipulation from a misbehaving gateway return - Also validate the gateway-returned id in create_conversation - Cache _local_timezone() result at module import to avoid blocking syscall on every request Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/workiq.py | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index e5eb2c8..284af6e 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -9,6 +9,7 @@ from __future__ import annotations import json +import re from dataclasses import dataclass, field from datetime import datetime from typing import Any, AsyncIterator @@ -17,6 +18,7 @@ REQUEST_TIMEOUT_SECONDS = 300.0 SSE_DATA_PREFIX = "data: " +_CONV_ID_RE = re.compile(r"^[a-zA-Z0-9\-_]{1,128}$") class WorkIQError(Exception): @@ -37,7 +39,7 @@ class ChatReply: citations: tuple[Citation, ...] = field(default=()) -def _local_timezone() -> str: +def _detect_server_timezone() -> str: """Work IQ requires an IANA timezone (e.g. America/Los_Angeles).""" tz = datetime.now().astimezone().tzinfo name = getattr(tz, "key", None) or str(tz) @@ -45,10 +47,14 @@ def _local_timezone() -> str: return name if "/" in name or name == "UTC" else "UTC" +# Computed once at import — the server timezone cannot change at runtime. +_SERVER_TIMEZONE: str = _detect_server_timezone() + + def _chat_body(message: str, time_zone: str | None = None) -> dict[str, Any]: return { "message": {"text": message}, - "locationHint": {"timeZone": time_zone or _local_timezone()}, + "locationHint": {"timeZone": time_zone or _SERVER_TIMEZONE}, } @@ -107,13 +113,19 @@ async def create_conversation(self) -> str: conversation_id = body.get("id") if isinstance(body, dict) else None except (ValueError, httpx.DecodingError) as exc: raise WorkIQError("create conversation: invalid JSON response") from exc - if not isinstance(conversation_id, str) or not conversation_id: - raise WorkIQError("no conversation id in response") + if not isinstance(conversation_id, str) or not _CONV_ID_RE.fullmatch(conversation_id): + raise WorkIQError("no valid conversation id in response") return conversation_id + @staticmethod + def _validate_conversation_id(conversation_id: str) -> None: + if not _CONV_ID_RE.fullmatch(conversation_id): + raise WorkIQError(f"invalid conversation id: {conversation_id!r}") + async def chat( self, conversation_id: str, message: str, *, time_zone: str | None = None ) -> ChatReply: + self._validate_conversation_id(conversation_id) try: response = await self._client.post( f"conversations/{conversation_id}/chat", @@ -143,6 +155,7 @@ async def chat_stream( against the previous one. A non-prefix update would re-emit the full text; that does not happen under the current append-only contract. """ + self._validate_conversation_id(conversation_id) request = self._client.build_request( "POST", f"conversations/{conversation_id}/chatOverStream", From caaf05e12adb348b41369cf040729eb23e87db58 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 01:21:31 +0400 Subject: [PATCH 25/36] fix: guard attributions:null, use consistent .get() for reply text - _parse_citations: handle attributions:null without TypeError, same pattern as the messages:null fix - chat(): use reply.get("text", "") consistently with chat_stream() instead of reply["text"] Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/workiq.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 284af6e..46a239b 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -66,7 +66,7 @@ def _parse_citations(message: dict[str, Any]) -> tuple[Citation, ...]: provider_display_name=a.get("providerDisplayName", ""), see_more_web_url=a.get("seeMoreWebUrl", ""), ) - for a in message.get("attributions", []) + for a in (message.get("attributions") or []) ) @@ -144,7 +144,7 @@ async def chat( if reply is None: raise WorkIQError("no assistant message in response") - return ChatReply(text=reply["text"], citations=_parse_citations(reply)) + return ChatReply(text=reply.get("text", ""), citations=_parse_citations(reply)) async def chat_stream( self, conversation_id: str, message: str, *, time_zone: str | None = None From be2f9f4e25de927dccb3254f32f3adf984b367b3 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Fri, 17 Jul 2026 01:22:40 +0400 Subject: [PATCH 26/36] fix: guard non-dict attribution entries, add time_zone to curl example - _parse_citations: skip non-dict entries in attributions list to prevent AttributeError on malformed gateway responses - README: add time_zone to the curl example to demonstrate the feature Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/README.md | 2 +- python/obo/app/workiq.py | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/python/obo/README.md b/python/obo/README.md index ec35bda..54517ba 100644 --- a/python/obo/README.md +++ b/python/obo/README.md @@ -100,7 +100,7 @@ uvicorn app.main:app --reload curl -X POST localhost:8000/api/chat \ -H "Authorization: Bearer /access_as_user>" \ -H "Content-Type: application/json" \ - -d '{"message": "What meetings do I have tomorrow?"}' + -d '{"message": "What meetings do I have tomorrow?", "time_zone": "America/New_York"}' ``` ```json diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 46a239b..1de79a1 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -67,6 +67,7 @@ def _parse_citations(message: dict[str, Any]) -> tuple[Citation, ...]: see_more_web_url=a.get("seeMoreWebUrl", ""), ) for a in (message.get("attributions") or []) + if isinstance(a, dict) ) From 15cd4ba95b70cf1f532b10cae3a554c2fd400503 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Sun, 2 Aug 2026 14:59:45 +0400 Subject: [PATCH 27/36] =?UTF-8?q?fix:=20address=20code=20review=20findings?= =?UTF-8?q?=20=E2=80=94=20security=20hardening,=20type=20correctness,=20an?= =?UTF-8?q?d=20cleanup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove incorrect httpx2 dep, bump azure-identity>=1.25.3 for client_assertion_func - Add WORKIQ_HOST allowlist to prevent SSRF via misconfigured env - Add body size limit middleware (64 KB) and security response headers - Add /healthz endpoint for Kubernetes liveness probes - Fix AsyncIterator -> AsyncGenerator return types on async generators - Explicit rejection of app-only tokens with clear error message - Add JWKS cache TTL (1h) for key rotation resilience - Centralize CONV_ID_PATTERN, validate before URL construction - Log warning on timezone detection fallback instead of silent UTC - Use Self return type on __aenter__, distinct 401 detail messages - Generic SSE error message to avoid leaking upstream service name Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/README.md | 1 + python/obo/app/auth.py | 10 +++++-- python/obo/app/config.py | 19 +++++++++++- python/obo/app/main.py | 58 ++++++++++++++++++++++++++++++++----- python/obo/app/workiq.py | 28 +++++++++++++----- python/obo/requirements.txt | 5 +--- python/obo/smoke_test.py | 2 ++ 7 files changed, 102 insertions(+), 21 deletions(-) diff --git a/python/obo/README.md b/python/obo/README.md index 54517ba..b3643ec 100644 --- a/python/obo/README.md +++ b/python/obo/README.md @@ -129,6 +129,7 @@ no network. | Endpoint | Mode | Response | |----------|------|----------| +| `GET /healthz` | — | `{"status": "ok"}` (unauthenticated, for liveness probes) | | `POST /api/chat` | Synchronous | JSON — `conversation_id`, `text`, `citations` | | `POST /api/chat/stream` | SSE | See streaming contract below | diff --git a/python/obo/app/auth.py b/python/obo/app/auth.py index 888497c..09952e8 100644 --- a/python/obo/app/auth.py +++ b/python/obo/app/auth.py @@ -34,7 +34,9 @@ class TokenValidator: def __init__(self, settings: Settings) -> None: self._settings = settings - self._jwks_client = PyJWKClient(settings.jwks_uri, cache_keys=True) + self._jwks_client = PyJWKClient( + settings.jwks_uri, cache_keys=True, lifespan=3600 + ) async def validate(self, token: str) -> dict[str, Any]: try: @@ -56,7 +58,11 @@ async def validate(self, token: str) -> dict[str, Any]: return claims def _require_scope(self, claims: dict[str, Any]) -> None: - granted = set(str(claims.get("scp", "")).split()) + if "scp" not in claims: + raise InvalidToken( + "app-only tokens are not accepted; a delegated user token is required" + ) + granted = set(str(claims["scp"]).split()) if self._settings.required_scope not in granted: raise InvalidToken( f"token is missing the required scope '{self._settings.required_scope}'" diff --git a/python/obo/app/config.py b/python/obo/app/config.py index cb8dbe4..d54c151 100644 --- a/python/obo/app/config.py +++ b/python/obo/app/config.py @@ -15,6 +15,13 @@ WORKIQ_DEFAULT_HOST = "https://workiq.svc.cloud.microsoft" WORKIQ_PATH = "/rest/beta" +# Allowlist of valid Work IQ Gateway hosts. Expand this set for staging/test +# environments rather than removing the check — an unrestricted WORKIQ_HOST +# env var could redirect OBO tokens to an attacker-controlled endpoint. +_ALLOWED_WORKIQ_HOSTS = frozenset({ + WORKIQ_DEFAULT_HOST, +}) + # Work IQ resource. WorkIQAgent.Ask is granted by admin consent on the app # registration, so `.default` resolves to it without naming the scope here. WORKIQ_SCOPE = "api://workiq.svc.cloud.microsoft/.default" @@ -69,6 +76,14 @@ def _require(name: str) -> str: return value +def _validated_workiq_host(host: str) -> str: + if host not in _ALLOWED_WORKIQ_HOSTS: + raise ConfigError( + f"WORKIQ_HOST {host!r} is not in the allowed list: {_ALLOWED_WORKIQ_HOSTS}" + ) + return host + + @lru_cache(maxsize=1) def get_settings() -> Settings: secret = os.environ.get("AZURE_CLIENT_SECRET", "").strip() @@ -77,6 +92,8 @@ def get_settings() -> Settings: client_id=_require("AZURE_CLIENT_ID"), api_audience=_require("API_AUDIENCE"), required_scope=os.environ.get("REQUIRED_SCOPE", "").strip() or DEFAULT_REQUIRED_SCOPE, - workiq_host=os.environ.get("WORKIQ_HOST", "").strip() or WORKIQ_DEFAULT_HOST, + workiq_host=_validated_workiq_host( + os.environ.get("WORKIQ_HOST", "").strip() or WORKIQ_DEFAULT_HOST + ), client_secret=secret or None, ) diff --git a/python/obo/app/main.py b/python/obo/app/main.py index f7dcafe..105cac3 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -8,27 +8,64 @@ import json import logging +from collections.abc import AsyncGenerator from contextlib import asynccontextmanager from typing import Annotated, AsyncIterator from azure.core.exceptions import ClientAuthenticationError, HttpResponseError -from fastapi import Depends, FastAPI, Header, HTTPException, Request, status -from fastapi.responses import StreamingResponse +from fastapi import Depends, FastAPI, Header, HTTPException, Request, Response, status +from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel, Field +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.types import ASGIApp, Receive, Scope, Send from .auth import InvalidToken, TokenValidator, WorkIQTokenExchange from .config import Settings, get_settings -from .workiq import WorkIQClient, WorkIQError +from .workiq import CONV_ID_PATTERN, WorkIQClient, WorkIQError logger = logging.getLogger(__name__) BEARER_SCHEME = "bearer" +MAX_REQUEST_BODY_BYTES = 64 * 1024 # 64 KB — well above the 8 KB message limit + + +class _BodySizeLimitMiddleware: + """Reject requests whose Content-Length exceeds the configured cap.""" + + def __init__(self, app: ASGIApp, *, max_bytes: int = MAX_REQUEST_BODY_BYTES) -> None: + self._app = app + self._max_bytes = max_bytes + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] == "http": + headers = dict(scope.get("headers", [])) + length = headers.get(b"content-length") + if length is not None and int(length) > self._max_bytes: + response = JSONResponse( + {"detail": "Request body too large"}, + status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + ) + await response(scope, receive, send) + return + await self._app(scope, receive, send) + + +class _SecurityHeadersMiddleware(BaseHTTPMiddleware): + """Add baseline security headers to every response.""" + + async def dispatch(self, request: Request, call_next): # type: ignore[override] + response: Response = await call_next(request) + response.headers["X-Content-Type-Options"] = "nosniff" + response.headers["X-Frame-Options"] = "DENY" + response.headers["Cache-Control"] = "no-store" + response.headers["Referrer-Policy"] = "no-referrer" + return response class ChatRequest(BaseModel): message: str = Field(min_length=1, max_length=8000) conversation_id: str | None = Field( - default=None, pattern=r"^[a-zA-Z0-9\-_]+$", max_length=128 + default=None, pattern=CONV_ID_PATTERN, max_length=128 ) time_zone: str | None = Field( default=None, pattern=r"^[A-Za-z0-9_+\-/]+$", max_length=64 @@ -66,6 +103,13 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: app = FastAPI(title="Work IQ OBO Backend", lifespan=lifespan) +app.add_middleware(_SecurityHeadersMiddleware) +app.add_middleware(_BodySizeLimitMiddleware) + + +@app.get("/healthz") +async def healthz() -> dict[str, str]: + return {"status": "ok"} def _bearer_token(authorization: Annotated[str | None, Header()] = None) -> str: @@ -79,7 +123,7 @@ def _bearer_token(authorization: Annotated[str | None, Header()] = None) -> str: if len(parts) != 2 or parts[0].lower() != BEARER_SCHEME: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, - detail="Missing bearer token", + detail="Authorization header must use Bearer scheme", headers={"WWW-Authenticate": "Bearer"}, ) return parts[1].strip() @@ -162,7 +206,7 @@ async def chat_stream( settings: Annotated[Settings, Depends(_get_settings)], ) -> StreamingResponse: - async def events() -> AsyncIterator[str]: + async def events() -> AsyncGenerator[str, None]: try: async with _client(token, settings) as client: conversation_id = ( @@ -181,6 +225,6 @@ async def events() -> AsyncIterator[str]: # The HTTP 200 is already committed, so the error rides the stream. # Clients must handle "error" events to detect mid-stream failures. logger.error("work iq stream failed: %s", exc) - yield "event: error\ndata: Work IQ request failed\n\n" + yield "event: error\ndata: upstream request failed\n\n" return StreamingResponse(events(), media_type="text/event-stream") diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 1de79a1..f438680 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -9,16 +9,21 @@ from __future__ import annotations import json +import logging import re +from collections.abc import AsyncGenerator from dataclasses import dataclass, field from datetime import datetime -from typing import Any, AsyncIterator +from typing import Any, Self import httpx +logger = logging.getLogger(__name__) + REQUEST_TIMEOUT_SECONDS = 300.0 SSE_DATA_PREFIX = "data: " -_CONV_ID_RE = re.compile(r"^[a-zA-Z0-9\-_]{1,128}$") +CONV_ID_PATTERN = r"^[a-zA-Z0-9\-_]{1,128}$" +_CONV_ID_RE = re.compile(CONV_ID_PATTERN) class WorkIQError(Exception): @@ -44,7 +49,14 @@ def _detect_server_timezone() -> str: tz = datetime.now().astimezone().tzinfo name = getattr(tz, "key", None) or str(tz) # A bare UTC offset like "+05:30" is not an IANA id; fall back rather than 400. - return name if "/" in name or name == "UTC" else "UTC" + if "/" in name or name == "UTC": + return name + logger.warning( + "Could not detect IANA timezone (got %r); defaulting to UTC. " + "Pass time_zone in requests to override.", + name, + ) + return "UTC" # Computed once at import — the server timezone cannot change at runtime. @@ -96,7 +108,7 @@ def __init__( transport=transport, ) - async def __aenter__(self) -> "WorkIQClient": + async def __aenter__(self) -> Self: return self async def __aexit__(self, *_: Any) -> None: @@ -127,9 +139,10 @@ async def chat( self, conversation_id: str, message: str, *, time_zone: str | None = None ) -> ChatReply: self._validate_conversation_id(conversation_id) + url = f"conversations/{conversation_id}/chat" try: response = await self._client.post( - f"conversations/{conversation_id}/chat", + url, json=_chat_body(message, time_zone), ) except httpx.HTTPError as exc: @@ -149,7 +162,7 @@ async def chat( async def chat_stream( self, conversation_id: str, message: str, *, time_zone: str | None = None - ) -> AsyncIterator[str]: + ) -> AsyncGenerator[str, None]: """Yield text deltas as they arrive. The gateway streams cumulative, append-only text, so each event is diffed @@ -157,9 +170,10 @@ async def chat_stream( that does not happen under the current append-only contract. """ self._validate_conversation_id(conversation_id) + url = f"conversations/{conversation_id}/chatOverStream" request = self._client.build_request( "POST", - f"conversations/{conversation_id}/chatOverStream", + url, json=_chat_body(message, time_zone), ) response = None diff --git a/python/obo/requirements.txt b/python/obo/requirements.txt index 89b4390..7122ad4 100644 --- a/python/obo/requirements.txt +++ b/python/obo/requirements.txt @@ -1,8 +1,5 @@ fastapi>=0.115 uvicorn[standard]>=0.32 httpx>=0.27 -azure-identity>=1.19 +azure-identity>=1.25.3 pyjwt[crypto]>=2.9 - -# Test dependency: starlette.testclient requires httpx2 at runtime. -httpx2>=2.7 diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py index c15fd12..3afa36e 100644 --- a/python/obo/smoke_test.py +++ b/python/obo/smoke_test.py @@ -19,6 +19,8 @@ from fastapi.testclient import TestClient from app.config import get_settings +# _last_text_message is module-private but tested here intentionally to cover +# edge cases (non-dict payloads, missing/null messages) without a full HTTP round trip. from app.workiq import WorkIQClient, WorkIQError, _last_text_message # Ensure env vars set above are picked up, even if config was imported earlier. From ca43d064a025a9ea025a5ce46c38cc944ac42d86 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Sun, 2 Aug 2026 15:11:32 +0400 Subject: [PATCH 28/36] =?UTF-8?q?fix:=20round-2=20review=20=E2=80=94=20str?= =?UTF-8?q?eam=20error=20handling,=20middleware=20hardening,=20timeout=20t?= =?UTF-8?q?uning?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Catch httpx.StreamError alongside HTTPError in chat_stream to prevent 500s - Add proper type annotations on SecurityHeadersMiddleware.dispatch() - Harden body size middleware with streaming byte counter for chunked encoding - Remove cache_keys=True from PyJWKClient to respect JWKS key rotation - Use differentiated httpx.Timeout (connect=10s, read=300s, write=30s, pool=5s) - Add pip-compile --generate-hashes comment for production pinning - Add smoke test coverage for /healthz, security headers, and 413 body limit Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/auth.py | 4 +--- python/obo/app/main.py | 38 +++++++++++++++++++++++++++++++++---- python/obo/app/workiq.py | 7 ++++--- python/obo/requirements.txt | 2 ++ python/obo/smoke_test.py | 29 +++++++++++++++++++++++++++- 5 files changed, 69 insertions(+), 11 deletions(-) diff --git a/python/obo/app/auth.py b/python/obo/app/auth.py index 09952e8..6870e4b 100644 --- a/python/obo/app/auth.py +++ b/python/obo/app/auth.py @@ -34,9 +34,7 @@ class TokenValidator: def __init__(self, settings: Settings) -> None: self._settings = settings - self._jwks_client = PyJWKClient( - settings.jwks_uri, cache_keys=True, lifespan=3600 - ) + self._jwks_client = PyJWKClient(settings.jwks_uri, lifespan=3600) async def validate(self, token: str) -> dict[str, Any]: try: diff --git a/python/obo/app/main.py b/python/obo/app/main.py index 105cac3..e30a602 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -16,7 +16,7 @@ from fastapi import Depends, FastAPI, Header, HTTPException, Request, Response, status from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel, Field -from starlette.middleware.base import BaseHTTPMiddleware +from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint from starlette.types import ASGIApp, Receive, Scope, Send from .auth import InvalidToken, TokenValidator, WorkIQTokenExchange @@ -30,7 +30,11 @@ class _BodySizeLimitMiddleware: - """Reject requests whose Content-Length exceeds the configured cap.""" + """Reject requests whose body exceeds the configured cap. + + Handles both Content-Length (fast reject) and chunked transfer encoding + (streaming byte counter) so the limit cannot be bypassed. + """ def __init__(self, app: ASGIApp, *, max_bytes: int = MAX_REQUEST_BODY_BYTES) -> None: self._app = app @@ -38,6 +42,7 @@ def __init__(self, app: ASGIApp, *, max_bytes: int = MAX_REQUEST_BODY_BYTES) -> async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: if scope["type"] == "http": + # Fast path: reject immediately if Content-Length is declared and oversized. headers = dict(scope.get("headers", [])) length = headers.get(b"content-length") if length is not None and int(length) > self._max_bytes: @@ -47,14 +52,39 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: ) await response(scope, receive, send) return + + # Slow path: count bytes as they arrive (covers chunked encoding). + seen = 0 + rejected = False + + async def limited_receive() -> dict: # type: ignore[type-arg] + nonlocal seen, rejected + message = await receive() + if message.get("type") == "http.request": + seen += len(message.get("body", b"")) + if seen > self._max_bytes: + rejected = True + response = JSONResponse( + {"detail": "Request body too large"}, + status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + ) + await response(scope, receive, send) + return {"type": "http.disconnect"} + return message + + await self._app(scope, limited_receive, send) + return + await self._app(scope, receive, send) class _SecurityHeadersMiddleware(BaseHTTPMiddleware): """Add baseline security headers to every response.""" - async def dispatch(self, request: Request, call_next): # type: ignore[override] - response: Response = await call_next(request) + async def dispatch( + self, request: Request, call_next: RequestResponseEndpoint + ) -> Response: + response = await call_next(request) response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Frame-Options"] = "DENY" response.headers["Cache-Control"] = "no-store" diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index f438680..2e7a586 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -20,7 +20,8 @@ logger = logging.getLogger(__name__) -REQUEST_TIMEOUT_SECONDS = 300.0 +# Streaming reads can take minutes; connect/write/pool should be tight. +REQUEST_TIMEOUT = httpx.Timeout(connect=10.0, read=300.0, write=30.0, pool=5.0) SSE_DATA_PREFIX = "data: " CONV_ID_PATTERN = r"^[a-zA-Z0-9\-_]{1,128}$" _CONV_ID_RE = re.compile(CONV_ID_PATTERN) @@ -104,7 +105,7 @@ def __init__( self._client = httpx.AsyncClient( base_url=base_url, headers={"Authorization": f"Bearer {access_token}"}, - timeout=REQUEST_TIMEOUT_SECONDS, + timeout=REQUEST_TIMEOUT, transport=transport, ) @@ -205,7 +206,7 @@ async def chat_stream( previous = text if delta: yield delta - except httpx.HTTPError as exc: + except (httpx.HTTPError, httpx.StreamError) as exc: raise WorkIQError(f"chat stream failed: {exc}") from exc finally: if response is not None: diff --git a/python/obo/requirements.txt b/python/obo/requirements.txt index 7122ad4..b21a97a 100644 --- a/python/obo/requirements.txt +++ b/python/obo/requirements.txt @@ -1,3 +1,5 @@ +# For production, pin exact versions with hashes: +# pip-compile --generate-hashes requirements.in -o requirements.txt fastapi>=0.115 uvicorn[standard]>=0.32 httpx>=0.27 diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py index 3afa36e..5de6b47 100644 --- a/python/obo/smoke_test.py +++ b/python/obo/smoke_test.py @@ -130,9 +130,36 @@ def transport_error_handler(request: httpx.Request) -> httpx.Response: print("helper edge cases -> ok") # Auth gate: no token and malformed token must both 401 before any Work IQ call. - from app.main import app + from app.main import MAX_REQUEST_BODY_BYTES, app with TestClient(app) as tc: + # -- /healthz (unauthenticated liveness probe) -- + r = tc.get("/healthz") + assert r.status_code == 200, r.status_code + assert r.json() == {"status": "ok"} + print("healthz ->", r.status_code) + + # -- Security response headers -- + assert r.headers["x-content-type-options"] == "nosniff" + assert r.headers["x-frame-options"] == "DENY" + assert r.headers["cache-control"] == "no-store" + assert r.headers["referrer-policy"] == "no-referrer" + print("security headers -> ok") + + # -- Body size limit (413) -- + oversized = b"x" * (MAX_REQUEST_BODY_BYTES + 1) + r = tc.post( + "/api/chat", + content=oversized, + headers={ + "Content-Type": "application/json", + "Content-Length": str(len(oversized)), + }, + ) + assert r.status_code == 413, r.status_code + print("body size limit ->", r.status_code) + + # -- Auth gate -- r = tc.post("/api/chat", json={"message": "hi"}) assert r.status_code == 401, r.status_code print("no bearer token ->", r.status_code, r.json()["detail"]) From d332af51a2c3f2e94df9ddbcd15b4f9e175f9f95 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Sun, 2 Aug 2026 15:23:42 +0400 Subject: [PATCH 29/36] =?UTF-8?q?fix:=20round-3=20review=20=E2=80=94=20dea?= =?UTF-8?q?d=20guard,=20chunked=20test,=20.env=20exclusion,=20log=20saniti?= =?UTF-8?q?sation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix rejected re-entry guard in body size middleware (was write-only dead code) - Guard int(Content-Length) against ValueError for isolated middleware tests - Add chunked-encoding body limit smoke test (slow path coverage) - Exclude .env from git to prevent accidental secret commits - Suppress Bandit B105 false positive on TOKEN_EXCHANGE_SCOPE - Log only exception type name in token rejection to avoid leaking JWT fragments Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/.gitignore | 2 ++ python/obo/app/auth.py | 2 +- python/obo/app/config.py | 2 +- python/obo/app/main.py | 8 +++++++- python/obo/smoke_test.py | 11 ++++++++++- 5 files changed, 21 insertions(+), 4 deletions(-) diff --git a/python/obo/.gitignore b/python/obo/.gitignore index 21d0b89..080dec5 100644 --- a/python/obo/.gitignore +++ b/python/obo/.gitignore @@ -1 +1,3 @@ .venv/ +.env +*.env diff --git a/python/obo/app/auth.py b/python/obo/app/auth.py index 6870e4b..d0c8618 100644 --- a/python/obo/app/auth.py +++ b/python/obo/app/auth.py @@ -50,7 +50,7 @@ async def validate(self, token: str) -> dict[str, Any]: options={"require": ["exp", "aud", "iss"]}, ) except jwt.PyJWTError as exc: - raise InvalidToken(f"token rejected: {exc}") from exc + raise InvalidToken(f"token rejected: {type(exc).__name__}") from exc self._require_scope(claims) return claims diff --git a/python/obo/app/config.py b/python/obo/app/config.py index d54c151..020ce46 100644 --- a/python/obo/app/config.py +++ b/python/obo/app/config.py @@ -33,7 +33,7 @@ # Audience used when exchanging a managed identity token for a client assertion # (workload identity federation). Only used in the secretless configuration. -TOKEN_EXCHANGE_SCOPE = "api://AzureADTokenExchange/.default" +TOKEN_EXCHANGE_SCOPE = "api://AzureADTokenExchange/.default" # nosec B105 — OAuth scope URI, not a secret class ConfigError(RuntimeError): diff --git a/python/obo/app/main.py b/python/obo/app/main.py index e30a602..162931e 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -45,7 +45,11 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: # Fast path: reject immediately if Content-Length is declared and oversized. headers = dict(scope.get("headers", [])) length = headers.get(b"content-length") - if length is not None and int(length) > self._max_bytes: + try: + declared = int(length) if length is not None else 0 + except ValueError: + declared = 0 # unparseable; slow path will enforce the limit + if declared > self._max_bytes: response = JSONResponse( {"detail": "Request body too large"}, status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, @@ -59,6 +63,8 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: async def limited_receive() -> dict: # type: ignore[type-arg] nonlocal seen, rejected + if rejected: + return {"type": "http.disconnect"} message = await receive() if message.get("type") == "http.request": seen += len(message.get("body", b"")) diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py index 5de6b47..e559e75 100644 --- a/python/obo/smoke_test.py +++ b/python/obo/smoke_test.py @@ -157,7 +157,16 @@ def transport_error_handler(request: httpx.Request) -> httpx.Response: }, ) assert r.status_code == 413, r.status_code - print("body size limit ->", r.status_code) + print("body size limit (CL) ->", r.status_code) + + # -- Body size limit without Content-Length (chunked / slow path) -- + r = tc.post( + "/api/chat", + content=oversized, + headers={"Content-Type": "application/json"}, + ) + assert r.status_code in (413, 500), r.status_code # 413 or 500 from disconnect + print("body size limit (chunked)->", r.status_code) # -- Auth gate -- r = tc.post("/api/chat", json={"message": "hi"}) From 7fc44e6be27eae3166580793595da2be41de23fa Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Sun, 2 Aug 2026 15:37:32 +0400 Subject: [PATCH 30/36] =?UTF-8?q?fix:=20round-4=20review=20=E2=80=94=20mid?= =?UTF-8?q?dleware=20protocol=20safety,=20timezone=20init,=20citation=20nu?= =?UTF-8?q?llability?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove JSONResponse from inside limited_receive to avoid ASGI protocol violation when inner app has already started its response - Remove private _last_text_message import from smoke test; edge cases are covered indirectly through the public WorkIQClient API - Move timezone detection from module import to lifespan startup so the fallback warning fires after logging is configured - Make Citation.see_more_web_url and CitationModel.url nullable (str | None) so absent URLs are explicit rather than empty strings Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/main.py | 13 ++++++------- python/obo/app/workiq.py | 22 ++++++++++++---------- python/obo/smoke_test.py | 13 +------------ 3 files changed, 19 insertions(+), 29 deletions(-) diff --git a/python/obo/app/main.py b/python/obo/app/main.py index 162931e..e553d8a 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -21,7 +21,7 @@ from .auth import InvalidToken, TokenValidator, WorkIQTokenExchange from .config import Settings, get_settings -from .workiq import CONV_ID_PATTERN, WorkIQClient, WorkIQError +from .workiq import CONV_ID_PATTERN, WorkIQClient, WorkIQError, init_server_timezone logger = logging.getLogger(__name__) @@ -70,11 +70,9 @@ async def limited_receive() -> dict: # type: ignore[type-arg] seen += len(message.get("body", b"")) if seen > self._max_bytes: rejected = True - response = JSONResponse( - {"detail": "Request body too large"}, - status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, - ) - await response(scope, receive, send) + # Do NOT send a response here — the inner app may have + # already started its response. Returning http.disconnect + # causes the inner app to abort cleanly. return {"type": "http.disconnect"} return message @@ -112,7 +110,7 @@ class CitationModel(BaseModel): type: str source: str provider: str - url: str + url: str | None = None class ChatResponse(BaseModel): @@ -123,6 +121,7 @@ class ChatResponse(BaseModel): @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: + init_server_timezone() settings = get_settings() app.state.settings = settings app.state.validator = TokenValidator(settings) diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index 2e7a586..bacc8b5 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -36,7 +36,7 @@ class Citation: attribution_type: str attribution_source: str provider_display_name: str - see_more_web_url: str + see_more_web_url: str | None = None @dataclass(frozen=True) @@ -45,23 +45,25 @@ class ChatReply: citations: tuple[Citation, ...] = field(default=()) -def _detect_server_timezone() -> str: - """Work IQ requires an IANA timezone (e.g. America/Los_Angeles).""" +# Defaults to UTC; call init_server_timezone() at startup (after logging is +# configured) so the fallback warning is actually visible in application logs. +_SERVER_TIMEZONE: str = "UTC" + + +def init_server_timezone() -> None: + """Detect the server's IANA timezone and cache it for request payloads.""" + global _SERVER_TIMEZONE # noqa: PLW0603 tz = datetime.now().astimezone().tzinfo name = getattr(tz, "key", None) or str(tz) # A bare UTC offset like "+05:30" is not an IANA id; fall back rather than 400. if "/" in name or name == "UTC": - return name + _SERVER_TIMEZONE = name + return logger.warning( "Could not detect IANA timezone (got %r); defaulting to UTC. " "Pass time_zone in requests to override.", name, ) - return "UTC" - - -# Computed once at import — the server timezone cannot change at runtime. -_SERVER_TIMEZONE: str = _detect_server_timezone() def _chat_body(message: str, time_zone: str | None = None) -> dict[str, Any]: @@ -77,7 +79,7 @@ def _parse_citations(message: dict[str, Any]) -> tuple[Citation, ...]: attribution_type=a.get("attributionType", ""), attribution_source=a.get("attributionSource", ""), provider_display_name=a.get("providerDisplayName", ""), - see_more_web_url=a.get("seeMoreWebUrl", ""), + see_more_web_url=a.get("seeMoreWebUrl"), ) for a in (message.get("attributions") or []) if isinstance(a, dict) diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py index e559e75..8227e1c 100644 --- a/python/obo/smoke_test.py +++ b/python/obo/smoke_test.py @@ -19,9 +19,7 @@ from fastapi.testclient import TestClient from app.config import get_settings -# _last_text_message is module-private but tested here intentionally to cover -# edge cases (non-dict payloads, missing/null messages) without a full HTTP round trip. -from app.workiq import WorkIQClient, WorkIQError, _last_text_message +from app.workiq import WorkIQClient, WorkIQError # Ensure env vars set above are picked up, even if config was imported earlier. get_settings.cache_clear() @@ -120,15 +118,6 @@ def transport_error_handler(request: httpx.Request) -> httpx.Response: except WorkIQError as exc: print("transport error (stream) ->", str(exc)[:60]) - # Pure helper — edge cases including non-dict payloads - assert _last_text_message({"messages": []}) is None - assert _last_text_message({"messages": [{"role": "user"}]}) is None - assert _last_text_message([1, 2, 3]) is None # non-dict payload - assert _last_text_message(None) is None # type: ignore[arg-type] - assert _last_text_message({"messages": ["not-a-dict"]}) is None - assert _last_text_message({"messages": None}) is None # messages: null - print("helper edge cases -> ok") - # Auth gate: no token and malformed token must both 401 before any Work IQ call. from app.main import MAX_REQUEST_BODY_BYTES, app From 9ced8f3a17a6d7ac62683ac9b0c35bcb81c535e5 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Sun, 2 Aug 2026 15:46:10 +0400 Subject: [PATCH 31/36] =?UTF-8?q?fix:=20round-5=20review=20=E2=80=94=20rep?= =?UTF-8?q?r=20safety,=20ASGI=20middleware,=20SSE=20docs,=20env=20scoping?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Hide client_secret from Settings repr to prevent accidental credential leaks in logs or tracebacks (repr=False) - Replace BaseHTTPMiddleware with raw ASGI middleware for security headers to avoid response buffering that breaks true SSE streaming - Add OpenAPI responses= annotation documenting the SSE event contract on /api/chat/stream - Add EXTRA_WORKIQ_HOSTS env var (comma-separated) so staging/test hosts can be allowlisted without editing source code - Scope os.environ mutation in smoke_test inside patch.dict so fake credentials don't leak when imported by a test runner - Document print() usage as a deliberate choice in smoke_test docstring Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/config.py | 24 +++++++++++----- python/obo/app/main.py | 60 ++++++++++++++++++++++++++++++---------- python/obo/smoke_test.py | 53 +++++++++++++++++++++-------------- 3 files changed, 96 insertions(+), 41 deletions(-) diff --git a/python/obo/app/config.py b/python/obo/app/config.py index 020ce46..a279322 100644 --- a/python/obo/app/config.py +++ b/python/obo/app/config.py @@ -7,7 +7,7 @@ from __future__ import annotations import os -from dataclasses import dataclass +from dataclasses import dataclass, field from functools import lru_cache # The Work IQ Gateway multiplexes REST/A2A/MCP on one host behind a path prefix @@ -15,9 +15,9 @@ WORKIQ_DEFAULT_HOST = "https://workiq.svc.cloud.microsoft" WORKIQ_PATH = "/rest/beta" -# Allowlist of valid Work IQ Gateway hosts. Expand this set for staging/test -# environments rather than removing the check — an unrestricted WORKIQ_HOST -# env var could redirect OBO tokens to an attacker-controlled endpoint. +# Allowlist of valid Work IQ Gateway hosts. For staging/test environments, +# set EXTRA_WORKIQ_HOSTS (comma-separated HTTPS URLs) rather than removing +# the check — an unrestricted WORKIQ_HOST could redirect OBO tokens. _ALLOWED_WORKIQ_HOSTS = frozenset({ WORKIQ_DEFAULT_HOST, }) @@ -49,7 +49,7 @@ class Settings: api_audience: str required_scope: str workiq_host: str - client_secret: str | None + client_secret: str | None = field(default=None, repr=False) @property def issuer(self) -> str: @@ -76,10 +76,20 @@ def _require(name: str) -> str: return value +def _allowed_hosts() -> frozenset[str]: + """Built-in hosts plus any from the EXTRA_WORKIQ_HOSTS env var.""" + extra = os.environ.get("EXTRA_WORKIQ_HOSTS", "").strip() + if not extra: + return _ALLOWED_WORKIQ_HOSTS + additions = frozenset(h.strip() for h in extra.split(",") if h.strip()) + return _ALLOWED_WORKIQ_HOSTS | additions + + def _validated_workiq_host(host: str) -> str: - if host not in _ALLOWED_WORKIQ_HOSTS: + allowed = _allowed_hosts() + if host not in allowed: raise ConfigError( - f"WORKIQ_HOST {host!r} is not in the allowed list: {_ALLOWED_WORKIQ_HOSTS}" + f"WORKIQ_HOST {host!r} is not in the allowed list: {allowed}" ) return host diff --git a/python/obo/app/main.py b/python/obo/app/main.py index e553d8a..5565fb0 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -13,10 +13,9 @@ from typing import Annotated, AsyncIterator from azure.core.exceptions import ClientAuthenticationError, HttpResponseError -from fastapi import Depends, FastAPI, Header, HTTPException, Request, Response, status +from fastapi import Depends, FastAPI, Header, HTTPException, Request, status from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel, Field -from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint from starlette.types import ASGIApp, Receive, Scope, Send from .auth import InvalidToken, TokenValidator, WorkIQTokenExchange @@ -82,18 +81,37 @@ async def limited_receive() -> dict: # type: ignore[type-arg] await self._app(scope, receive, send) -class _SecurityHeadersMiddleware(BaseHTTPMiddleware): - """Add baseline security headers to every response.""" +_SECURITY_HEADERS: list[tuple[bytes, bytes]] = [ + (b"x-content-type-options", b"nosniff"), + (b"x-frame-options", b"DENY"), + (b"cache-control", b"no-store"), + (b"referrer-policy", b"no-referrer"), +] - async def dispatch( - self, request: Request, call_next: RequestResponseEndpoint - ) -> Response: - response = await call_next(request) - response.headers["X-Content-Type-Options"] = "nosniff" - response.headers["X-Frame-Options"] = "DENY" - response.headers["Cache-Control"] = "no-store" - response.headers["Referrer-Policy"] = "no-referrer" - return response + +class _SecurityHeadersMiddleware: + """Add baseline security headers to every response. + + Implemented as a raw ASGI middleware (not BaseHTTPMiddleware) to avoid + response buffering that would break true streaming on SSE endpoints. + """ + + def __init__(self, app: ASGIApp) -> None: + self._app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http": + await self._app(scope, receive, send) + return + + async def send_with_headers(message: dict) -> None: # type: ignore[type-arg] + if message["type"] == "http.response.start": + headers = list(message.get("headers", [])) + headers.extend(_SECURITY_HEADERS) + message = {**message, "headers": headers} + await send(message) + + await self._app(scope, receive, send_with_headers) class ChatRequest(BaseModel): @@ -234,7 +252,21 @@ async def chat( ) -@app.post("/api/chat/stream") +@app.post( + "/api/chat/stream", + responses={ + 200: { + "description": ( + "SSE stream. Events in order:\n" + "1. `event: conversation` — `data: {\"conversation_id\": \"...\"}`\n" + "2. (repeated) default event — `data: {\"text\": \"\"}`\n" + "3. `event: done` — signals clean completion, OR\n" + " `event: error` — `data: upstream request failed`" + ), + "content": {"text/event-stream": {}}, + }, + }, +) async def chat_stream( request: ChatRequest, token: Annotated[str, Depends(workiq_token)], diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py index 8227e1c..afde798 100644 --- a/python/obo/smoke_test.py +++ b/python/obo/smoke_test.py @@ -1,33 +1,34 @@ -"""Smoke test: fakes the Work IQ gateway with httpx.MockTransport.""" +"""Smoke test: fakes the Work IQ gateway with httpx.MockTransport. + +Diagnostic output uses print() deliberately — this is a standalone script, +not a pytest suite, and structured logging adds no value for manual runs. +""" import asyncio import json import os import sys from pathlib import Path - -os.environ.update( - AZURE_TENANT_ID="11111111-1111-1111-1111-111111111111", - AZURE_CLIENT_ID="22222222-2222-2222-2222-222222222222", - API_AUDIENCE="api://22222222-2222-2222-2222-222222222222", - AZURE_CLIENT_SECRET="local-dev-secret", -) +from unittest.mock import patch + +# Fake credentials — must be set before any app.* imports so get_settings() +# picks them up. Scoped via patch.dict so they don't leak into the process +# environment if this module is ever imported by a test runner. +_TEST_ENV = { + "AZURE_TENANT_ID": "11111111-1111-1111-1111-111111111111", + "AZURE_CLIENT_ID": "22222222-2222-2222-2222-222222222222", + "API_AUDIENCE": "api://22222222-2222-2222-2222-222222222222", + "AZURE_CLIENT_SECRET": "local-dev-secret", +} sys.path.insert(0, str(Path(__file__).parent)) -import httpx -from fastapi.testclient import TestClient - -from app.config import get_settings -from app.workiq import WorkIQClient, WorkIQError - -# Ensure env vars set above are picked up, even if config was imported earlier. -get_settings.cache_clear() - BASE = "https://workiq.test/rest/beta/" -def handler(request: httpx.Request) -> httpx.Response: +def handler(request: "httpx.Request") -> "httpx.Response": + import httpx + path = request.url.path if path.endswith("/conversations"): return httpx.Response(200, json={"id": "conv-42"}) @@ -61,11 +62,22 @@ def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(500, json={"error": "unexpected path"}) -def error_handler(request: httpx.Request) -> httpx.Response: +def error_handler(request: "httpx.Request") -> "httpx.Response": + import httpx + return httpx.Response(403, json={"error": "no copilot license"}, headers={"request-id": "abc-123"}) async def main() -> None: + import httpx + from fastapi.testclient import TestClient + + from app.config import get_settings + from app.workiq import WorkIQClient, WorkIQError + + # Ensure env vars are picked up by config. + get_settings.cache_clear() + transport = httpx.MockTransport(handler) async with WorkIQClient("fake-token", BASE, transport=transport) as client: @@ -181,4 +193,5 @@ def transport_error_handler(request: httpx.Request) -> httpx.Response: if __name__ == "__main__": - asyncio.run(main()) + with patch.dict(os.environ, _TEST_ENV): + asyncio.run(main()) From 05e86ca463fd2c3a6cf54e879967d8831b690e66 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Sun, 2 Aug 2026 15:58:45 +0400 Subject: [PATCH 32/36] fix: enforce https:// scheme on EXTRA_WORKIQ_HOSTS entries Reject non-HTTPS entries at startup to prevent OBO tokens from being forwarded over plaintext HTTP to misconfigured or malicious hosts. Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/config.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/python/obo/app/config.py b/python/obo/app/config.py index a279322..9156cdf 100644 --- a/python/obo/app/config.py +++ b/python/obo/app/config.py @@ -81,8 +81,17 @@ def _allowed_hosts() -> frozenset[str]: extra = os.environ.get("EXTRA_WORKIQ_HOSTS", "").strip() if not extra: return _ALLOWED_WORKIQ_HOSTS - additions = frozenset(h.strip() for h in extra.split(",") if h.strip()) - return _ALLOWED_WORKIQ_HOSTS | additions + additions: set[str] = set() + for raw in extra.split(","): + host = raw.strip() + if not host: + continue + if not host.startswith("https://"): + raise ConfigError( + f"EXTRA_WORKIQ_HOSTS entry {host!r} must use the https:// scheme" + ) + additions.add(host) + return _ALLOWED_WORKIQ_HOSTS | frozenset(additions) def _validated_workiq_host(host: str) -> str: From 5492f3578a27a3d386c0cde95abd6d9036bf5e83 Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Sun, 2 Aug 2026 16:19:01 +0400 Subject: [PATCH 33/36] fix: send 413 when response not yet started, normalize WORKIQ_HOST trailing slash - Body size middleware now tracks whether the inner app has started its response; sends a proper 413 when it hasn't, disconnects when it has - Normalize WORKIQ_HOST by stripping trailing slashes before allowlist check so "https://host/" matches "https://host" Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/config.py | 6 ++++-- python/obo/app/main.py | 20 ++++++++++++++++---- 2 files changed, 20 insertions(+), 6 deletions(-) diff --git a/python/obo/app/config.py b/python/obo/app/config.py index 9156cdf..70c196d 100644 --- a/python/obo/app/config.py +++ b/python/obo/app/config.py @@ -95,12 +95,14 @@ def _allowed_hosts() -> frozenset[str]: def _validated_workiq_host(host: str) -> str: + # Normalize trailing slashes so "https://host/" matches "https://host". + normalized = host.rstrip("/") allowed = _allowed_hosts() - if host not in allowed: + if normalized not in allowed: raise ConfigError( f"WORKIQ_HOST {host!r} is not in the allowed list: {allowed}" ) - return host + return normalized @lru_cache(maxsize=1) diff --git a/python/obo/app/main.py b/python/obo/app/main.py index 5565fb0..6dcac5f 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -59,6 +59,13 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: # Slow path: count bytes as they arrive (covers chunked encoding). seen = 0 rejected = False + response_started = False + + async def send_wrapper(message: dict) -> None: # type: ignore[type-arg] + nonlocal response_started + if message["type"] == "http.response.start": + response_started = True + await send(message) async def limited_receive() -> dict: # type: ignore[type-arg] nonlocal seen, rejected @@ -69,13 +76,18 @@ async def limited_receive() -> dict: # type: ignore[type-arg] seen += len(message.get("body", b"")) if seen > self._max_bytes: rejected = True - # Do NOT send a response here — the inner app may have - # already started its response. Returning http.disconnect - # causes the inner app to abort cleanly. + if not response_started: + # Safe to send a proper 413 — the app hasn't + # started its response yet. + err = JSONResponse( + {"detail": "Request body too large"}, + status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + ) + await err(scope, receive, send) return {"type": "http.disconnect"} return message - await self._app(scope, limited_receive, send) + await self._app(scope, limited_receive, send_wrapper) return await self._app(scope, receive, send) From 9dcb48e6ae6fa7ac516aa358542db8d27dfee04b Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Sun, 2 Aug 2026 16:30:38 +0400 Subject: [PATCH 34/36] fix: deterministic 413 for chunked bodies, correct middleware ordering - Return clean end-of-body (not http.disconnect) after rejection so FastAPI doesn't raise request-parsing errors - Suppress downstream response writes after sending 413 to prevent the inner app from corrupting the response - Swap middleware order: body-size outermost, security headers inner, so 413 rejections also carry security headers - Smoke test now requires 413 deterministically (not 413-or-500) Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/main.py | 20 ++++++++++++++------ python/obo/smoke_test.py | 2 +- 2 files changed, 15 insertions(+), 7 deletions(-) diff --git a/python/obo/app/main.py b/python/obo/app/main.py index 6dcac5f..e22c889 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -60,31 +60,37 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: seen = 0 rejected = False response_started = False + error_sent = False async def send_wrapper(message: dict) -> None: # type: ignore[type-arg] nonlocal response_started + # Once we've sent our own 413, suppress any downstream writes + # so the inner app can't corrupt the response. + if error_sent: + return if message["type"] == "http.response.start": response_started = True await send(message) async def limited_receive() -> dict: # type: ignore[type-arg] - nonlocal seen, rejected + nonlocal seen, rejected, error_sent if rejected: - return {"type": "http.disconnect"} + # Return a clean end-of-body instead of disconnect so + # FastAPI doesn't raise request-parsing errors. + return {"type": "http.request", "body": b"", "more_body": False} message = await receive() if message.get("type") == "http.request": seen += len(message.get("body", b"")) if seen > self._max_bytes: rejected = True if not response_started: - # Safe to send a proper 413 — the app hasn't - # started its response yet. + error_sent = True err = JSONResponse( {"detail": "Request body too large"}, status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, ) await err(scope, receive, send) - return {"type": "http.disconnect"} + return {"type": "http.request", "body": b"", "more_body": False} return message await self._app(scope, limited_receive, send_wrapper) @@ -168,8 +174,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: app = FastAPI(title="Work IQ OBO Backend", lifespan=lifespan) -app.add_middleware(_SecurityHeadersMiddleware) +# Body-size runs outermost (added last = LIFO), security headers wraps the +# inner app so 413 rejections also carry the security headers. app.add_middleware(_BodySizeLimitMiddleware) +app.add_middleware(_SecurityHeadersMiddleware) @app.get("/healthz") diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py index afde798..4da4357 100644 --- a/python/obo/smoke_test.py +++ b/python/obo/smoke_test.py @@ -166,7 +166,7 @@ def transport_error_handler(request: httpx.Request) -> httpx.Response: content=oversized, headers={"Content-Type": "application/json"}, ) - assert r.status_code in (413, 500), r.status_code # 413 or 500 from disconnect + assert r.status_code == 413, r.status_code print("body size limit (chunked)->", r.status_code) # -- Auth gate -- From ab9d112e4a3f7ff941d23147fa7529ee9aab92cc Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Sun, 2 Aug 2026 17:08:53 +0400 Subject: [PATCH 35/36] fix: Python 3.10 compat, middleware comment, SSE done frame, nbf enforcement - Guard typing.Self import with fallback to typing_extensions for Python 3.10 - Fix middleware ordering comment to match LIFO reality - Emit empty data line in done SSE frame (no trailing space) - Require nbf claim in JWT validation for defense-in-depth - Update README error event text to match code (upstream request failed) Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/README.md | 2 +- python/obo/app/auth.py | 2 +- python/obo/app/main.py | 6 +++--- python/obo/app/workiq.py | 7 ++++++- 4 files changed, 11 insertions(+), 6 deletions(-) diff --git a/python/obo/README.md b/python/obo/README.md index b3643ec..ba52375 100644 --- a/python/obo/README.md +++ b/python/obo/README.md @@ -137,7 +137,7 @@ no network. 1. `event: conversation` — `data: {"conversation_id": "..."}` (first frame) 2. `data: {"text": ""}` — one per text chunk (default event type) -3. `event: done` — signals clean completion, or `event: error` — `data: Work IQ request failed` on failure +3. `event: done` — signals clean completion, or `event: error` — `data: upstream request failed` on failure Request body for both: `{"message": "...", "conversation_id": "...", "time_zone": "..."}` (`conversation_id` and `time_zone` optional; `time_zone` is an IANA identifier like `America/New_York`). diff --git a/python/obo/app/auth.py b/python/obo/app/auth.py index d0c8618..e5c9d2f 100644 --- a/python/obo/app/auth.py +++ b/python/obo/app/auth.py @@ -47,7 +47,7 @@ async def validate(self, token: str) -> dict[str, Any]: algorithms=["RS256"], audience=self._settings.api_audience, issuer=self._settings.issuer, - options={"require": ["exp", "aud", "iss"]}, + options={"require": ["exp", "nbf", "aud", "iss"]}, ) except jwt.PyJWTError as exc: raise InvalidToken(f"token rejected: {type(exc).__name__}") from exc diff --git a/python/obo/app/main.py b/python/obo/app/main.py index e22c889..8f8b436 100644 --- a/python/obo/app/main.py +++ b/python/obo/app/main.py @@ -174,8 +174,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: app = FastAPI(title="Work IQ OBO Backend", lifespan=lifespan) -# Body-size runs outermost (added last = LIFO), security headers wraps the -# inner app so 413 rejections also carry the security headers. +# Starlette add_middleware is LIFO: last-added is outermost. +# SecurityHeaders is outermost so 413s from BodySize also carry the headers. app.add_middleware(_BodySizeLimitMiddleware) app.add_middleware(_SecurityHeadersMiddleware) @@ -307,7 +307,7 @@ async def events() -> AsyncGenerator[str, None]: conversation_id, request.message, time_zone=request.time_zone ): yield f"data: {json.dumps({'text': delta})}\n\n" - yield "event: done\ndata: \n\n" + yield "event: done\ndata:\n\n" except WorkIQError as exc: # The HTTP 200 is already committed, so the error rides the stream. # Clients must handle "error" events to detect mid-stream failures. diff --git a/python/obo/app/workiq.py b/python/obo/app/workiq.py index bacc8b5..62a4aac 100644 --- a/python/obo/app/workiq.py +++ b/python/obo/app/workiq.py @@ -14,7 +14,12 @@ from collections.abc import AsyncGenerator from dataclasses import dataclass, field from datetime import datetime -from typing import Any, Self +from typing import Any + +try: + from typing import Self +except ImportError: # Python 3.10 + from typing_extensions import Self import httpx From b376ac1a5ea0794702f280ce6f32375bc969b16b Mon Sep 17 00:00:00 2001 From: Muhammad Shakir Fattani Date: Sun, 2 Aug 2026 23:01:50 +0400 Subject: [PATCH 36/36] fix: normalize trailing slash on EXTRA_WORKIQ_HOSTS entries _validated_workiq_host() strips trailing slashes from WORKIQ_HOST before comparing against the allowlist, but _allowed_hosts() stored EXTRA_WORKIQ_HOSTS entries as-is. An entry like "https://workiq.test/" would never match the normalized form, causing a false ConfigError. Co-Authored-By: Shakir's Advisor Claude(Opus 4.6 (1M context)) --- python/obo/app/config.py | 2 +- python/obo/smoke_test.py | 14 ++++++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/python/obo/app/config.py b/python/obo/app/config.py index 70c196d..8ae95a4 100644 --- a/python/obo/app/config.py +++ b/python/obo/app/config.py @@ -90,7 +90,7 @@ def _allowed_hosts() -> frozenset[str]: raise ConfigError( f"EXTRA_WORKIQ_HOSTS entry {host!r} must use the https:// scheme" ) - additions.add(host) + additions.add(host.rstrip("/")) return _ALLOWED_WORKIQ_HOSTS | frozenset(additions) diff --git a/python/obo/smoke_test.py b/python/obo/smoke_test.py index 4da4357..5bf98fd 100644 --- a/python/obo/smoke_test.py +++ b/python/obo/smoke_test.py @@ -189,6 +189,20 @@ def transport_error_handler(request: httpx.Request) -> httpx.Response: assert r.status_code == 401, r.status_code print("auth precedes validation ->", r.status_code) + # Config: EXTRA_WORKIQ_HOSTS entries with trailing slashes must match + # after _validated_workiq_host normalizes WORKIQ_HOST. + from app.config import _allowed_hosts, _validated_workiq_host, get_settings + + get_settings.cache_clear() + with patch.dict(os.environ, {**_TEST_ENV, "EXTRA_WORKIQ_HOSTS": "https://workiq.test/"}): + get_settings.cache_clear() + hosts = _allowed_hosts() + assert "https://workiq.test" in hosts, f"trailing slash not normalized: {hosts}" + result = _validated_workiq_host("https://workiq.test/") + assert result == "https://workiq.test", result + print("extra host trailing slash -> ok") + get_settings.cache_clear() + print("\nAll checks passed.")