From 80b85204cd557ec033153363bd15d690809e44fc Mon Sep 17 00:00:00 2001 From: Michael Chou Date: Tue, 25 Aug 2026 13:41:48 -0700 Subject: [PATCH 1/3] feat(agentex): run Slack turns as the invoking user, and the flow to link them Part of the event-driven agents work. Builds on the identity-link storage from #409, which nothing consumed until now. Before this, every Slack-originated turn ran as a single shared bot service account, so the agent acted with one fixed identity no matter who asked. Its tools could only ever reach whatever that bot could reach. Now, when the person who sent the message has linked their account, the turn runs as them and their own connected integrations (Notion, Linear, ...) resolve. Two pieces: 1. Gateway wiring. `_turn_identity()` resolves the Slack (team, user) to a link and returns that user's principal plus acting headers; absent a link it falls back to the existing bot behavior, so unlinked users are unaffected. Task naming becomes per-user (`slack:{thread_ts}:{sgp_user_id}`) for linked users so two people in one thread don't share a task; unlinked users keep the legacy key. `slack_user_id` / `sgp_user_id` land in task_metadata for attribution. 2. The link flow. The mapping cannot be derived, since nothing in a webhook says anything about SGP, so it is established in the one moment where both identities are authenticated at once: - Slack's HMAC proves the Slack side; that verified identity is parked in a single-use, short-TTL nonce (`link_nonce_service`). - The user clicks through to `/integrations/slack/link`, which is deliberately NOT auth-whitelisted (unlike `/slack`), so the auth middleware turns their own browser session into the SGP side. - Both halves now present in one request: a confirmation page names each identity, then POST mints a key as that user, encrypts and stores it, and burns the nonce. The nonce matters: with ids in the query string an attacker could bind someone else's chat identity to their own SGP account by editing the URL, after which the victim's messages would run as the attacker. An opaque token means nothing in the URL is forgeable. Notes on the pieces that are easy to get wrong: - Keys must come from identity-service, not egp-api-backend. Both mint API keys and only the former's are accepted by sgp-secrets; the latter authenticate fine against their own issuer and then fail at the vault with an opaque 401. A stored key of the wrong kind yields a link that looks healthy and silently reads nothing, so the client refuses anything not `ssk_`-shaped. - Minting requires the user's *cookie*. `POST /api-keys` is guarded by a JWT cookie guard that rejects `x-api-key`, which is why the browser leg is load-bearing and cannot be scripted away. - The identity-service URL is read from the environment with no default. It is deployment-specific, and a wrong default would POST a user's forwarded session credentials at whatever answers. - `acting_headers()` is never cached, while resolution is (with a shorter TTL for negative results). Caching a credential is how you serve a revoked one. - Every "cannot act as them" case returns None rather than a partial identity, so the gateway falls back explicitly instead of half-assuming a user. Not wired up yet: nothing sends the user a link. On an unlinked mention the gateway still runs as the bot rather than offering to connect, so the flow is only reachable via scripts/dev_seed_link_nonce.py. The DM trigger (plus its rate-limiting and replay of the stashed turn) is the next change. The live mint is also unverified: it needs a real browser session against a deployed host, which no test can stand in for. Everything up to the mint, and everything after it, is covered. Testing: 69 new unit tests -- 58 across four new files (link service, nonce service, identity-service client, routes) and 11 added to the gateway suite. 139 pass across the five touched test files; the full unit suite passes. Local integration tests error at fixture setup for want of a working Docker socket, which reproduces identically on a clean checkout of main. Co-Authored-By: Claude Opus 5 (1M context) --- agentex/openapi.yaml | 68 +++++ agentex/scripts/dev_seed_link_nonce.py | 134 +++++++++ .../src/adapters/identity_service/__init__.py | 0 .../adapter_identity_service.py | 195 +++++++++++++ agentex/src/api/app.py | 5 + agentex/src/api/routes/integrations.py | 275 ++++++++++++++++++ .../domain/services/identity_link_service.py | 248 ++++++++++++++++ .../src/domain/services/link_nonce_service.py | 166 +++++++++++ .../use_cases/slack_gateway_use_case.py | 196 +++++++++++-- .../adapters/test_identity_service_adapter.py | 199 +++++++++++++ .../unit/api/test_integrations_routes.py | 211 ++++++++++++++ .../services/test_identity_link_service.py | 238 +++++++++++++++ .../unit/services/test_link_nonce_service.py | 155 ++++++++++ .../use_cases/test_slack_gateway_use_case.py | 213 ++++++++++++++ 14 files changed, 2286 insertions(+), 17 deletions(-) create mode 100755 agentex/scripts/dev_seed_link_nonce.py create mode 100644 agentex/src/adapters/identity_service/__init__.py create mode 100644 agentex/src/adapters/identity_service/adapter_identity_service.py create mode 100644 agentex/src/api/routes/integrations.py create mode 100644 agentex/src/domain/services/identity_link_service.py create mode 100644 agentex/src/domain/services/link_nonce_service.py create mode 100644 agentex/tests/unit/adapters/test_identity_service_adapter.py create mode 100644 agentex/tests/unit/api/test_integrations_routes.py create mode 100644 agentex/tests/unit/services/test_identity_link_service.py create mode 100644 agentex/tests/unit/services/test_link_nonce_service.py diff --git a/agentex/openapi.yaml b/agentex/openapi.yaml index b62ae7a1..c73d2544 100644 --- a/agentex/openapi.yaml +++ b/agentex/openapi.yaml @@ -3015,6 +3015,66 @@ paths: additionalProperties: true type: object title: Response Linear Events Linear Events Post + /integrations/slack/link: + get: + tags: + - Integrations + summary: Confirm linking a Slack identity to SGP + description: 'Render the confirmation screen. Does NOT consume the nonce, so + a refresh or + + a link-prefetching browser doesn''t break the flow.' + operationId: slack_link_page_integrations_slack_link_get + parameters: + - name: nonce + in: query + required: false + schema: + type: string + default: '' + title: Nonce + responses: + '200': + description: Successful Response + content: + application/json: + schema: {} + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' + post: + tags: + - Integrations + summary: Complete a Slack identity link + description: 'Mint a key as the signed-in user and store the mapping. + + + Order matters: the nonce is consumed only after a successful mint, so a + + transient identity-service failure leaves the link clickable instead of + + burning it and forcing the user back to Slack.' + operationId: slack_link_confirm_integrations_slack_link_post + requestBody: + content: + application/x-www-form-urlencoded: + schema: + $ref: '#/components/schemas/Body_slack_link_confirm_integrations_slack_link_post' + responses: + '200': + description: Successful Response + content: + application/json: + schema: {} + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /deployment-history/{deployment_id}: get: tags: @@ -4728,6 +4788,14 @@ components: - version - type title: BlobResponse + Body_slack_link_confirm_integrations_slack_link_post: + properties: + nonce: + type: string + title: Nonce + default: '' + type: object + title: Body_slack_link_confirm_integrations_slack_link_post CancelTaskRequest: properties: task_id: diff --git a/agentex/scripts/dev_seed_link_nonce.py b/agentex/scripts/dev_seed_link_nonce.py new file mode 100755 index 00000000..1f4057a0 --- /dev/null +++ b/agentex/scripts/dev_seed_link_nonce.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python3 +"""DEV ONLY: seed a link nonce and print the URL to click. + +Exists because three steps of the identity-link flow can only be exercised by a real +browser, and no script can stand in for them: + + 1. Does the SGP session cookie actually reach agentex? The cookie is scoped to a + parent domain, so it only travels if the agentex host is a sibling subdomain of + the SGP host — worth confirming rather than assuming. + 2. Does agentex's auth middleware turn that cookie into a principal? + 3. Will identity-service mint a key for it? ``POST /api-keys`` is guarded by + ``CustomerIdentityJwtGuard``, which reads ``_identityJwt`` / ``_jwt`` **cookies** + and rejects ``x-api-key`` outright — so an API key cannot substitute. + +This writes a nonce exactly as the Slack leg would (same service, same payload +shape), then prints the link. Clicking it runs the genuine callback: confirmation +page, mint, encrypt, store. + + # against a locally running API + ./scripts/dev_seed_link_nonce.py --slack-user U01ABCDEF --team T01EXAMPLE + + # against a deployed API (your browser's session cookie must cover that host) + ./scripts/dev_seed_link_nonce.py --slack-user U01ABCDEF --team T01EXAMPLE \ + --base-url https:// + +Requires REDIS_URL (the nonce store the callback will read from). Seeds nothing +sensitive: a nonce holds provider ids and the pending message, never a credential. +""" + +from __future__ import annotations + +import argparse +import asyncio +import os +import sys +from pathlib import Path +from urllib.parse import urlencode + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from src.domain.services.link_nonce_service import ( # noqa: E402 + LinkNonceService, + LinkRequest, +) + + +async def main() -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n")[0]) + parser.add_argument("--slack-user", required=True, help="Slack member ID (U…)") + parser.add_argument("--team", required=True, help="Slack team ID (T…)") + parser.add_argument( + "--display-name", + default="", + help="Shown on the confirmation screen; defaults to the Slack user id", + ) + parser.add_argument( + "--message", + default="what's in my notion?", + help="Pending turn stashed on the nonce (replayed after linking)", + ) + parser.add_argument( + "--base-url", + default=os.getenv("AGENTEX_BASE_URL", "http://localhost:5003"), + help="Where the agentex API is reachable from your browser", + ) + parser.add_argument( + "--redis-url", + default=os.getenv("REDIS_URL", "redis://localhost:6379"), + help="Nonce store the callback will read (must match the API's REDIS_URL)", + ) + args = parser.parse_args() + + try: + import redis.asyncio as redis + except ImportError: + print("redis package not available", file=sys.stderr) + return 1 + + client = redis.Redis.from_url(args.redis_url) + try: + await client.ping() + except Exception as exc: # noqa: BLE001 + print(f"cannot reach Redis at {args.redis_url}: {exc}", file=sys.stderr) + print( + "start the dev stack first (./dev.sh), or pass --redis-url", file=sys.stderr + ) + return 1 + + token = await LinkNonceService(redis_client=client).create( + LinkRequest( + provider="slack", + external_team_id=args.team, + external_user_id=args.slack_user, + display_name=args.display_name or args.slack_user, + pending_turn={ + "team_id": args.team, + "channel": "C_DEV", + "user": args.slack_user, + "text": args.message, + "thread_ts": "1700000000.000100", + "selector": None, + }, + ) + ) + await client.aclose() + + url = f"{args.base_url.rstrip('/')}/integrations/slack/link?{urlencode({'nonce': token})}" + print() + print("Open this in a browser that is SIGNED IN to SGP:") + print() + print(f" {url}") + print() + print("What to look for:") + print(" * the confirmation page naming BOTH identities -> cookie reached agentex") + print(" and the middleware resolved a principal") + print( + " * after Connect: 'You're connected' -> identity-service minted" + ) + print(" an ssk_ key and it is stored encrypted") + print(" * 'Please sign in to SGP' (401) -> the cookie did NOT") + print(" arrive; check the cookie domain against this base-url's host") + print() + print("Then verify what landed:") + print( + ' psql "$DATABASE_URL" -c "SELECT external_user_id, sgp_user_id, linked_via, ' + "credential_expires_at, (credential_ciphertext IS NOT NULL) AS has_key " + 'FROM identity_links WHERE revoked_at IS NULL;"' + ) + print() + return 0 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/agentex/src/adapters/identity_service/__init__.py b/agentex/src/adapters/identity_service/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/agentex/src/adapters/identity_service/adapter_identity_service.py b/agentex/src/adapters/identity_service/adapter_identity_service.py new file mode 100644 index 00000000..5b4da1e2 --- /dev/null +++ b/agentex/src/adapters/identity_service/adapter_identity_service.py @@ -0,0 +1,195 @@ +"""Mint SGP API keys via identity-service, acting as the user who asked. + +Why this adapter exists at all: an event-driven turn has to act as the invoking +human to read that person's connected integrations, and the secrets service derives +the owner from the caller — there is no "fetch on behalf of" parameter. So the +gateway has to hold a credential belonging to that user, and this is where it comes +from. + +Why identity-service and not egp-api-backend's ``/v5/api-keys``: there are two +independent API-key issuers on the platform and only one is accepted by the secrets +service. + + identity-service POST /api-keys -> ``ssk_is_<32 hex>`` ACCEPTED + egp-api-backend POST /v5/api-keys -> ``sk_<...>`` 401 at sgp-secrets + +That was established empirically: a key minted from egp-api-backend for the correct +user, which authenticated fine against egp-api-backend itself, was rejected by +sgp-secrets with ``INVALID_API_KEY``. The discriminator is the issuer, not the +identity. Mint from the wrong one and everything looks right until the first secret +read fails. + +Note the field names are camelCase here (``identityId``, ``identityType``, +``expiresOn``) — egp-api-backend's equivalent endpoint uses snake_case, so the two +are easy to confuse. + +Auth: the caller's own credentials are forwarded, so the user mints their *own* key. +identity-service permits exactly that — ``assertCanManageTarget`` short-circuits on +"users may always manage their own keys" — with no admin role needed. We deliberately +do NOT hold a privileged key-minting credential: one that could mint for any user +would be strictly more dangerous than the per-user keys it would produce. +""" + +from __future__ import annotations + +import os +from datetime import datetime +from typing import Any + +import httpx + +from src.utils.logging import make_logger + +logger = make_logger(__name__) + +# Read from the environment with no default on purpose. The address is +# deployment-specific and the service is cluster-internal (no public route), so a +# baked-in default would be wrong in most environments — and a *wrong* default here +# would POST a user's session credentials at whatever answers. +IDENTITY_SERVICE_URL_ENV = "IDENTITY_SERVICE_URL" + +# Credential headers worth forwarding so identity-service sees the *user*. Mirrors +# what agentex-auth itself forwards, plus the bearer form the SGP UI uses when +# OneAuth is on. Nothing else is passed through — notably not cookies beyond the +# session ones, and never the gateway's own bot key. +_FORWARDABLE = ("cookie", "authorization", "x-api-key", "x-selected-account-id") + +_TIMEOUT_S = float(os.getenv("IDENTITY_SERVICE_TIMEOUT_S", "15")) + + +class IdentityServiceError(RuntimeError): + """A key could not be minted. Carries a user-safe reason.""" + + def __init__(self, message: str, *, status: int | None = None): + super().__init__(message) + self.status = status + + +def base_url() -> str: + """Identity-service base URL from the environment. + + Raises rather than falling back, because the alternative is sending a user's + forwarded session credentials to an unintended host. + """ + raw = os.getenv(IDENTITY_SERVICE_URL_ENV, "").strip() + if not raw: + raise IdentityServiceError( + f"{IDENTITY_SERVICE_URL_ENV} is not configured on this deployment." + ) + return raw.rstrip("/") + + +def forwardable_headers(headers: dict[str, str]) -> dict[str, str]: + """The subset of an inbound request's headers that identify the caller.""" + lowered = {k.lower(): v for k, v in headers.items()} + return {k: lowered[k] for k in _FORWARDABLE if lowered.get(k)} + + +class IdentityServiceClient: + """Thin client. Only the one operation the link flow needs.""" + + def __init__(self, url: str | None = None): + self.url = (url or base_url()).rstrip("/") + + async def mint_user_api_key( + self, + *, + sgp_user_id: str, + name: str, + auth_headers: dict[str, str], + expires_on: datetime | None = None, + ) -> tuple[str, datetime | None]: + """Mint an ``ssk_is_`` key owned by ``sgp_user_id``. + + ``auth_headers`` must authenticate AS that user — the caller's own session + from the link callback. Returns ``(secret, expires_on)``; the secret is + returned exactly once by identity-service and is never retrievable again, + so the caller must persist it before doing anything else that can fail. + + ``expires_on`` bounds the credential. Strongly recommended: the stored key + is the most sensitive thing agentex holds, and an expiry converts "valid + until someone notices" into a known window. + """ + if not sgp_user_id: + raise IdentityServiceError("cannot mint a key without a user id") + payload: dict[str, Any] = { + "name": name, + "identityId": sgp_user_id, + "identityType": "user", + } + if expires_on is not None: + # NestJS class-validator @IsDate with a transform accepts ISO-8601. + payload["expiresOn"] = expires_on.isoformat() + + try: + async with httpx.AsyncClient(timeout=_TIMEOUT_S) as client: + resp = await client.post( + f"{self.url}/api-keys", + json=payload, + headers={**auth_headers, "content-type": "application/json"}, + ) + except Exception as exc: # noqa: BLE001 - network/DNS; surfaced to the user + logger.warning("[identity-service] mint request failed", exc_info=True) + raise IdentityServiceError( + "Couldn't reach the identity service to create a key." + ) from exc + + if resp.status_code in (401, 403): + # The session didn't carry through, or it isn't this user's own key. + raise IdentityServiceError( + "Your session wasn't accepted when creating the key. " + "Sign in to SGP and try the link again.", + status=resp.status_code, + ) + if resp.status_code >= 400: + detail = (resp.text or "")[:200] + logger.warning( + "[identity-service] mint failed (%s): %s", resp.status_code, detail + ) + raise IdentityServiceError( + f"The identity service rejected the request (HTTP {resp.status_code}).", + status=resp.status_code, + ) + + try: + body = resp.json() + except Exception as exc: # noqa: BLE001 + raise IdentityServiceError( + "Unreadable response from identity service." + ) from exc + + secret = body.get("secret") + if not secret: + # The secret is only present on create/rotate. Its absence means we got + # a record back but no usable credential — nothing to store. + raise IdentityServiceError( + "The identity service created a key but returned no secret." + ) + if not str(secret).startswith("ssk_"): + # Guards against the wrong-issuer mistake this module exists to avoid: + # a non-ssk_ key will authenticate elsewhere but 401 at sgp-secrets, and + # failing here is far cheaper than debugging that later. + raise IdentityServiceError( + "The identity service returned an unexpected key format; refusing " + "to store a credential the secrets service won't accept." + ) + + returned_expiry = body.get("expiresOn") + parsed_expiry: datetime | None = expires_on + if isinstance(returned_expiry, str): + try: + parsed_expiry = datetime.fromisoformat( + returned_expiry.replace("Z", "+00:00") + ) + except ValueError: + pass # keep what we requested + + logger.info( + "identity_service_key_minted", + extra={ + "sgp_user_id": sgp_user_id, + "key_id": body.get("id"), + "expires_on": str(parsed_expiry), + }, + ) + return str(secret), parsed_expiry diff --git a/agentex/src/api/app.py b/agentex/src/api/app.py index 9fc26ee1..1d197d34 100644 --- a/agentex/src/api/app.py +++ b/agentex/src/api/app.py @@ -37,6 +37,7 @@ deployment_history, deployments, events, + integrations, linear, messages, slack, @@ -206,6 +207,10 @@ async def handle_unexpected(request, exc): fastapi_app.include_router(agent_task_tracker.router) fastapi_app.include_router(agent_api_keys.router) fastapi_app.include_router(linear.router) +# Identity linking. Deliberately NOT under /slack — that prefix is +# auth-whitelisted, and this router must run authenticated so the callback can +# read the caller's SGP identity from their own session. +fastapi_app.include_router(integrations.router) fastapi_app.include_router(deployment_history.router) fastapi_app.include_router(deployments.router) # Agent run schedules are feature-flagged (off by default, enabled in development). diff --git a/agentex/src/api/routes/integrations.py b/agentex/src/api/routes/integrations.py new file mode 100644 index 00000000..5e1bbb0f --- /dev/null +++ b/agentex/src/api/routes/integrations.py @@ -0,0 +1,275 @@ +"""Identity-link routes — the browser leg of connecting a Slack user to SGP. + +These routes deliberately live under ``/integrations`` and NOT under ``/slack``. +``/slack`` is auth-whitelisted (Slack's signature is its auth), so a callback placed +there would run unauthenticated — which would defeat the entire mechanism, since the +whole point of this leg is to learn who the caller is in SGP from their own session. + +The flow: + + GET /integrations/slack/link?nonce=… confirmation screen (does not consume) + POST /integrations/slack/link confirm -> mint -> store + +By the time the POST runs, both halves of the identity are present in one request: +the Slack side from the nonce (parked when Slack's HMAC verified the event), the SGP +side from the authenticated session. That coincidence is the only moment the mapping +can be established safely. + +Rendered as plain HTML rather than JSON: the user arrives here by clicking a link in +Slack, so the response is for a human, and the confirmation step is a security +control — naming both identities is what makes a mis-clicked link visible. +""" + +from __future__ import annotations + +import html +import os +from datetime import UTC, datetime, timedelta + +from fastapi import APIRouter, Form, Request +from fastapi.responses import HTMLResponse + +from src.adapters.identity_service.adapter_identity_service import ( + IdentityServiceClient, + IdentityServiceError, + forwardable_headers, +) +from src.api.middleware_utils import get_request_headers_to_forward +from src.config.dependencies import ( + database_async_read_only_session_maker, + database_async_read_write_engine, + database_async_read_write_session_maker, +) +from src.domain.entities.identity_links import IdentityLinkMethod, IdentityProvider +from src.domain.repositories.identity_link_repository import IdentityLinkRepository +from src.domain.services.identity_link_service import IdentityLinkService +from src.domain.services.link_nonce_service import LinkNonceService +from src.utils.credential_encryption import CredentialEncryptionError +from src.utils.logging import make_logger + +logger = make_logger(__name__) + +router = APIRouter(prefix="/integrations", tags=["Integrations"]) + +# How long a minted key lasts. Bounded on purpose: this is the most sensitive thing +# agentex stores, and an expiry turns "valid until someone notices" into a known +# window. Long enough that re-linking isn't a weekly chore. +_KEY_TTL_DAYS = int(os.getenv("IDENTITY_LINK_KEY_TTL_DAYS", "30")) + +_KEY_NAME_PREFIX = "agentex-slack-link" + + +def _page(title: str, body: str, *, status: int = 200) -> HTMLResponse: + """Minimal self-contained page. No external assets — this renders inside + whatever browser Slack opened, possibly without network access to our CDN.""" + return HTMLResponse( + status_code=status, + content=( + "" + "" + f"{html.escape(title)}" + "" + f"{body}" + ), + ) + + +def _identity_link_service() -> IdentityLinkService: + engine = database_async_read_write_engine() + return IdentityLinkService( + IdentityLinkRepository( + database_async_read_write_session_maker(engine), + database_async_read_only_session_maker(engine), + ) + ) + + +def _principal(request: Request) -> tuple[str | None, str | None, str | None]: + """(sgp_user_id, sgp_account_id, email) from the authenticated session. + + Populated by AgentexAuthMiddleware, which is why this route must not be + whitelisted. With authz disabled locally there is no principal, so linking is + refused rather than guessed at — binding an identity is exactly the operation + that must not proceed on an assumption. + """ + ctx = getattr(request.state, "principal_context", None) or {} + if not isinstance(ctx, dict): + ctx = getattr(ctx, "__dict__", {}) or {} + raw_user = ctx.get("raw_user") or {} + return ( + ctx.get("user_id"), + ctx.get("account_id"), + raw_user.get("email") if isinstance(raw_user, dict) else None, + ) + + +@router.get("/slack/link", summary="Confirm linking a Slack identity to SGP") +async def slack_link_page(request: Request, nonce: str = "") -> HTMLResponse: + """Render the confirmation screen. Does NOT consume the nonce, so a refresh or + a link-prefetching browser doesn't break the flow.""" + link_request = await LinkNonceService().peek(nonce) + if link_request is None: + return _page( + "Link expired", + "

