The Slack account this link was made for and the SGP account " + "you're signed in as belong to different people.
" + "If someone sent you this link, don't use it — it " + "would let their Slack messages run as you. Mention the agent in " + "Slack yourself to get your own link.
", + status=403, + ) + secret = _session_credential(request) if not secret: # The middleware authenticated this caller somehow, but not by a session 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 7f635e24..867abaeb 100644 --- a/agentex/src/domain/use_cases/slack_gateway_use_case.py +++ b/agentex/src/domain/use_cases/slack_gateway_use_case.py @@ -131,6 +131,18 @@ "Connect your account and try again." ) +# Public origin for the link we DM. Must be a host the user's browser can reach AND +# a sibling subdomain of the SGP host, or their session cookie never arrives and the +# callback can't tell who they are. Unset => no offers (a broken link is worse than +# no link). +_PUBLIC_BASE_URL = os.getenv("SLACK_GATEWAY_PUBLIC_BASE_URL", "").rstrip("/") + +# How long before an unlinked user is offered the link again. ``claim_send`` already +# caps DMs at 2 per live nonce, but a nonce only lives ~10 minutes, so without this a +# persistent mentioner re-arms that budget every 10 minutes. Worst case becomes ~2 +# DMs an hour instead of ~12. +_LINK_OFFER_COOLDOWN_S = int(os.getenv("SLACK_LINK_OFFER_COOLDOWN_S", "3600")) + # 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. @@ -360,6 +372,46 @@ def _agent_text(messages) -> str | None: return "\n\n".join(parts) if parts else None +async def slack_user_profile(user_id: str) -> dict[str, str | None]: + """``{"display_name": …, "email": …}`` for a Slack user, best-effort. + + Both fields can be None and callers must cope: ``display_name`` is only used to + make the confirmation page legible, and ``email`` requires the + ``users:read.email`` scope, which is *not* granted by default. A missing email is + therefore "unknown", never "does not match" — see the email-match check in the + link route, which refuses rather than assuming when it can't read one. + + Never raises: an identity-link attempt shouldn't fail because a Slack lookup + hiccupped. + """ + token = os.getenv("SLACK_BOT_TOKEN", "") + if not token or not user_id: + return {"display_name": None, "email": None} + try: + async with httpx.AsyncClient(timeout=10) as client: + resp = await client.get( + "https://slack.com/api/users.info", + headers={"Authorization": f"Bearer {token}"}, + params={"user": user_id}, + ) + body = resp.json() + except Exception: # noqa: BLE001 - best-effort lookup + logger.warning("[slack] users.info failed for %s", user_id, exc_info=True) + return {"display_name": None, "email": None} + if not body.get("ok"): + # missing_scope here means users:read.email isn't granted; that's expected + # until the app is reinstalled, so it's info rather than a warning. + logger.info("[slack] users.info -> %s", body.get("error")) + return {"display_name": None, "email": None} + user = body.get("user") or {} + profile = user.get("profile") or {} + handle = user.get("name") + return { + "display_name": f"@{handle}" if handle else profile.get("real_name"), + "email": profile.get("email"), + } + + # --------------------------------------------------------------------------- use case @@ -614,6 +666,15 @@ async def _run_turn(self, inbound: InboundSlack) -> None: # 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 sgp_user_id is None: + # Not running as a person: either unlinked, or linked with a + # credential we can't use (expired session, undecryptable). Offer the + # link either way — a dead credential needs the same fix as no + # credential. Rate-limited and best-effort; it never affects the turn, + # which continues as the bot below (or is refused just after). + await self._offer_link(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) @@ -1180,6 +1241,156 @@ async def _fetch_bot_token(self) -> str: # Bot token from env / k8s-secret. return os.getenv("SLACK_BOT_TOKEN", "") + async def _offer_link(self, inbound: InboundSlack) -> bool: + """DM the invoking user a one-time link to connect their SGP account. + + Returns True only when a DM actually went out. Entirely best-effort: this runs + alongside a turn that is already proceeding (as the shared bot, or being + refused), and no failure here may change that outcome. + + **The link is DMed, never posted in channel.** The nonce is a bearer token — + whoever holds it gets linked to this Slack identity by signing in as + themselves. In a channel, the first person to click it would bind *this* + user's Slack identity to *their own* SGP account. So if the DM can't be sent + we say so and stop, rather than falling back to somewhere visible. + + Rate limiting is two-layer and deliberately so: ``claim_send`` caps DMs about + one live link (default 2, so a re-mention re-sends rather than going quiet), + and a cooldown key stops a fresh nonce from re-arming that budget on every + mention once the old one expires. + """ + if not _PUBLIC_BASE_URL: + logger.info( + "[slack] link offer skipped: SLACK_GATEWAY_PUBLIC_BASE_URL is unset" + ) + return False + + from src.domain.services.link_nonce_service import LinkNonceService, LinkRequest + + if not await self._claim_offer_cooldown(inbound): + logger.info( + "[slack] link offer suppressed by cooldown for %s", inbound.user + ) + return False + + profile = await slack_user_profile(inbound.user) + request = LinkRequest( + provider="slack", + external_team_id=inbound.team_id, + external_user_id=inbound.user, + display_name=profile.get("display_name") or inbound.user, + # Stored so a later change can answer the original question; nothing + # replays it yet. + pending_turn={ + "text": inbound.text, + "channel": inbound.channel, + "thread_ts": inbound.thread_ts, + }, + ) + + service = LinkNonceService() + try: + token, reused = await service.create_or_reuse(request) + allowed = await service.claim_send(request) + except Exception: # noqa: BLE001 - Redis down: no offer, turn unaffected + logger.warning("[slack] link offer failed to mint a nonce", exc_info=True) + return False + + if not allowed: + # Already DMed about this link. Acknowledge in-channel (ephemerally) so + # the user isn't left wondering, but don't send another DM. + await self._post_ephemeral( + inbound, + "I've already sent you a DM with a link to connect your account — " + "check your direct messages with me.", + ) + return False + + opened = await self._slack_api("conversations.open", {"users": inbound.user}) + dm_channel = ( + (opened.get("channel") or {}).get("id") if opened.get("ok") else None + ) + if not dm_channel: + logger.warning( + "[slack] conversations.open failed for %s: %s", + inbound.user, + opened.get("error"), + ) + return False + + url = f"{_PUBLIC_BASE_URL}/integrations/slack/link?nonce={token}" + posted = await self._slack_api( + "chat.postMessage", + { + "channel": dm_channel, + "unfurl_links": False, + "text": ( + "Connect your SGP account and I'll use *your* tools " + "(Notion, Linear, …) when you ask me things in Slack.\n\n" + f"<{url}|Connect my account>\n\n" + "This link is just for you and expires in a few minutes. " + "Don't forward it — anyone who opens it could connect your " + "Slack identity to their own SGP account." + ), + }, + ) + if not posted.get("ok"): + logger.warning( + "[slack] link DM failed for %s: %s", inbound.user, posted.get("error") + ) + return False + + logger.info( + "[slack] link offer DMed to %s (nonce %s)", + inbound.user, + "reused" if reused else "new", + ) + await self._post_ephemeral( + inbound, + "I've DM'd you a link to connect your SGP account — once you do, I'll " + "use your own tools when you ask me things here.", + ) + return True + + async def _claim_offer_cooldown(self, inbound: InboundSlack) -> bool: + """True if we may offer this user a link now, and records the offer. + + Fails *open* on a Redis problem: the alternative is never offering, and the + per-link ``claim_send`` cap still bounds the damage. + """ + key = f"slack:link_offer:{inbound.team_id}:{inbound.user}" + try: + pool = GlobalDependencies().redis_pool + if pool is None: + return True + import redis.asyncio as redis + + client = redis.Redis(connection_pool=pool) + # SET NX: only the first caller in the window wins. + return bool(await client.set(key, "1", nx=True, ex=_LINK_OFFER_COOLDOWN_S)) + except Exception: # noqa: BLE001 - see docstring + logger.warning("[slack] link offer cooldown check failed", exc_info=True) + return True + + async def _post_ephemeral(self, inbound: InboundSlack, text: str) -> None: + """Post a message only the invoking user sees. Best-effort. + + Ephemeral so a channel isn't cluttered with onboarding nudges aimed at one + person — and Slack rejects it outside a channel context (e.g. an assistant + pane), which we swallow. + """ + body = await self._slack_api( + "chat.postEphemeral", + { + "channel": inbound.channel, + "user": inbound.user, + "thread_ts": inbound.thread_ts, + "text": text, + }, + ) + if not body.get("ok"): + logger.info("[slack] postEphemeral -> %s", body.get("error")) + async def _set_status(self, inbound: InboundSlack, status: str) -> None: """AI-app 'thinking…' indicator (assistant.threads.setStatus). Shows in the assistant pane while the turn runs; cleared when the reply is posted. Best-effort: diff --git a/agentex/tests/unit/api/test_integrations_routes.py b/agentex/tests/unit/api/test_integrations_routes.py index f1dc038d..ed0d5dca 100644 --- a/agentex/tests/unit/api/test_integrations_routes.py +++ b/agentex/tests/unit/api/test_integrations_routes.py @@ -255,3 +255,82 @@ async def test_unconfigured_encryption_key_is_reported_not_a_500(self, wiring): ) # Nonce preserved so the link works once the key is configured. wiring.nonce.consume.assert_not_awaited() + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestEmailMatch: + """The flagged defence against a *forwarded* link. + + The nonce stops an attacker forging someone else's Slack identity. It does not + stop them sending their OWN link to a victim: if the victim clicks it while + signed in, the attacker's Slack identity binds to the victim's SGP account, and + from then on the attacker's Slack messages run as the victim. Comparing the two + accounts' emails is what closes that. + + Off by default: it needs the `users:read.email` Slack scope, which isn't granted + until the app is reinstalled. + """ + + def _slack_email(self, monkeypatch, email): + import src.domain.use_cases.slack_gateway_use_case as sg + + monkeypatch.setattr( + sg, "slack_user_profile", AsyncMock(return_value={"email": email}) + ) + + async def test_disabled_by_default_does_not_call_slack(self, wiring, monkeypatch): + import src.domain.use_cases.slack_gateway_use_case as sg + + probe = AsyncMock(return_value={"email": "someone.else@example.com"}) + monkeypatch.setattr(sg, "slack_user_profile", probe) + monkeypatch.setattr(mod, "_REQUIRE_EMAIL_MATCH", False) + + resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + assert resp.status_code == 200 + probe.assert_not_awaited() + + async def test_matching_emails_link_successfully(self, wiring, monkeypatch): + monkeypatch.setattr(mod, "_REQUIRE_EMAIL_MATCH", True) + self._slack_email(monkeypatch, _SGP_EMAIL) + resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + assert resp.status_code == 200 + wiring.repo.upsert_link.assert_awaited_once() + + async def test_match_is_case_insensitive(self, wiring, monkeypatch): + monkeypatch.setattr(mod, "_REQUIRE_EMAIL_MATCH", True) + self._slack_email(monkeypatch, _SGP_EMAIL.upper()) + resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + assert resp.status_code == 200 + + async def test_mismatch_is_refused_and_stores_nothing(self, wiring, monkeypatch): + monkeypatch.setattr(mod, "_REQUIRE_EMAIL_MATCH", True) + self._slack_email(monkeypatch, "attacker@example.com") + + resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + + assert resp.status_code == 403 + wiring.repo.upsert_link.assert_not_awaited() + # The nonce survives, so a legitimate owner can still use their own link. + wiring.nonce.consume.assert_not_awaited() + body = resp.body.decode().lower() + assert "don't use it" in body or "don't use it" in body + + async def test_unreadable_slack_email_fails_closed(self, wiring, monkeypatch): + # Missing scope, deleted user, API hiccup -> None. Treating that as "skip the + # check" would silently disable the defence the moment the scope lapsed. + monkeypatch.setattr(mod, "_REQUIRE_EMAIL_MATCH", True) + self._slack_email(monkeypatch, None) + + resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + assert resp.status_code == 403 + wiring.repo.upsert_link.assert_not_awaited() + + async def test_missing_sgp_email_fails_closed(self, wiring, monkeypatch): + monkeypatch.setattr(mod, "_REQUIRE_EMAIL_MATCH", True) + self._slack_email(monkeypatch, "someone@example.com") + principal = {**_PRINCIPAL, "raw_user": {}} + + resp = await mod.slack_link_confirm(_request(principal), nonce="tok") + assert resp.status_code == 403 + wiring.repo.upsert_link.assert_not_awaited() diff --git a/agentex/tests/unit/use_cases/test_slack_gateway_use_case.py b/agentex/tests/unit/use_cases/test_slack_gateway_use_case.py index 5a655797..2b4dc740 100644 --- a/agentex/tests/unit/use_cases/test_slack_gateway_use_case.py +++ b/agentex/tests/unit/use_cases/test_slack_gateway_use_case.py @@ -1683,3 +1683,123 @@ async def test_unlinked_first_turn_omits_the_sgp_user_id(self, monkeypatch): create = acp.handle_rpc_request.await_args_list[0].kwargs["params"] assert create.name == "slack:1" assert "sgp_user_id" not in create.task_metadata + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestLinkOffer: + """DMing an unlinked user a connect link. + + The security-critical property under test is that the link goes to a DM and + NOWHERE else. The nonce is a bearer token: whoever opens it gets linked to this + Slack identity by signing in as themselves, so a link posted in a channel lets + the first reader bind someone else's Slack identity to their own SGP account. + """ + + def _wire(self, monkeypatch, *, allowed=True, cooldown_ok=True, open_ok=True): + uc = SlackGatewayUseCase() + monkeypatch.setattr(sg, "_PUBLIC_BASE_URL", "https://agentex.example.com") + monkeypatch.setattr( + sg, "slack_user_profile", AsyncMock(return_value={"display_name": "@ada"}) + ) + monkeypatch.setattr( + uc, "_claim_offer_cooldown", AsyncMock(return_value=cooldown_ok) + ) + + nonce = MagicMock() + nonce.create_or_reuse = AsyncMock(return_value=("TOKEN123", False)) + nonce.claim_send = AsyncMock(return_value=allowed) + monkeypatch.setattr( + "src.domain.services.link_nonce_service.LinkNonceService", + lambda *a, **k: nonce, + ) + + def _api(method, payload): + if method == "conversations.open": + return ( + {"ok": True, "channel": {"id": "D_DM"}} + if open_ok + else {"ok": False, "error": "cannot_dm_bot"} + ) + return {"ok": True} + + api = AsyncMock(side_effect=_api) + monkeypatch.setattr(uc, "_slack_api", api) + return uc, api, nonce + + async def test_dms_the_link_and_acknowledges_in_channel(self, monkeypatch): + uc, api, _ = self._wire(monkeypatch) + assert await uc._offer_link(_inbound()) is True + + calls = dict(c.args for c in api.await_args_list) + assert set(calls) == { + "conversations.open", + "chat.postMessage", + "chat.postEphemeral", + } + dm = calls["chat.postMessage"] + assert dm["channel"] == "D_DM" + assert "TOKEN123" in dm["text"] + assert "/integrations/slack/link?nonce=" in dm["text"] + # The nudge is ephemeral, so a channel isn't littered with one person's + # onboarding. + assert calls["chat.postEphemeral"]["user"] == "U1" + + async def test_the_link_never_goes_to_the_origin_channel(self, monkeypatch): + uc, api, _ = self._wire(monkeypatch) + await uc._offer_link(_inbound(channel="C_PUBLIC")) + for method, payload in (c.args for c in api.await_args_list): + if payload.get("channel") == "C_PUBLIC": + assert "TOKEN123" not in str( + payload + ), f"{method} leaked the nonce into the origin channel" + + async def test_dm_warns_against_forwarding(self, monkeypatch): + uc, api, _ = self._wire(monkeypatch) + await uc._offer_link(_inbound()) + dm = next( + p + for m, p in (c.args for c in api.await_args_list) + if m == "chat.postMessage" + ) + assert "forward" in dm["text"].lower() + + async def test_no_offer_without_a_public_base_url(self, monkeypatch): + uc, api, _ = self._wire(monkeypatch) + monkeypatch.setattr(sg, "_PUBLIC_BASE_URL", "") + # A link the user can't reach is worse than no link. + assert await uc._offer_link(_inbound()) is False + api.assert_not_awaited() + + async def test_failed_dm_does_not_fall_back_to_the_channel(self, monkeypatch): + uc, api, _ = self._wire(monkeypatch, open_ok=False) + assert await uc._offer_link(_inbound()) is False + for _method, payload in (c.args for c in api.await_args_list): + assert "TOKEN123" not in str(payload) + + async def test_send_cap_reached_says_so_without_a_second_dm(self, monkeypatch): + uc, api, _ = self._wire(monkeypatch, allowed=False) + assert await uc._offer_link(_inbound()) is False + methods = [m for m, _ in (c.args for c in api.await_args_list)] + # Acknowledge in-channel rather than going silent, but send no new DM. + assert methods == ["chat.postEphemeral"] + + async def test_cooldown_suppresses_the_offer_entirely(self, monkeypatch): + uc, api, nonce = self._wire(monkeypatch, cooldown_ok=False) + assert await uc._offer_link(_inbound()) is False + nonce.create_or_reuse.assert_not_awaited() + api.assert_not_awaited() + + async def test_redis_failure_is_swallowed(self, monkeypatch): + # The turn is already proceeding; a nonce-store outage must not break it. + uc, api, nonce = self._wire(monkeypatch) + nonce.create_or_reuse = AsyncMock(side_effect=RuntimeError("redis down")) + assert await uc._offer_link(_inbound()) is False + + async def test_pending_turn_is_stashed_for_later_replay(self, monkeypatch): + uc, _api, nonce = self._wire(monkeypatch) + await uc._offer_link(_inbound(text="what's in my linear?", channel="C9")) + req = nonce.create_or_reuse.await_args.args[0] + assert req.external_user_id == "U1" + assert req.pending_turn["text"] == "what's in my linear?" + assert req.pending_turn["channel"] == "C9"