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/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..8e6ab792 --- /dev/null +++ b/agentex/src/api/routes/integrations.py @@ -0,0 +1,346 @@ +"""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 -> store the caller's session + +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. + +The credential we keep is the caller's **own session cookie**, not a freshly minted +API key. Minting was the original design and it does not work: identity-service +permits one API key per user and every active user already has one, so ``POST +/api-keys`` answers 409 ("API key already exists for this user") and the secret of +the existing key cannot be read back. Rotating theirs would silently break whatever +else uses it. + +The session cookie avoids all of that, and is a better credential besides. It is +already in this request, so there is no outbound call to fail; it carries its own +expiry in the JWT, where a minted key defaults to none; taking it changes nothing +about the user's existing credentials; and the secrets service accepts it directly +(verified: a session cookie plus ``x-selected-account-id`` authenticates, while the +cookie alone is refused for want of account context). + +The tradeoff is that the link now lives and dies with the session: sign-out or +revocation ends it, which surfaces downstream as a rejected credential and should +prompt a re-link rather than an error. + +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.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 ( + SESSION_COOKIE_NAME, + IdentityLinkService, +) +from src.domain.services.link_nonce_service import LinkNonceService +from src.utils import session_jwt +from src.utils.credential_encryption import CredentialEncryptionError +from src.utils.logging import make_logger + +logger = make_logger(__name__) + +router = APIRouter(prefix="/integrations", tags=["Integrations"]) + +# Used only when the session token doesn't declare its own expiry. Never "no +# expiry": storing a credential with an unbounded lifetime is how you end up +# holding one indefinitely, so an unknown expiry becomes a short known one. +_FALLBACK_TTL_DAYS = int(os.getenv("IDENTITY_LINK_FALLBACK_TTL_DAYS", "30")) + + +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, + ) + + +def _session_credential(request: Request) -> str | None: + """The caller's session cookie value, or None if it isn't on the request. + + Parsed by splitting on ``;`` rather than with ``http.cookies``: a real browser + sends a long Cookie header full of analytics morsels that the stdlib parser + chokes on, silently dropping every morsel after the first bad one — which can + include the session cookie itself. The same reasoning (and the same approach) + applies in ``delegation_headers``. + """ + raw = request.headers.get("cookie") or "" + for part in raw.split(";"): + name, sep, value = part.strip().partition("=") + if sep and name.strip() == SESSION_COOKIE_NAME: + return value.strip() or None + return 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, + ) + + if not sgp_account_id: + # The secrets service refuses a session credential with no account context + # ("Account ID is required"), so a link without one would store a credential + # that can never be used. Better to refuse now than to look connected and + # quietly resolve nothing. + logger.warning( + "identity link refused: principal carried no account id", + extra={"sgp_user_id": sgp_user_id}, + ) + return _page( + "No account selected", + "

Couldn't finish connecting

" + "

Your session isn't scoped to an account.

" + "

Open SGP, pick the account whose tools you want the " + "agent to use, then click the link again.

", + status=400, + ) + + 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, + ) + + secret = _session_credential(request) + if not secret: + # The middleware authenticated this caller somehow, but not by a session + # cookie — an api-key or bearer caller, or a cookie under a different name. + # There is nothing here we can act through later, so refuse rather than + # store an empty credential. + logger.warning( + "identity link refused: no session cookie on an authenticated request", + extra={"sgp_user_id": sgp_user_id, "cookie": SESSION_COOKIE_NAME}, + ) + return _page( + "Couldn't read your session", + "

Couldn't finish connecting

" + "

This needs to be opened in a browser signed in to SGP.

" + "

If you're already signed in, open the link again in " + "that same browser rather than a new window or a different app.

", + status=400, + ) + + # The token's own expiry beats any TTL we could invent — the credential's real + # lifetime belongs to the session. Unknown expiry falls back to a bounded + # window rather than to "never". + actual_expiry = session_jwt.expires_at(secret) + if actual_expiry is None: + actual_expiry = datetime.now(UTC) + timedelta(days=_FALLBACK_TTL_DAYS) + logger.info( + "identity link: session token declared no expiry; using fallback", + extra={"sgp_user_id": sgp_user_id, "fallback_days": _FALLBACK_TTL_DAYS}, + ) + elif actual_expiry <= datetime.now(UTC): + # Already expired: the middleware accepted it, but storing it would create a + # link that cannot work. Say so instead of failing later and silently. + return _page( + "Session expired", + "

Your session has expired

" + "

Sign in to SGP again, then click the link once more.

", + status=401, + ) + + 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, + linked_via=IdentityLinkMethod.EXPLICIT, + credential=secret, + credential_expires_at=actual_expiry, + ) + except CredentialEncryptionError: + # AGENTEX_CREDENTIAL_ENCRYPTION_KEY is missing or malformed, so the session + # credential 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..811d476c --- /dev/null +++ b/agentex/src/domain/services/identity_link_service.py @@ -0,0 +1,270 @@ +"""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_COOKIE = "cookie" +HEADER_SELECTED_ACCOUNT_ID = "x-selected-account-id" + +# The stored credential is the linking user's own session cookie, so it goes back +# out as a cookie. ``build_delegation_headers`` filters a Cookie header down to its +# allowlisted names and re-emits it as ``x-acting-user-cookie``, so this name has to +# match that allowlist (``AGENTEX_DELEGATION_SESSION_COOKIE_NAMES``, default +# ``_identityJwt``) or the credential is silently dropped on the way to the agent. +SESSION_COOKIE_NAME = os.getenv("IDENTITY_LINK_SESSION_COOKIE_NAME", "_identityJwt") + + +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-cookie`` 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. + + The account id is not decoration: the secrets service rejects a session + credential presented without one ("Account ID is required"), and an account + the user isn't a member of is a 403. It has to be *their* account, captured + from their own principal when they linked — which is why it's stored + alongside the credential rather than taken from the gateway's config. + + 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 + if not identity.sgp_account_id: + # Without an account the credential is unusable downstream, so this is a + # "can't act as them" case rather than a partial identity to pass along. + # Reachable only for rows linked before the account id was captured. + logger.info( + "identity_link_no_account_id", + extra={"sgp_user_id": identity.sgp_user_id}, + ) + return None + return { + HEADER_COOKIE: f"{SESSION_COOKIE_NAME}={credential}", + HEADER_SELECTED_ACCOUNT_ID: identity.sgp_account_id, + } + + 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..b6041eed --- /dev/null +++ b/agentex/src/domain/services/link_nonce_service.py @@ -0,0 +1,304 @@ +"""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. + +One live nonce per identity, enforced by a pointer key. A nonce is a bearer token: +whoever holds it gets linked to that provider identity by signing in as themselves. +So a user who mentions the agent repeatedly must not accumulate a handful of +separately-redeemable links — each is another chance for one to be clicked by the +wrong person, and consuming one does not invalidate its siblings. Repeat mentions +therefore reuse the live token (``create_or_reuse``) and re-send that same link, +capped by ``claim_send`` so the DMs stop while the link stays valid. +""" + +from __future__ import annotations + +import json +import os +import secrets +from dataclasses import asdict, dataclass, field, replace +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 + +# How many times we will DM a user about the *same* pending link. Repeated mentions +# reuse the live nonce rather than minting another, so this caps DM noise without +# multiplying live tokens. Past the cap the caller should fall back to an ephemeral +# in-channel notice rather than going silent. +_MAX_SENDS = int(os.getenv("IDENTITY_LINK_MAX_DMS", "2")) + +_KEY_PREFIX = "link_nonce:" +# identity -> its one live token, so a second mention finds the first nonce instead +# of minting a parallel one. See create(). +_USER_PREFIX = "link_nonce_user:" +# identity -> how many DMs we have sent about the live token. +_SEND_PREFIX = "link_nonce_sends:" + + +@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}" + + +def _identity(provider: str, external_team_id: str, external_user_id: str) -> str: + return f"{provider}:{external_team_id}:{external_user_id}" + + +def _user_key(provider: str, external_team_id: str, external_user_id: str) -> str: + return f"{_USER_PREFIX}{_identity(provider, external_team_id, external_user_id)}" + + +def _send_key(provider: str, external_team_id: str, external_user_id: str) -> str: + return f"{_SEND_PREFIX}{_identity(provider, external_team_id, external_user_id)}" + + +def _as_str(raw: Any) -> str | None: + return ( + None if raw is None else (raw.decode() if isinstance(raw, bytes) else str(raw)) + ) + + +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. + + Invalidates any nonce this identity already holds, so one user never has two + redeemable links at once. That matters because a nonce is a bearer token: + every extra live one is another chance for a link to be redeemed by the + wrong person, and consuming one would not invalidate its siblings. + """ + client = self._redis() + user_key = _user_key( + request.provider, request.external_team_id, request.external_user_id + ) + superseded = _as_str(await client.get(user_key)) + if superseded is not None: + await client.delete(_key(superseded)) + + token = secrets.token_urlsafe(_TOKEN_BYTES) + await client.set(_key(token), json.dumps(asdict(request)), ex=_TTL_S) + await client.set(user_key, token, ex=_TTL_S) + # A genuinely new link gets a fresh send budget; the cap is per link, not + # per user for all time. + await client.delete( + _send_key( + request.provider, request.external_team_id, request.external_user_id + ) + ) + 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, + "superseded_previous": superseded is not None, + "ttl_s": _TTL_S, + }, + ) + return token + + async def create_or_reuse(self, request: LinkRequest) -> tuple[str, bool]: + """Return this identity's live token if it has one, else mint a fresh one. + + Returns ``(token, reused)``. Reuse deliberately does **not** extend the TTL: + otherwise someone mentioning the agent every few minutes could keep a single + token alive indefinitely, and the bounded lifetime is the point. The pending + turn is refreshed within whatever window remains, so linking answers what + the user most recently asked rather than their first attempt. + """ + client = self._redis() + token = _as_str( + await client.get( + _user_key( + request.provider, + request.external_team_id, + request.external_user_id, + ) + ) + ) + if token is not None: + live = await self.peek(token) + # The pointer is keyed by identity so a match is expected; verified + # anyway rather than trusting a stale pointer to name the right person. + if live is not None and ( + live.provider, + live.external_team_id, + live.external_user_id, + ) == ( + request.provider, + request.external_team_id, + request.external_user_id, + ): + await self._refresh_pending_turn(token, live, request.pending_turn) + return token, True + return await self.create(request), False + + async def claim_send(self, request: LinkRequest) -> bool: + """Record intent to DM this user their link. False once the cap is reached. + + Counted per live link (the counter is cleared whenever a fresh nonce is + minted), so a user stops being DMed about a link they are ignoring, while a + genuinely new link is never silently withheld. On False the caller should + still acknowledge in-channel — ephemerally — rather than appearing to do + nothing. + """ + key = _send_key( + request.provider, request.external_team_id, request.external_user_id + ) + client = self._redis() + count = int(await client.incr(key)) + # Bound the counter to the life of the link it describes. The TTL re-check + # covers a crash between INCR and EXPIRE, which would otherwise leave a key + # with no expiry and a user permanently un-DMable. + if count == 1 or int(await client.ttl(key)) < 0: + await client.expire(key, _TTL_S) + return count <= _MAX_SENDS + + async def _refresh_pending_turn( + self, token: str, live: LinkRequest, pending_turn: dict[str, Any] | None + ) -> None: + """Point a reused nonce at the user's latest message, keeping its TTL. + + Best-effort: if the payload cannot be rewritten, the earlier question + stands. That is worse UX than the newest one, but it is not wrong, and it + is much better than dropping the nonce and forcing a re-link. + """ + if pending_turn is None or pending_turn == live.pending_turn: + return + updated = replace(live, pending_turn=pending_turn) + try: + await self._redis().set( + _key(token), json.dumps(asdict(updated)), keepttl=True + ) + except Exception: # noqa: BLE001 - no KEEPTTL: keep the older pending turn + logger.warning( + "[link_nonce] could not refresh pending turn; keeping the earlier one" + ) + + 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..7f635e24 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,38 @@ 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 stored +session credential rides on the delegation headers, becomes ``x-acting-user-cookie`` +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. + +It is their session cookie rather than a per-link API key because identity-service +allows one API key per user and every active user already has one, so minting +answers 409 and the existing key's secret cannot be read back. The session cookie is +also the better credential: it carries its own expiry, and taking it leaves the +user's existing credentials untouched. The cost is that the link ends when the +session does. + +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 (workspace, +channel, thread, user), so each participant in a shared thread owns their own +conversation and a task never has two owners. Workspace and channel are in the key +because ``task/create`` is get-or-create on the name — two turns yielding the same +name become one task, merging their prompts, config and account context — and +``thread_ts`` is unique only within a workspace. Unlinked users keep the legacy +thread-wide key, which carries that same weakness and predates this work. + +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 +113,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 +601,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 +641,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 +655,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,16 +675,156 @@ 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 (workspace, channel, 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). + + **Team and channel are in the key deliberately.** ``task/create`` is + get-or-create on the name, so two turns that produce the same name become one + task — mixing their prompts, metadata, agent config and account context into + a single conversation. ``thread_ts`` alone does not rule that out: it is + unique within a workspace but nothing makes it unique *across* workspaces, + and this gateway is multi-workspace (``team_id`` is part of the identity + everywhere else). The odds of two workspaces minting the same microsecond + timestamp are tiny, but the failure is silent and cross-tenant, and the two + extra segments cost nothing. + + They also bound the damage if ``thread_ts`` is ever empty — ``normalize()`` + falls back to ``""`` when an event carries neither ``thread_ts`` nor ``ts``. + With team and channel present that degrades to one task per (workspace, + channel, user), which is a reasonable conversation anyway; on the old key it + would have collapsed every such turn into a single global task. + + Unlinked users keep the legacy thread-wide key, so turning this on doesn't + orphan conversations already in flight. That key has the same cross-workspace + weakness and predates this change; widening it would re-key live threads, so + it's left alone here. + """ + if sgp_user_id: + return ( + f"slack:{inbound.team_id}:{inbound.channel}:" + f"{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: """Resolve an agent_config NAME -> id via SGP's directory - (GET {SGP}/v5/agent_configs?name=), authenticated with the acting identity's - headers (x-api-key + x-selected-account-id). Cached by (account, name) for the - process lifetime. Fail-safe -> None (no SGP base / no acting key / any error) so - the caller falls back to the fixed default id.""" - api_key = auth_headers.get("x-api-key") - if not (_SGP_BASE_URL and name and api_key): + (GET {SGP}/v5/agent_configs?name=), authenticated with whatever credential the + acting identity carries. Cached by (account, name) for the process lifetime. + Fail-safe -> None (no SGP base / no credential / any error) so the caller + falls back to the fixed default id. + + The credential check is deliberately *form-agnostic*. It used to require + ``x-api-key``, which the shared bot has but a linked user does not: a linked + user's acting headers carry their session cookie instead. That mismatch made + this silently return None for exactly the users this feature is for, so they + would land on the default config while a bot-run turn resolved the name + correctly — a difference in agent behavior with nothing in the logs pointing + at the cause. Forward whatever we hold and let the directory decide. + + Unverified: whether that endpoint accepts cookie auth. If it doesn't, the + request fails and we fall back to the default id, which is the same outcome + as before this change — so this is safe either way, just no longer silently + wrong for api-key callers only. It could not be checked because + ``SLACK_GATEWAY_SGP_BASE_URL`` is unset in dev, which also means this whole + path is inert there today. + """ + has_credential = any( + auth_headers.get(h) for h in ("x-api-key", "cookie", "authorization") + ) + if not (_SGP_BASE_URL and name and has_credential): return None cache_key = (auth_headers.get("x-selected-account-id", ""), name) if cache_key in _CONFIG_ID_CACHE: @@ -692,6 +897,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 +920,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 +960,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/src/temporal/scheduled_agent_run_factory.py b/agentex/src/temporal/scheduled_agent_run_factory.py index aebe244a..1365553b 100644 --- a/agentex/src/temporal/scheduled_agent_run_factory.py +++ b/agentex/src/temporal/scheduled_agent_run_factory.py @@ -67,8 +67,9 @@ def __init__( ) # Default empty = scheduled-run behavior (no live credentials forwarded). # Callers that DO want delegation (e.g. the Slack gateway acting as a user) - # pass headers carrying x-api-key, which build_delegation_headers converts to - # x-acting-user-api-key downstream. + # pass a credential header, which build_delegation_headers converts to its + # x-acting-user-* form downstream: x-api-key for the shared bot, or a + # session cookie for a linked user. self.headers: dict[str, str] = headers or {} @@ -95,10 +96,11 @@ def build_acp_use_case_for_principal( *creator_principal* instead of the request principal. ``request_headers`` (default: none) are forwarded downstream for runtime - delegation — e.g. an ``x-api-key`` becomes ``x-acting-user-api-key`` on the ACP - call so the agent's tools act as that user. Scheduled runs pass nothing (they - deliberately forward no live credentials); the Slack gateway passes the shared - acting-user key. + delegation — an ``x-api-key`` becomes ``x-acting-user-api-key`` on the ACP call, + and a session cookie becomes ``x-acting-user-cookie``, so the agent's tools act + as that identity. Scheduled runs pass nothing (they deliberately forward no live + credentials); the Slack gateway passes the shared bot's key, or the linked user's + session cookie when the turn runs as them. """ env = EnvironmentVariables.refresh() engine = database_async_read_write_engine() diff --git a/agentex/src/utils/session_jwt.py b/agentex/src/utils/session_jwt.py new file mode 100644 index 00000000..7133e4be --- /dev/null +++ b/agentex/src/utils/session_jwt.py @@ -0,0 +1,76 @@ +"""Reading the expiry out of an SGP session JWT. + +The identity-link flow stores the linking user's own session cookie as the +credential it later acts through, so it needs to know when that cookie stops +working. The JWT carries that in its ``exp`` claim, which is more accurate than any +TTL we could invent: the credential's real lifetime belongs to the session, not to +us. + +**This deliberately does not verify the signature.** It is not an authentication +check and must never be used as one. The token arrives on an already-authenticated +request — the auth middleware verified it, upstream, by asking the auth service — +and by the time we get here the only open question is "how long is this good for". +Verifying again would mean holding the signing key, which is precisely the thing +agentex should not have. + +Because the claims are unverified, the expiry is treated as a hint: a *shorter* +expiry than reality only causes an early, recoverable re-link prompt, and a longer +one is caught anyway when the credential is rejected downstream. +""" + +from __future__ import annotations + +import base64 +import binascii +import json +from datetime import UTC, datetime + +from src.utils.logging import make_logger + +logger = make_logger(__name__) + + +def _b64url_decode(segment: str) -> bytes: + # JWT segments are base64url with the padding stripped; put it back or the + # stdlib decoder rejects them. + return base64.urlsafe_b64decode(segment + "=" * (-len(segment) % 4)) + + +def claims(token: str) -> dict | None: + """The JWT's payload claims, or None if it isn't a readable JWT. + + Unverified — see the module docstring. + """ + if not token: + return None + parts = token.split(".") + if len(parts) != 3: + return None + try: + decoded = json.loads(_b64url_decode(parts[1])) + except (ValueError, binascii.Error, UnicodeDecodeError): + return None + return decoded if isinstance(decoded, dict) else None + + +def expires_at(token: str) -> datetime | None: + """When this session token stops being valid, or None if it doesn't say. + + None is a meaningful answer and must not be read as "never expires" — the + caller is expected to substitute a bounded fallback, because storing a + credential with no known expiry is how you end up holding one indefinitely. + """ + payload = claims(token) + if payload is None: + return None + exp = payload.get("exp") + # bool is an int subclass, so it has to be excluded explicitly: exp=True would + # otherwise become 1970-01-01, i.e. a credential that looks already-expired. + if not isinstance(exp, int | float) or isinstance(exp, bool): + return None + try: + return datetime.fromtimestamp(exp, UTC) + except (OverflowError, OSError, ValueError): + # A nonsense exp (far-future, negative) shouldn't crash a link attempt. + logger.warning("[session_jwt] unusable exp claim; treating as unknown") + return None diff --git a/agentex/tests/integration/test_link_nonce_service_redis.py b/agentex/tests/integration/test_link_nonce_service_redis.py new file mode 100644 index 00000000..2abf0fc6 --- /dev/null +++ b/agentex/tests/integration/test_link_nonce_service_redis.py @@ -0,0 +1,225 @@ +"""Integration tests for the link nonce against a real Redis. + +The unit tests for this service run against a hand-written ``_FakeRedis``, so they +assert *our model* of Redis rather than Redis itself. Where that model is wrong, the +unit tests pass and production breaks. The cases here target exactly the places the +model could be wrong: + +- ``decode_responses=False`` (what the app uses), so real Redis returns **bytes** + where the fake returns ``str``. Every read path has to survive that. +- ``GETDEL`` really removing the key, and doing so atomically under concurrency. +- ``KEEPTTL`` really preserving an expiry while rewriting a value. The fake cannot + prove this at all, and the whole "reuse must not extend the lifetime" guarantee + rests on it. +- ``INCR`` / ``EXPIRE`` / ``TTL`` behaving as the send cap assumes, including the + distinction between "exists, no expiry" (-1) and "missing" (-2). + +Depends only on ``redis_url``, deliberately: the broader ``isolated_repositories`` +fixture also starts Postgres and MongoDB, and none of this needs either. That keeps +the tests fast and lets them run in environments where the Mongo image won't boot. +The Redis container is session-scoped and shared, so each test namespaces its keys +by test name rather than flushing the database out from under its neighbours. +""" + +import asyncio +import re + +import pytest +import pytest_asyncio +from src.domain.services import link_nonce_service as mod +from src.domain.services.link_nonce_service import LinkNonceService, LinkRequest + + +@pytest_asyncio.fixture +async def redis(redis_url): + """Client configured the way the application configures it — bytes, not str.""" + import redis.asyncio as aioredis + + client = aioredis.from_url(redis_url, decode_responses=False) + try: + yield client + finally: + await client.aclose() + + +@pytest.fixture +def team(request): + """Key namespace unique to this test; the Redis container is shared.""" + return "T_" + re.sub(r"\W+", "_", request.node.name)[:60] + + +@pytest.fixture +def service(redis): + return LinkNonceService(redis_client=redis) + + +@pytest.fixture +def req(team): + def _make(**kw) -> LinkRequest: + return LinkRequest( + **{ + "provider": "slack", + "external_team_id": team, + "external_user_id": "U1", + "display_name": "@test.user", + "pending_turn": {"text": "what's in my notion?"}, + **kw, + } + ) + + return _make + + +@pytest.mark.integration +@pytest.mark.asyncio +class TestRealBytesHandling: + async def test_payload_round_trips_through_real_redis(self, service, req): + token = await service.create(req()) + got = await service.peek(token) + assert got is not None + assert got.external_user_id == "U1" + assert got.pending_turn == {"text": "what's in my notion?"} + + async def test_reuse_finds_a_pointer_stored_as_bytes( + self, service, req, redis, team + ): + # Real Redis returns the pointer as bytes. If that isn't decoded, reuse + # silently misses and mints a parallel token — the very accumulation this + # service exists to prevent. The fake hands back str, so it cannot catch it. + first = await service.create(req()) + raw = await redis.get(f"link_nonce_user:slack:{team}:U1") + assert isinstance(raw, bytes), "expected a client with decode_responses=False" + + second, reused = await service.create_or_reuse(req()) + assert (second, reused) == (first, True) + + async def test_stored_value_is_json(self, service, req, redis): + token = await service.create(req()) + raw = await redis.get(f"link_nonce:{token}") + # Guards against a refactor to str()/pickle, which a fake-backed round-trip + # would still happily pass. + assert raw.lstrip().startswith(b"{") + + +@pytest.mark.integration +@pytest.mark.asyncio +class TestSingleUseIsReal: + async def test_consume_actually_removes_the_key(self, service, req, redis): + token = await service.create(req()) + assert await service.consume(token) is not None + assert await redis.get(f"link_nonce:{token}") is None + assert await service.consume(token) is None + + async def test_concurrent_consume_yields_exactly_one_winner(self, service, req): + # Two confirms racing on one token: a double-submit, or a retry. GETDEL is + # what makes that safe; a read-then-delete would let both through and mint + # two credentials for one link. + token = await service.create(req()) + results = await asyncio.gather(*(service.consume(token) for _ in range(5))) + assert sum(1 for r in results if r is not None) == 1 + + async def test_superseded_token_is_really_deleted(self, service, req, redis): + first = await service.create(req()) + second = await service.create(req()) + # Genuinely deleted, not merely unreachable: a live leftover is another + # chance for a link to be redeemed by the wrong person, and consuming the + # new one would not invalidate it. + assert await redis.get(f"link_nonce:{first}") is None + assert await redis.get(f"link_nonce:{second}") is not None + + +@pytest.mark.integration +@pytest.mark.asyncio +class TestExpiryIsReal: + async def test_nonce_and_pointer_both_get_a_ttl(self, service, req, redis, team): + token = await service.create(req()) + # -1 = exists with no expiry, -2 = missing. Either is a bug here: a nonce + # that never expires is a permanent bearer token. + for key in (f"link_nonce:{token}", f"link_nonce_user:slack:{team}:U1"): + ttl = await redis.ttl(key) + assert 0 < ttl <= mod._TTL_S, f"{key} ttl={ttl}" + + async def test_reuse_preserves_the_remaining_ttl(self, service, req, redis): + """KEEPTTL, verified against Redis instead of against our own fake. + + The load-bearing case. Reuse rewrites the payload to carry the user's latest + message; if that write dropped the expiry, someone mentioning the agent + every few minutes would keep one token alive indefinitely and the bounded + lifetime — the entire point of a nonce — would be gone. + """ + token = await service.create(req(pending_turn={"text": "first"})) + key = f"link_nonce:{token}" + await redis.expire(key, 60) # stand in for "most of the window has elapsed" + + again, reused = await service.create_or_reuse( + req(pending_turn={"text": "second"}) + ) + assert (again, reused) == (token, True) + + ttl = await redis.ttl(key) + assert 0 < ttl <= 60, f"reuse extended the lifetime: ttl={ttl}" + # ...and the rewrite still landed. + assert (await service.peek(token)).pending_turn == {"text": "second"} + + +@pytest.mark.integration +@pytest.mark.asyncio +class TestSendCapAgainstRealRedis: + async def test_cap_holds_and_counter_expires(self, service, req, redis, team): + r = req() + await service.create_or_reuse(r) + allowed = [await service.claim_send(r) for _ in range(4)] + assert allowed == [True, True, False, False] + + ttl = await redis.ttl(f"link_nonce_sends:slack:{team}:U1") + assert 0 < ttl <= mod._TTL_S, f"send counter ttl={ttl}" + + async def test_concurrent_sends_do_not_exceed_the_cap(self, service, req): + # INCR is atomic, so simultaneous mentions cannot both slip past the cap. + r = req() + await service.create_or_reuse(r) + results = await asyncio.gather(*(service.claim_send(r) for _ in range(10))) + assert sum(1 for x in results if x) == mod._MAX_SENDS + + async def test_a_fresh_nonce_resets_the_counter(self, service, req, redis, team): + r = req() + await service.create_or_reuse(r) + await service.claim_send(r) + await service.claim_send(r) + assert await service.claim_send(r) is False + + await service.create(r) + assert await redis.get(f"link_nonce_sends:slack:{team}:U1") is None + # A genuinely new link must never be silently withheld. + assert await service.claim_send(r) is True + + async def test_counter_left_without_an_expiry_is_repaired( + self, service, req, redis, team + ): + # Simulates a crash between INCR and EXPIRE. Real Redis reports -1 for such + # a key; left alone it would outlive every nonce and the user could never be + # DMed again. + key = f"link_nonce_sends:slack:{team}:U1" + await redis.set(key, "1") + assert await redis.ttl(key) == -1 + + assert await service.claim_send(req()) is True + assert await redis.ttl(key) > 0 + + +@pytest.mark.integration +@pytest.mark.asyncio +class TestIdentityIsolation: + async def test_two_users_linking_at_once_do_not_collide(self, service, req): + a = await service.create(req(external_user_id="U1")) + b = await service.create(req(external_user_id="U2")) + assert a != b + assert (await service.peek(a)).external_user_id == "U1" + assert (await service.peek(b)).external_user_id == "U2" + + async def test_send_budgets_are_independent(self, service, req): + a, b = req(external_user_id="U1"), req(external_user_id="U2") + await service.claim_send(a) + await service.claim_send(a) + assert await service.claim_send(a) is False + assert await service.claim_send(b) is True 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..f1dc038d --- /dev/null +++ b/agentex/tests/unit/api/test_integrations_routes.py @@ -0,0 +1,257 @@ +"""Unit tests for the identity-link routes. + +The credential stored here is the caller's own session cookie, so what needs testing +is mostly refusals: every path where we could end up storing something unusable, or +storing nothing while telling the user they're connected. + +There is no minting to test. Minting was the original design and it cannot work — +identity-service permits one API key per user, every active user already has one, so +the create returns 409 and the existing key's secret can't be read back. +""" + +import base64 +import json +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" + + +def _jwt(exp: datetime | None) -> str: + """A JWT-shaped token. Only the payload matters: nothing verifies the signature, + and nothing should — see src/utils/session_jwt.py.""" + claims = {"sub": "abc"} + if exp is not None: + claims["exp"] = int(exp.timestamp()) + payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip("=") + return f"header.{payload}.signature" + + +_VALID_JWT = _jwt(datetime.now(UTC) + timedelta(days=150)) + + +def _request(principal: dict | None, *, cookie: str | None = None): + """Minimal Request stand-in: what the auth middleware would have populated. + + ``cookie`` defaults to a realistic browser header — the session cookie buried + among unrelated morsels — because that is the case the parser has to survive. + """ + if cookie is None: + cookie = f"_ga=GA1.2.x; _identityJwt={_VALID_JWT}; __utmzz=(not set)" + return SimpleNamespace( + state=SimpleNamespace(principal_context=principal), + headers={"cookie": cookie} if cookie else {}, + ) + + +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 two remaining collaborators: nonce store and repository.""" + 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) + + return SimpleNamespace(nonce=nonce, repo=repo, service=service) + + +@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 "