This link has expired

Links are single-use and " + "valid for a few minutes. Mention the agent in Slack again to get a " + "fresh one.

", + status=400, + ) + + sgp_user_id, _account_id, email = _principal(request) + if not sgp_user_id: + return _page( + "Sign in required", + "

Please sign in to SGP

Sign in and open this " + "link again to finish connecting your account.

", + status=401, + ) + + slack_who = link_request.display_name or link_request.external_user_id + return _page( + "Connect your account", + "

Connect your Slack account?

" + "

This lets the agent use your connected tools " + "(Notion, Linear, …) when you ask it something in Slack.

" + "
" + f"
Slack
{html.escape(slack_who)}
" + f"
SGP
{html.escape(email or sgp_user_id)}
" + "
" + "
" + f"" + "
" + "

If either name above isn't you, " + "close this page and don't continue.

", + ) + + +@router.post("/slack/link", summary="Complete a Slack identity link") +async def slack_link_confirm(request: Request, nonce: str = Form("")) -> HTMLResponse: + """Mint a key as the signed-in user and store the mapping. + + Order matters: the nonce is consumed only after a successful mint, so a + transient identity-service failure leaves the link clickable instead of + burning it and forcing the user back to Slack. + """ + link_request = await LinkNonceService().peek(nonce) + if link_request is None: + return _page( + "Link expired", + "

This link has expired

Mention the agent in " + "Slack again to get a fresh one.

", + status=400, + ) + + sgp_user_id, sgp_account_id, email = _principal(request) + if not sgp_user_id: + return _page( + "Sign in required", + "

Please sign in to SGP

", + status=401, + ) + + provider = IdentityProvider(link_request.provider) + service = _identity_link_service() + + # Report a conflict before the partial unique index does, so the user sees a + # sentence instead of an integrity error. + existing = await service.repository.get_active_by_sgp_user( + provider=provider, + external_team_id=link_request.external_team_id, + sgp_user_id=sgp_user_id, + ) + if existing and existing.external_user_id != link_request.external_user_id: + return _page( + "Already linked", + "

That SGP account is already linked

" + "

It's connected to a different Slack user in this " + "workspace. Disconnect that one first.

", + status=409, + ) + + expires_on = datetime.now(UTC) + timedelta(days=_KEY_TTL_DAYS) + try: + secret, actual_expiry = await IdentityServiceClient().mint_user_api_key( + sgp_user_id=sgp_user_id, + name=f"{_KEY_NAME_PREFIX}-{link_request.external_user_id}", + auth_headers=forwardable_headers(get_request_headers_to_forward(request)), + expires_on=expires_on, + ) + except IdentityServiceError as exc: + logger.warning("identity link mint failed: %s", exc) + return _page( + "Couldn't create a key", + f"

Couldn't finish connecting

{html.escape(str(exc))}

" + "

Your link is still valid — you can retry.

", + status=502, + ) + + try: + await service.repository.upsert_link( + provider=provider, + external_team_id=link_request.external_team_id, + external_user_id=link_request.external_user_id, + sgp_user_id=sgp_user_id, + sgp_account_id=sgp_account_id or "", + linked_via=IdentityLinkMethod.EXPLICIT, + credential=secret, + credential_expires_at=actual_expiry, + ) + except CredentialEncryptionError: + # AGENTEX_CREDENTIAL_ENCRYPTION_KEY is missing or malformed, so the freshly + # minted key cannot be stored safely. Deliberately NOT stored in plaintext. + # Reported as an operator problem rather than a 500, because the user can't + # do anything about it and would otherwise just retry forever. The nonce is + # left intact so the link still works once the key is configured. + logger.error( + "identity link failed: credential encryption is not configured; " + "set AGENTEX_CREDENTIAL_ENCRYPTION_KEY", + exc_info=True, + ) + return _page( + "Not configured", + "

Couldn't finish connecting

" + "

This deployment isn't set up to store credentials yet.

" + "

Nothing was saved. Please let the team know — the " + "server needs its credential encryption key configured.

", + status=503, + ) + # Burn the nonce only now that the link is durable. + await LinkNonceService().consume(nonce) + # Drop the negative cache entry so the very next Slack message resolves. + await service.invalidate( + provider=provider, + external_team_id=link_request.external_team_id, + external_user_id=link_request.external_user_id, + ) + logger.info( + "identity_link_completed", + extra={ + "provider": provider.value, + "external_user_id": link_request.external_user_id, + "sgp_user_id": sgp_user_id, + "expires_on": str(actual_expiry), + }, + ) + + when = actual_expiry.date().isoformat() if actual_expiry else "further notice" + return _page( + "Connected", + "

You're connected

" + f"

The agent will now use your own tools when you ask it something in " + f"Slack, as {html.escape(email or sgp_user_id)}.

" + f"

This connection is valid until {html.escape(when)}, after " + "which the agent will ask you to reconnect. You can close this page and go " + "back to Slack.

", + ) diff --git a/agentex/src/domain/services/identity_link_service.py b/agentex/src/domain/services/identity_link_service.py new file mode 100644 index 00000000..6a1fd984 --- /dev/null +++ b/agentex/src/domain/services/identity_link_service.py @@ -0,0 +1,248 @@ +"""Resolve a provider identity to the SGP identity an event-driven turn should act as. + +Two lookups with deliberately different caching, because they carry different things: + +``resolve()`` -> the link (who this Slack user is). Cached in Redis. The + entity holds no credential, so the cache holds no secret. +``acting_headers()`` -> the delegation headers, including that user's SGP API key. + NEVER cached. Read from Postgres per turn, decrypted in + memory, handed straight to the ACP call. One indexed + lookup is cheap; a key sitting in Redis is not. + +Negative results are cached too. Unlinked users are the common case during rollout, +and without a negative entry every event from one is a guaranteed miss plus a DB +round trip — exactly the traffic a busy shared channel produces. Negatives get a +shorter TTL so a freshly linked user starts working promptly. + +Deliberately NOT fail-open: a cache miss falls through to Postgres, but a *lookup +failure* propagates. "We could not determine who this is" must never collapse into +"this is nobody", because the caller treats the latter as "run as the shared bot" +and would silently downgrade a user-scoped turn. +""" + +from __future__ import annotations + +import json +import os +from datetime import UTC, datetime +from typing import Annotated, Any + +from fastapi import Depends + +from src.domain.entities.identity_links import IdentityLinkEntity, IdentityProvider +from src.domain.repositories.identity_link_repository import DIdentityLinkRepository +from src.utils.credential_encryption import CredentialEncryptionError +from src.utils.logging import make_logger + +logger = make_logger(__name__) + +# Positive entries are stable — a link changes only on an explicit link/unlink, and +# both paths invalidate. Negatives expire fast so linking feels immediate. +_CACHE_TTL_S = int(os.getenv("IDENTITY_LINK_CACHE_TTL", "300")) +_NEGATIVE_CACHE_TTL_S = int(os.getenv("IDENTITY_LINK_NEGATIVE_CACHE_TTL", "30")) + +# Distinguishes "cached: known to be unlinked" from "not in cache". +_UNLINKED = "-" + +HEADER_API_KEY = "x-api-key" +HEADER_SELECTED_ACCOUNT_ID = "x-selected-account-id" + + +def _cache_key( + provider: IdentityProvider, external_team_id: str, external_user_id: str +) -> str: + return f"identity_link:{provider.value}:{external_team_id}:{external_user_id}" + + +class ResolvedIdentity: + """A provider identity resolved to an SGP identity. + + ``principal`` is shaped for agentex-auth's ``SGPPrincipalContext`` and is passed + to ``/v1/authz/*`` verbatim. It carries no api_key because permission checks + only need (user_id, account_id) — the credential travels separately, on the + delegation headers, and only when the caller asks for it. + """ + + def __init__(self, link: IdentityLinkEntity): + self.link = link + self.sgp_user_id = link.sgp_user_id + self.sgp_account_id = link.sgp_account_id + + @property + def principal(self) -> dict[str, Any]: + return {"user_id": self.sgp_user_id, "account_id": self.sgp_account_id} + + def credential_is_usable(self, *, now: datetime | None = None) -> bool: + return self.link.credential_is_usable(now=now or datetime.now(UTC)) + + +class IdentityLinkService: + def __init__(self, repository: DIdentityLinkRepository): + self.repository = repository + + async def resolve( + self, + provider: IdentityProvider, + external_team_id: str, + external_user_id: str, + ) -> ResolvedIdentity | None: + """Resolve a provider identity, or None when it isn't linked. + + None means "definitively not linked". A lookup failure raises. + """ + if not (external_team_id and external_user_id): + return None + + cached = await self._cache_get(provider, external_team_id, external_user_id) + if cached is _UNLINKED: + return None + if cached is not None: + return ResolvedIdentity(cached) + + link = await self.repository.get_active_by_external_user( + provider=provider, + external_team_id=external_team_id, + external_user_id=external_user_id, + ) + await self._cache_put(provider, external_team_id, external_user_id, link) + return ResolvedIdentity(link) if link else None + + async def acting_headers(self, identity: ResolvedIdentity) -> dict[str, str] | None: + """The delegation headers for acting as this user, or None if we can't. + + These are what ``build_delegation_headers`` converts into + ``x-acting-user-api-key`` on the ACP call, which is what makes the agent's + user-scoped tools (Notion, Linear, Slack) resolve *this user's* connections + instead of the gateway bot's. + + Returns None — never raises — for every "can't act as them" case: no stored + credential, an expired one, or a ciphertext that won't decrypt. The caller + decides what to do about it (fall back, or prompt a re-link), and a + None-vs-exception split would make that awkward at the call site. The reason + is logged, since the three cases need different fixes. + """ + if not identity.link.has_credential: + logger.info( + "identity_link_no_credential", + extra={"sgp_user_id": identity.sgp_user_id}, + ) + return None + if not identity.credential_is_usable(): + logger.info( + "identity_link_credential_expired", + extra={ + "sgp_user_id": identity.sgp_user_id, + "expired_at": str(identity.link.credential_expires_at), + }, + ) + return None + try: + credential = await self.repository.get_credential(identity.link.id) + except CredentialEncryptionError: + # Wrong key or tampered ciphertext. Not recoverable here; the owner has + # to re-link. Logged loudly because it usually means a key rotation + # left existing rows unreadable. + logger.warning( + "identity_link_credential_unreadable", + extra={"sgp_user_id": identity.sgp_user_id}, + exc_info=True, + ) + return None + if not credential: + return None + headers = {HEADER_API_KEY: credential} + if identity.sgp_account_id: + headers[HEADER_SELECTED_ACCOUNT_ID] = identity.sgp_account_id + return headers + + async def invalidate( + self, + provider: IdentityProvider, + external_team_id: str, + external_user_id: str, + ) -> None: + """Drop the cached entry, so a link/unlink takes effect on the next event + rather than after the TTL.""" + client = self._redis() + if client is None: + return + try: + await client.delete( + _cache_key(provider, external_team_id, external_user_id) + ) + except Exception: # noqa: BLE001 - invalidation is best-effort + logger.warning("[identity_link] cache invalidation failed", exc_info=True) + + # ----------------------------------------------------------------- cache layer + + def _redis(self): + """The shared Redis client, or None when unavailable (unit tests, deps not + loaded). A missing cache degrades to DB-only, never to a wrong answer.""" + try: + from src.config.dependencies import GlobalDependencies + + pool = GlobalDependencies().redis_pool + except Exception: # noqa: BLE001 - deps not initialized -> no cache + return None + if pool is None: + return None + try: + import redis.asyncio as redis + + return redis.Redis(connection_pool=pool) + except Exception: # noqa: BLE001 + return None + + async def _cache_get( + self, + provider: IdentityProvider, + external_team_id: str, + external_user_id: str, + ) -> IdentityLinkEntity | str | None: + """Entity on a hit, ``_UNLINKED`` on a cached negative, None for a miss + (including any cache failure).""" + client = self._redis() + if client is None: + return None + try: + raw = await client.get( + _cache_key(provider, external_team_id, external_user_id) + ) + except Exception: # noqa: BLE001 - a cache error is just a miss + logger.warning("[identity_link] cache read failed", exc_info=True) + return None + if raw is None: + return None + if isinstance(raw, bytes): + raw = raw.decode() + if raw == _UNLINKED: + return _UNLINKED + try: + return IdentityLinkEntity.model_validate(json.loads(raw)) + except Exception: # noqa: BLE001 - stale/incompatible payload -> re-read DB + logger.warning("[identity_link] cache payload unusable", exc_info=True) + return None + + async def _cache_put( + self, + provider: IdentityProvider, + external_team_id: str, + external_user_id: str, + link: IdentityLinkEntity | None, + ) -> None: + client = self._redis() + if client is None: + return + key = _cache_key(provider, external_team_id, external_user_id) + try: + if link is None: + await client.set(key, _UNLINKED, ex=_NEGATIVE_CACHE_TTL_S) + else: + # Safe to cache: IdentityLinkEntity has no credential field, so this + # payload cannot contain key material. + await client.set(key, link.model_dump_json(), ex=_CACHE_TTL_S) + except Exception: # noqa: BLE001 - caching is best-effort + logger.warning("[identity_link] cache write failed", exc_info=True) + + +DIdentityLinkService = Annotated[IdentityLinkService, Depends(IdentityLinkService)] diff --git a/agentex/src/domain/services/link_nonce_service.py b/agentex/src/domain/services/link_nonce_service.py new file mode 100644 index 00000000..2ffd94ec --- /dev/null +++ b/agentex/src/domain/services/link_nonce_service.py @@ -0,0 +1,166 @@ +"""Short-lived handshake state for the identity-link flow. + +The link flow spans two HTTP requests that each prove one half of an identity: + + 1. A Slack event or slash command. Slack's HMAC proves the *provider* identity — + we know this really is ``U…`` in team ``T…``, because only Slack could have + signed it. + 2. A browser hit on an authenticated agentex route. The session proves the *SGP* + identity. + +Nothing carries between them on its own, so the first proof has to be parked +somewhere the second request can pick it up. This is that parking spot. + +Why a server-side nonce rather than putting the Slack ids in the link URL: a URL is +user-editable. Given ``?slack_user=``, an attacker could click their +own link while signed in as themselves and bind *your* Slack identity to *their* SGP +account — after which your Slack messages would run as them, using their +integrations, with the resulting task (and your prompt) landing in their account. +Handing out an opaque token instead means nothing in the URL is meaningful, so +nothing in it is forgeable. + +Signed URL parameters would also close that hole, and would need no Redis. They are +rejected here for two reasons: a signed URL is replayable for its whole validity +window, whereas a nonce is consumed on first use; and the pending turn (so the user +gets an answer to the question they originally asked) does not fit in a query string. + +The nonce holds no secrets — only public-ish identifiers and the user's own message +— so its blast radius if Redis were read is "someone learns a Slack user id". The +credential it eventually produces is never stored here. +""" + +from __future__ import annotations + +import json +import os +import secrets +from dataclasses import asdict, dataclass, field +from typing import Annotated, Any + +from fastapi import Depends + +from src.utils.logging import make_logger + +logger = make_logger(__name__) + +# Long enough for a human to switch windows and sign in, short enough that an +# abandoned link stops being interesting. +_TTL_S = int(os.getenv("IDENTITY_LINK_NONCE_TTL", "600")) + +# 32 bytes of urlsafe randomness. Guessing is not a threat model at this size, but +# the token is still consumed on first use rather than relying on entropy alone. +_TOKEN_BYTES = 32 + +_KEY_PREFIX = "link_nonce:" + + +@dataclass +class LinkRequest: + """The verified provider identity, parked for the browser leg of the flow.""" + + provider: str + external_team_id: str + external_user_id: str + # For the confirmation screen. Naming both sides is what makes a mis-clicked + # link visible to the person clicking it, so this is a security affordance + # rather than decoration. + display_name: str = "" + # The turn that triggered the prompt, so linking can end with an answer to the + # original question instead of "now ask me again". + pending_turn: dict[str, Any] | None = field(default=None) + + +def _key(token: str) -> str: + return f"{_KEY_PREFIX}{token}" + + +class LinkNonceService: + """Create / read / consume link nonces. + + Requires Redis. Unlike the identity-link cache — where a missing cache just + means "read the database" — there is no fallback here: without somewhere to + park the Slack identity, the flow cannot be completed safely, and the + alternative (trusting ids from the URL) is the vulnerability described above. + So a missing Redis raises rather than degrading. + """ + + def __init__(self, redis_client: Any | None = None): + self._client = redis_client + + def _redis(self): + if self._client is not None: + return self._client + from src.config.dependencies import GlobalDependencies + + pool = GlobalDependencies().redis_pool + if pool is None: + raise RuntimeError( + "identity linking requires Redis (nonce storage) and no pool is " + "configured" + ) + import redis.asyncio as redis + + self._client = redis.Redis(connection_pool=pool) + return self._client + + async def create(self, request: LinkRequest) -> str: + """Park a verified provider identity and return its opaque token.""" + token = secrets.token_urlsafe(_TOKEN_BYTES) + await self._redis().set(_key(token), json.dumps(asdict(request)), ex=_TTL_S) + logger.info( + "link_nonce_created", + extra={ + "provider": request.provider, + "external_team_id": request.external_team_id, + "external_user_id": request.external_user_id, + "has_pending_turn": request.pending_turn is not None, + "ttl_s": _TTL_S, + }, + ) + return token + + async def peek(self, token: str) -> LinkRequest | None: + """Read without consuming — for rendering the confirmation screen. + + Deliberately separate from ``consume``: if loading the page burned the + nonce, a refresh (or a browser prefetching the link) would break the flow + before the user could confirm. + """ + if not token: + return None + raw = await self._redis().get(_key(token)) + return self._decode(raw) + + async def consume(self, token: str) -> LinkRequest | None: + """Read and delete atomically — single use, on confirm. + + ``GETDEL`` so two concurrent confirms can't both succeed. Falls back to + GET+DELETE on Redis older than 6.2, which is very slightly racy but only + between two requests already holding the same token. + """ + if not token: + return None + client = self._redis() + try: + raw = await client.getdel(_key(token)) + except Exception: # noqa: BLE001 - GETDEL unsupported on older Redis + raw = await client.get(_key(token)) + if raw is not None: + await client.delete(_key(token)) + return self._decode(raw) + + @staticmethod + def _decode(raw: Any) -> LinkRequest | None: + if raw is None: + return None + if isinstance(raw, bytes): + raw = raw.decode() + try: + data = json.loads(raw) + return LinkRequest(**data) + except Exception: # noqa: BLE001 - a malformed nonce is an expired nonce + logger.warning("[link_nonce] undecodable payload; treating as expired") + return None + + +DLinkNonceService = Annotated[LinkNonceService, Depends(LinkNonceService)] diff --git a/agentex/src/domain/use_cases/slack_gateway_use_case.py b/agentex/src/domain/use_cases/slack_gateway_use_case.py index 0c5c6e02..63ed1eeb 100644 --- a/agentex/src/domain/use_cases/slack_gateway_use_case.py +++ b/agentex/src/domain/use_cases/slack_gateway_use_case.py @@ -11,16 +11,27 @@ per-agent verifying proxy. This is its own module so Slack-specific logic stays out of the generic path. -Identity: every Slack turn acts as the gateway's own SGP identity — a dedicated bot -service account (``SLACK_GATEWAY_ACTING_BOT_API_KEY`` + ``SLACK_GATEWAY_ACCOUNT_ID``, -env / k8s-secret only). The key is forwarded as ``x-api-key``, which the platform (a) -verifies -> principal for authz and (b) converts to ``x-acting-user-api-key`` so the -agent's tools act as the bot via resolve_user_secrets. The bot is a first-class entity, -not a proxy for the invoking user: all Slack traffic shares its account and its tasks -are owned by it — fine for a controlled internal deploy, NOT per-user multi-tenant. -Deliberately NOT per-user: we don't conflate the invoking user's SGP identity with the -bot's. The bot's Slack credentials (signing secret, bot token) live in the same -env / k8s-secret set. Dispatch, delegation, and idempotency are real. +Identity: a turn runs as the **invoking human** when that Slack user has an active +identity link with a usable stored credential (see ``_turn_identity``). Their SGP API +key rides on the delegation headers, becomes ``x-acting-user-api-key`` on the ACP +call, and is what makes the agent's user-scoped tools resolve *their* connected +integrations — Notion, Linear, the hosted Slack MCP — rather than a shared account's. +That per-user resolution is the whole point: the secrets service derives the owner +from the caller and offers no way to ask for someone else's, so acting as a person +requires holding a credential belonging to that person. + +Everyone else falls back to the gateway's own bot service account +(``SLACK_GATEWAY_ACTING_BOT_API_KEY`` + ``SLACK_GATEWAY_ACCOUNT_ID``, env / +k8s-secret only), which still produces a working turn — just without personal +integrations. So user scoping is opt-in per person, and NOT an isolation guarantee +while the fallback is enabled; ``SLACK_GATEWAY_REQUIRE_LINKED_USER`` closes it. + +Task keying follows the identity: a linked user gets one task per (thread, user), so +each participant in a shared thread owns their own conversation and a task never has +two owners. Unlinked users keep the legacy thread-wide key. + +The bot's Slack credentials (signing secret, bot token) are separate from all of this +and remain shared — the gateway posts as the app, not as the user. """ from __future__ import annotations @@ -91,6 +102,24 @@ _MESSAGE_PAGE = 200 # per-poll page size when collecting the reply +# What an unlinked user gets. Default OFF: an unlinked user falls back to the shared +# bot identity and still gets a working turn, just without their personal +# integrations. Turn it ON once enough of the workspace has linked. +# +# Be clear-eyed about what OFF means: with the fallback in place, running as the +# invoking human is opt-in, so it is NOT an isolation guarantee — anyone who hasn't +# linked simply inherits the bot's (much narrower) access instead. +_REQUIRE_LINKED_USER = os.getenv("SLACK_GATEWAY_REQUIRE_LINKED_USER", "").lower() in ( + "1", + "true", + "yes", +) + +_UNLINKED_MESSAGE = ( + "I don't know who you are in SGP yet, so I can't run this as you. " + "Connect your account and try again." +) + # Slack's HTTP Events API is at-least-once — it retries a delivery (up to ~3x, with an # X-Slack-Retry-Num header) if we don't 200 within ~3s. Dedup on the envelope's # ``event_id`` via Redis with a short TTL so a retry can't start a duplicate turn. @@ -561,10 +590,24 @@ async def _submit_agents_modal( async def _run_turn(self, inbound: InboundSlack) -> None: try: - # One shared v1 identity per turn, resolved once and threaded into both target - # resolution (the SGP agent_config lookup) and dispatch (principal + the - # delegated x-api-key headers). - principal, auth_headers = await self._acting_identity() + # Whose identity runs this turn. + # + # If the invoking Slack user has an active link with a usable stored + # credential, the turn runs as THEM: their principal for authz/ownership, + # and their SGP API key on the delegation headers. That key is what + # becomes x-acting-user-api-key on the ACP call, which is what makes the + # agent's user-scoped tools (Notion, Linear, the hosted Slack MCP) resolve + # that person's own connections instead of a shared account's. + # + # Otherwise we fall back to the shared gateway bot — unchanged behavior, + # so an unlinked user still gets a working turn, just without their + # personal integrations. Prompting them to link is a separate concern. + principal, auth_headers, sgp_user_id = await self._turn_identity(inbound) + if principal is None and auth_headers is None: + # Only reachable when linking is mandatory and this user hasn't. + await self._deliver(inbound, _UNLINKED_MESSAGE) + return + target, prompt = await self._resolve_target(inbound, auth_headers) if not await self._authorize(target): @@ -587,7 +630,13 @@ async def _run_turn(self, inbound: InboundSlack) -> None: if target.agent_name == _DEFAULT_AGENT_NAME: await self._set_status(inbound, "is thinking…") await self._dispatch( - target, inbound, prompt, principal, auth_headers, collect=False + target, + inbound, + prompt, + principal, + auth_headers, + collect=False, + sgp_user_id=sgp_user_id, ) return @@ -595,7 +644,12 @@ async def _run_turn(self, inbound: InboundSlack) -> None: # automatically when we post the reply. No-op outside an assistant thread. await self._set_status(inbound, "is thinking…") reply = await self._dispatch( - target, inbound, prompt, principal, auth_headers + target, + inbound, + prompt, + principal, + auth_headers, + sgp_user_id=sgp_user_id, ) note = f"_via {target.label()}_" # attribution await self._deliver( @@ -610,6 +664,107 @@ async def _run_turn(self, inbound: InboundSlack) -> None: inbound, "Something went wrong handling that. Please retry." ) + async def _turn_identity( + self, inbound: InboundSlack + ) -> tuple[Any, dict[str, str] | None, str | None]: + """Decide whose identity this turn runs as. + + Returns ``(principal, auth_headers, sgp_user_id)``: + + - linked user with a usable credential -> their principal, their delegation + headers, their SGP user id. The turn acts as them end to end. + - otherwise -> the shared bot's principal and headers, and ``None`` for the + user id (which keeps the legacy thread-wide task key, so nothing that's + already running gets orphaned). + - ``(None, None, None)`` only when ``_REQUIRE_LINKED_USER`` is set and this + user has no usable link, telling the caller to refuse the turn. + + A resolution *failure* propagates rather than falling back: silently running + as the bot because the database hiccuped would be indistinguishable from + "this person isn't linked", and the two need different handling. + """ + identity = await self._resolve_invoking_identity(inbound) + if identity is not None: + headers = await self._identity_link_service().acting_headers(identity) + if headers is not None: + logger.info( + "[slack] turn acting as sgp user %s (linked from %s)", + identity.sgp_user_id, + inbound.user, + ) + return identity.principal, headers, identity.sgp_user_id + # Linked but unusable — no stored key, expired, or undecryptable. The + # service has already logged which. Treated the same as unlinked here; + # re-link prompting is handled separately. + + if _REQUIRE_LINKED_USER: + logger.info( + "[slack] refusing turn: user %s in team %s has no usable link", + inbound.user, + inbound.team_id, + ) + return None, None, None + + bot_principal, bot_headers = await self._acting_identity() + logger.info( + "[slack] turn falling back to the shared bot identity for user %s", + inbound.user, + ) + return bot_principal, bot_headers, None + + def _identity_link_service(self): + """Build the identity-link service. + + Constructed inline for the same reason as ``_get_agent_by_name``: this use + case is instantiated per-request with no constructor deps, and the identity + map is infrastructure the gateway owns rather than something a caller passes + in. + """ + # Local imports keep these off the module-load path. + from src.domain.repositories.identity_link_repository import ( + IdentityLinkRepository, + ) + from src.domain.services.identity_link_service import IdentityLinkService + + engine = database_async_read_write_engine() + return IdentityLinkService( + IdentityLinkRepository( + database_async_read_write_session_maker(engine), + database_async_read_only_session_maker(engine), + ) + ) + + async def _resolve_invoking_identity(self, inbound: InboundSlack): + """Resolve the Slack user who triggered this turn to an SGP identity, or + None when they have no active link.""" + from src.domain.entities.identity_links import IdentityProvider + + return await self._identity_link_service().resolve( + provider=IdentityProvider.SLACK, + external_team_id=inbound.team_id, + external_user_id=inbound.user, + ) + + def _task_name(self, inbound: InboundSlack, sgp_user_id: str | None) -> str: + """The conversation key. + + For a linked user the task is per (thread, user): each invoker gets their own + task, in their own account, holding only their own turns. That's what makes + per-user ownership coherent in a shared thread — one task can't be owned by + two people — and it removes the reply-attribution race, since a task now only + ever contains one user's messages. + + The agent loses the other participants' turns from its own history by design; + it recovers that context by reading the thread with its Slack tools (the + ``[Slack context]`` prefix carries the channel and thread for exactly that). + + Unlinked users keep the legacy thread-wide key, so turning this on doesn't + orphan conversations already in flight. + """ + if sgp_user_id: + return f"slack:{inbound.thread_ts}:{sgp_user_id}" + return f"slack:{inbound.thread_ts}" + async def _resolve_config_id( self, name: str, auth_headers: dict[str, str] ) -> str | None: @@ -692,6 +847,7 @@ async def _dispatch( auth_headers: dict[str, str], *, collect: bool = True, + sgp_user_id: str | None = None, ) -> str | None: """Create-or-resume a task on the resolved agent, then inject the turn, acting as the shared v1 identity (x-api-key -> principal for authz, delegated downstream as @@ -714,7 +870,7 @@ async def _dispatch( GlobalDependencies(), principal, request_headers=auth_headers ) agent = await acp.agent_repository.get(name=target.agent_name) - task_name = f"slack:{inbound.thread_ts}" + task_name = self._task_name(inbound, sgp_user_id) # golden-agent isn't relayed (it self-posts), so its context gets the directive to # post its own reply. Keyed on the same signal _run_turn uses to skip the relay. content = TextContentEntity( @@ -754,7 +910,13 @@ async def _dispatch( "sender_id": target.label(), "thread_ts": inbound.thread_ts, "channel_id": inbound.channel, + # Who actually asked. The Slack id is what the agent's Slack tools + # act on; the SGP id (present only for a linked user) is what makes + # the task attributable to a human rather than to the gateway bot. + "slack_user_id": inbound.user, } + if sgp_user_id: + task_metadata["sgp_user_id"] = sgp_user_id if target.config_id: task_metadata["config_id"] = target.config_id try: diff --git a/agentex/tests/unit/adapters/test_identity_service_adapter.py b/agentex/tests/unit/adapters/test_identity_service_adapter.py new file mode 100644 index 00000000..a960e45c --- /dev/null +++ b/agentex/tests/unit/adapters/test_identity_service_adapter.py @@ -0,0 +1,199 @@ +"""Unit tests for the identity-service key-minting client. + +The single most valuable test here is the wrong-issuer guard. There are two API-key +issuers on the platform and only identity-service's ``ssk_`` keys are accepted by +sgp-secrets — a key from egp-api-backend authenticates fine against egp-api-backend +and then fails at the vault with an opaque 401. Storing one would produce a link +that looks healthy and silently can't read any secrets, so the client refuses +anything that isn't ``ssk_``-shaped. +""" + +from datetime import UTC, datetime, timedelta + +import pytest +from src.adapters.identity_service import adapter_identity_service as mod +from src.adapters.identity_service.adapter_identity_service import ( + IdentityServiceClient, + IdentityServiceError, + forwardable_headers, +) + +_USER = "11111111-2222-4333-8444-555555555555" +_GOOD = "ssk_is_" + "a" * 32 + + +def _fake_http(monkeypatch, *, status=200, body=None, raises=None, captured=None): + class _Resp: + status_code = status + text = "" if body is None else "body" + + def json(self): + if body is None: + raise ValueError("no json") + return body + + class _Client: + def __init__(self, **kw): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def post(self, url, json=None, headers=None): + if captured is not None: + captured.update(url=url, json=json, headers=headers) + if raises is not None: + raise raises + return _Resp() + + monkeypatch.setattr(mod.httpx, "AsyncClient", _Client) + + +@pytest.mark.unit +class TestBaseUrl: + def test_unset_raises_rather_than_defaulting(self, monkeypatch): + # No default on purpose: the address is deployment-specific, and guessing + # wrong would POST the user's forwarded session credentials at whatever + # host happens to answer. + monkeypatch.delenv(mod.IDENTITY_SERVICE_URL_ENV, raising=False) + with pytest.raises(IdentityServiceError): + mod.base_url() + + def test_blank_is_treated_as_unset(self, monkeypatch): + monkeypatch.setenv(mod.IDENTITY_SERVICE_URL_ENV, " ") + with pytest.raises(IdentityServiceError): + mod.base_url() + + def test_trailing_slash_is_stripped(self, monkeypatch): + monkeypatch.setenv(mod.IDENTITY_SERVICE_URL_ENV, "http://ident/") + assert mod.base_url() == "http://ident" + + +@pytest.mark.unit +class TestForwardableHeaders: + def test_keeps_only_credential_headers(self): + got = forwardable_headers( + { + "Cookie": "_identityJwt=abc", + "Authorization": "Bearer xyz", + "X-Api-Key": "ssk_is_x", + "X-Selected-Account-Id": "acct-1", + "User-Agent": "curl", + "Content-Length": "12", + } + ) + assert set(got) == { + "cookie", + "authorization", + "x-api-key", + "x-selected-account-id", + } + + def test_drops_empty_values_and_is_case_insensitive(self): + assert forwardable_headers({"COOKIE": "", "authorization": "Bearer z"}) == { + "authorization": "Bearer z" + } + + def test_nothing_to_forward_is_empty(self): + assert forwardable_headers({"user-agent": "curl"}) == {} + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMint: + async def test_sends_camelcase_fields_and_returns_the_secret(self, monkeypatch): + captured: dict = {} + expires = datetime(2026, 9, 30, tzinfo=UTC) + _fake_http( + monkeypatch, + body={"id": "k1", "secret": _GOOD, "expiresOn": "2026-09-30T00:00:00Z"}, + captured=captured, + ) + + secret, expiry = await IdentityServiceClient("http://ident").mint_user_api_key( + sgp_user_id=_USER, + name="agentex-slack-link-U1", + auth_headers={"cookie": "_identityJwt=abc"}, + expires_on=expires, + ) + + assert secret == _GOOD + assert expiry == expires + assert captured["url"] == "http://ident/api-keys" + # camelCase — egp-api-backend's equivalent uses snake_case, and mixing them + # up produces a 422 that reads like a server bug. + assert captured["json"]["identityId"] == _USER + assert captured["json"]["identityType"] == "user" + assert "expiresOn" in captured["json"] + # The caller's own session is forwarded, so the user mints their own key. + assert captured["headers"]["cookie"] == "_identityJwt=abc" + + async def test_omits_expiry_when_not_requested(self, monkeypatch): + captured: dict = {} + _fake_http(monkeypatch, body={"secret": _GOOD}, captured=captured) + await IdentityServiceClient("http://ident").mint_user_api_key( + sgp_user_id=_USER, name="n", auth_headers={} + ) + assert "expiresOn" not in captured["json"] + + async def test_falls_back_to_requested_expiry_when_response_omits_it( + self, monkeypatch + ): + want = datetime.now(UTC) + timedelta(days=30) + _fake_http(monkeypatch, body={"secret": _GOOD}) + _secret, expiry = await IdentityServiceClient("http://i").mint_user_api_key( + sgp_user_id=_USER, name="n", auth_headers={}, expires_on=want + ) + assert expiry == want + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMintFailures: + async def test_rejects_a_non_ssk_key(self, monkeypatch): + # THE important case: egp-api-backend mints sk_ keys that authenticate there + # but 401 at sgp-secrets. Failing here beats debugging that later. + _fake_http(monkeypatch, body={"secret": "sk_" + "b" * 100}) + with pytest.raises(IdentityServiceError, match="unexpected key format"): + await IdentityServiceClient("http://i").mint_user_api_key( + sgp_user_id=_USER, name="n", auth_headers={} + ) + + async def test_missing_secret_in_response(self, monkeypatch): + _fake_http(monkeypatch, body={"id": "k1"}) + with pytest.raises(IdentityServiceError, match="no secret"): + await IdentityServiceClient("http://i").mint_user_api_key( + sgp_user_id=_USER, name="n", auth_headers={} + ) + + @pytest.mark.parametrize("status", [401, 403]) + async def test_auth_failure_asks_the_user_to_sign_in(self, monkeypatch, status): + _fake_http(monkeypatch, status=status, body={}) + with pytest.raises(IdentityServiceError, match="session wasn't accepted"): + await IdentityServiceClient("http://i").mint_user_api_key( + sgp_user_id=_USER, name="n", auth_headers={} + ) + + @pytest.mark.parametrize("status", [422, 500]) + async def test_other_errors_surface_the_status(self, monkeypatch, status): + _fake_http(monkeypatch, status=status, body={}) + with pytest.raises(IdentityServiceError, match=str(status)): + await IdentityServiceClient("http://i").mint_user_api_key( + sgp_user_id=_USER, name="n", auth_headers={} + ) + + async def test_network_failure_is_reported_not_swallowed(self, monkeypatch): + _fake_http(monkeypatch, raises=OSError("dns")) + with pytest.raises(IdentityServiceError, match="Couldn't reach"): + await IdentityServiceClient("http://i").mint_user_api_key( + sgp_user_id=_USER, name="n", auth_headers={} + ) + + async def test_refuses_without_a_user_id(self): + with pytest.raises(IdentityServiceError, match="without a user id"): + await IdentityServiceClient("http://i").mint_user_api_key( + sgp_user_id="", name="n", auth_headers={} + ) diff --git a/agentex/tests/unit/api/test_integrations_routes.py b/agentex/tests/unit/api/test_integrations_routes.py new file mode 100644 index 00000000..5307abdf --- /dev/null +++ b/agentex/tests/unit/api/test_integrations_routes.py @@ -0,0 +1,211 @@ +"""Unit tests for the identity-link routes. + +Covers everything that doesn't require a real browser session: nonce handling, the +unauthenticated path, conflict detection, mint-failure behavior, and the ordering +guarantee that a failed mint doesn't burn the user's link. + +What these can NOT cover, and why: ``POST /api-keys`` on identity-service is guarded +by ``CustomerIdentityJwtGuard``, which reads ``_identityJwt`` / ``_jwt`` cookies and +rejects ``x-api-key``. So the live mint needs a genuine browser session and is +exercised via scripts/dev_seed_link_nonce.py against a deployed host, not here. +""" + +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest +from src.api.routes import integrations as mod +from src.domain.entities.identity_links import IdentityLinkMethod, IdentityProvider +from src.domain.services.link_nonce_service import LinkRequest + +_SGP_USER = "11111111-2222-4333-8444-555555555555" +_SGP_EMAIL = "test.user@example.com" +_SLACK_HANDLE = "@test.user" +_GOOD_KEY = "ssk_is_" + "a" * 32 + + +def _request(principal: dict | None): + """Minimal Request stand-in: what the auth middleware would have populated.""" + return SimpleNamespace( + state=SimpleNamespace(principal_context=principal), + headers={"cookie": "_identityJwt=abc"}, + ) + + +def _link_request(**kw) -> LinkRequest: + return LinkRequest( + **{ + "provider": "slack", + "external_team_id": "T1", + "external_user_id": "U1", + "display_name": _SLACK_HANDLE, + "pending_turn": {"text": "hi"}, + **kw, + } + ) + + +_PRINCIPAL = { + "user_id": _SGP_USER, + "account_id": "acct-1", + "raw_user": {"email": _SGP_EMAIL}, +} + + +@pytest.fixture +def wiring(monkeypatch): + """Stub the three collaborators: nonce store, repository, identity-service.""" + nonce = MagicMock() + nonce.peek = AsyncMock(return_value=_link_request()) + nonce.consume = AsyncMock(return_value=_link_request()) + monkeypatch.setattr(mod, "LinkNonceService", lambda *a, **k: nonce) + + repo = MagicMock() + repo.get_active_by_sgp_user = AsyncMock(return_value=None) + repo.upsert_link = AsyncMock(return_value=MagicMock(id="l1")) + service = SimpleNamespace(repository=repo, invalidate=AsyncMock()) + monkeypatch.setattr(mod, "_identity_link_service", lambda: service) + + client = MagicMock() + client.mint_user_api_key = AsyncMock( + return_value=(_GOOD_KEY, datetime.now(UTC) + timedelta(days=30)) + ) + monkeypatch.setattr(mod, "IdentityServiceClient", lambda *a, **k: client) + + return SimpleNamespace(nonce=nonce, repo=repo, service=service, client=client) + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestConfirmationPage: + async def test_names_both_identities(self, wiring): + resp = await mod.slack_link_page(_request(_PRINCIPAL), nonce="tok") + body = resp.body.decode() + assert resp.status_code == 200 + # Naming both sides is the security control: it's what makes a mis-clicked + # link visible to whoever clicked it. + assert _SLACK_HANDLE in body + assert _SGP_EMAIL in body + assert "isn't you" in body or "isn't you" in body + + async def test_does_not_consume_the_nonce(self, wiring): + await mod.slack_link_page(_request(_PRINCIPAL), nonce="tok") + wiring.nonce.consume.assert_not_awaited() + + async def test_expired_nonce_says_so(self, wiring): + wiring.nonce.peek = AsyncMock(return_value=None) + resp = await mod.slack_link_page(_request(_PRINCIPAL), nonce="stale") + assert resp.status_code == 400 + assert "expired" in resp.body.decode().lower() + + async def test_unauthenticated_asks_for_sign_in(self, wiring): + resp = await mod.slack_link_page(_request(None), nonce="tok") + assert resp.status_code == 401 + assert "sign in" in resp.body.decode().lower() + + async def test_display_name_is_html_escaped(self, wiring): + wiring.nonce.peek = AsyncMock( + return_value=_link_request(display_name="") + ) + body = ( + await mod.slack_link_page(_request(_PRINCIPAL), nonce="t") + ).body.decode() + assert "