refactor(agentex): cut the identity-link config surface from 8 vars to 1

The identity-link work accumulated eight environment variables. One was a real
hazard, one was a trap, and six were knobs nobody has ever turned. Removing them
also removes two failure modes.

The hazard: IDENTITY_LINK_SESSION_COOKIE_NAME

The session cookie name was configurable in two places -- here and
AGENTEX_DELEGATION_SESSION_COOKIE_NAMES in delegation_headers. acting_headers()
emits a Cookie header that build_delegation_headers then filters down to its
allowlist, so if the two ever disagreed the credential would be stripped in transit
and every linked turn would silently lose its acting identity, while the link sat
in the database looking stored, valid and healthy.

It is now derived: session_cookie_name() reads the delegation allowlist. One
source of truth, so divergence is unrepresentable rather than merely documented.
Empty allowlist (cookie delegation disabled) returns None and acting_headers()
refuses, since emitting a credential that will certainly be stripped is worse than
admitting we cannot act.

The trap: IDENTITY_LINK_REQUIRE_EMAIL_MATCH

The email check needs the users:read.email Slack scope, which isn't granted. The
flag existed to keep it off until the scope lands -- but flag and scope then had to
be flipped together: the flag alone refused every link (unreadable email treated as
mismatch), and the scope alone protected nothing.

It now enables itself. _email_mismatch() enforces whenever Slack answers with an
email and stands down when it won't, so granting the scope switches the protection
on with no config change and no ordering hazard.

That inverts the unverifiable case from refuse to allow, which is weaker, and
deliberately so: with no flag to distinguish "scope missing" from "Slack had a bad
minute", failing closed would make linking fail at random. The gap is not
attacker-reachable -- nobody outside our infrastructure influences whether our own
Slack lookup succeeds -- and the previous shipped state (flag off) verified nothing
at all, so this is strictly stronger than what it replaces.

The knobs -> module constants

IDENTITY_LINK_NONCE_TTL, _MAX_DMS, _CACHE_TTL, _NEGATIVE_CACHE_TTL,
_FALLBACK_TTL_DAYS and SLACK_LINK_OFFER_COOLDOWN_S are now constants at their
former defaults. Each was a config surface and a branch carrying a value that has
never been set to anything else, and therefore never tested at anything else. They
crept in by pattern-matching the surrounding file, which isn't a reason.

What remains: AGENTEX_CREDENTIAL_ENCRYPTION_KEY and SLACK_GATEWAY_PUBLIC_BASE_URL,
both deployment-specific with no sensible default, plus the pre-existing
SLACK_GATEWAY_REQUIRE_LINKED_USER, which is a genuine product choice.

No behavior change at current settings: every constant equals the default it
replaced, and the email check's effective behavior in production (no scope, flag
off -> no verification) is unchanged until the scope is granted.

Testing: 10 new unit tests. The cookie-name ones assert the end-to-end property
that whatever name is configured, what acting_headers emits is what
build_delegation_headers forwards -- and that a disabled allowlist refuses rather
than emitting something that gets stripped. The email ones pin both directions of
the asymmetry: verified-different refuses (and leaves the nonce intact for the
legitimate owner), while missing scope, lookup failure and a principal without an
email all allow. Full unit suite 691 passed; the 14 Redis integration tests still
pass against a real Redis after the TTL constants moved.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: michael-chou359

agentex/src/api/routes/integrations.py

@@ -41,7 +41,6 @@
 from __future__ import annotations
 
 import html
-import os
 from datetime import UTC, datetime, timedelta
 
 from fastapi import APIRouter, Form, Request
@@ -55,8 +54,8 @@
 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,
+    session_cookie_name,
 )
 from src.domain.services.link_nonce_service import LinkNonceService
 from src.utils import session_jwt
@@ -70,27 +69,10 @@
 # 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"))
-
-# Require the Slack account's email to match the signed-in SGP account's.
-#
-# This is the only real defence against a *forwarded* link. The nonce stops an
-# attacker forging someone else's Slack identity, but nothing stops 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 thereafter the
-# attacker's Slack messages run as the victim, using the victim's integrations. The
-# confirmation page names both identities, which catches a mis-click but reduces to
-# user vigilance against a deliberate attempt.
-#
-# OFF by default because it needs the ``users:read.email`` Slack scope, which is not
-# granted until the app is reinstalled. Enabling it without the scope would refuse
-# every link (the check treats an unreadable email as a mismatch, deliberately), so
-# the flag and the scope have to be turned on together.
-_REQUIRE_EMAIL_MATCH = os.getenv("IDENTITY_LINK_REQUIRE_EMAIL_MATCH", "").lower() in (
-    "1",
-    "true",
-    "yes",
-)
+_FALLBACK_TTL_DAYS = 30
+
+# Email matching has no flag: it enforces itself whenever Slack will tell us the
+# email, and stands down when it won't. See ``_email_mismatch``.
 
 
 def _page(title: str, body: str, *, status: int = 200) -> HTMLResponse:
@@ -155,15 +137,71 @@ def _session_credential(request: Request) -> str | None:
     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``.
+
+    The name comes from the delegation allowlist, so what we store here is by
+    construction what the delegation layer will forward later. None when cookie
+    delegation is disabled: there would be no way to act through the credential, so
+    there is no point storing one.
     """
+    wanted = session_cookie_name()
+    if wanted is None:
+        return None
     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:
+        if sep and name.strip() == wanted:
             return value.strip() or None
     return None
 
 
+async def _email_mismatch(external_user_id: str, sgp_email: str | None) -> bool:
+    """True when Slack and SGP demonstrably identify different people.
+
+    This is the only real 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, who — clicking it while signed in — would bind the
+    attacker's Slack identity to their SGP account, after which the attacker's Slack
+    messages run as them with their integrations.
+
+    **Self-enabling, with no flag.** The check needs the ``users:read.email`` Slack
+    scope, which may not be granted. Rather than gate that on configuration — where
+    the flag and the scope must be flipped together, and flipping one alone either
+    breaks every link or silently protects nothing — it simply enforces whenever
+    Slack answers with an email and stands down when it won't. Granting the scope
+    turns the protection on by itself.
+
+    So the asymmetry is deliberate: **verified different -> refuse; unverifiable ->
+    allow and warn.** Failing closed on an unreadable email would be stronger, but
+    with no flag to distinguish "scope missing" from "Slack had a bad minute" it
+    would make linking fail randomly. The gap it leaves is not attacker-reachable:
+    nobody outside our infrastructure influences whether our own Slack lookup
+    succeeds.
+    """
+    from src.domain.use_cases.slack_gateway_use_case import slack_user_profile
+
+    profile = await slack_user_profile(external_user_id)
+    slack_email = profile.get("email")
+    if not slack_email:
+        logger.warning(
+            "identity link: email not verified (Slack would not tell us)",
+            extra={
+                "external_user_id": external_user_id,
+                "slack_error": profile.get("error"),
+                "hint": "grant users:read.email to enable this check",
+            },
+        )
+        return False
+    if not sgp_email:
+        # Slack gave us an email but the SGP session didn't. Nothing to compare, so
+        # the same rule applies: can't verify, don't block.
+        logger.warning(
+            "identity link: email not verified (no email on the SGP principal)",
+            extra={"external_user_id": external_user_id},
+        )
+        return False
+    return slack_email.strip().lower() != sgp_email.strip().lower()
+
+
 @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
@@ -267,47 +305,34 @@ async def slack_link_confirm(request: Request, nonce: str = Form("")) -> HTMLRes
             status=409,
         )
 
-    if _REQUIRE_EMAIL_MATCH:
-        # Local import: the gateway module owns the Slack token and HTTP calls, and
-        # importing it at module load would pull the use case into the route's import
-        # graph for a feature that is off by default.
-        from src.domain.use_cases.slack_gateway_use_case import slack_user_profile
-
-        slack_email = (await slack_user_profile(link_request.external_user_id)).get(
-            "email"
+    if await _email_mismatch(link_request.external_user_id, email):
+        logger.warning(
+            "identity link refused: Slack/SGP email mismatch",
+            extra={
+                "sgp_user_id": sgp_user_id,
+                "external_user_id": link_request.external_user_id,
+            },
+        )
+        return _page(
+            "Accounts don't match",
+            "<h1>Those accounts don't match</h1>"
+            "<p>The Slack account this link was made for and the SGP account "
+            "you're signed in as belong to different people.</p>"
+            "<p class=muted>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.</p>",
+            status=403,
         )
-        # An unreadable email is treated as a mismatch, not as "skip the check".
-        # Failing open here would silently disable the only defence against a
-        # forwarded link the moment the Slack scope lapsed.
-        if not slack_email or not email or slack_email.lower() != email.lower():
-            logger.warning(
-                "identity link refused: Slack/SGP email mismatch",
-                extra={
-                    "sgp_user_id": sgp_user_id,
-                    "external_user_id": link_request.external_user_id,
-                    "slack_email_known": bool(slack_email),
-                },
-            )
-            return _page(
-                "Accounts don't match",
-                "<h1>Those accounts don't match</h1>"
-                "<p>The Slack account this link was made for and the SGP account "
-                "you're signed in as belong to different people.</p>"
-                "<p class=muted>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.</p>",
-                status=403,
-            )
 
     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.
+        # cookie — an api-key or bearer caller, a cookie under a different name, or
+        # cookie delegation switched off entirely. 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},
+            extra={"sgp_user_id": sgp_user_id, "cookie": session_cookie_name()},
         )
         return _page(
             "Couldn't read your session",

agentex/src/domain/services/identity_link_service.py

@@ -23,12 +23,12 @@
 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.delegation_headers import session_cookie_names_to_forward
 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
@@ -38,21 +38,33 @@
 
 # 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"))
+_CACHE_TTL_S = 300
+_NEGATIVE_CACHE_TTL_S = 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 session_cookie_name() -> str | None:
+    """The cookie name to store and emit, or None if cookie delegation is off.
+
+    **Derived from the delegation allowlist, never separately configurable.** The
+    stored credential leaves as a Cookie header that ``build_delegation_headers``
+    filters down to its allowlisted names before re-emitting as
+    ``x-acting-user-cookie``. If this name and that allowlist could disagree, the
+    credential would be dropped in transit and every linked turn would silently lose
+    its acting identity — with a stored, valid, apparently-healthy link. One source
+    of truth removes that failure mode entirely.
+
+    None when the allowlist is empty (cookie delegation explicitly disabled): there
+    is then no way to act through a stored session, so callers must refuse rather
+    than store or emit something that cannot work.
+    """
+    names = session_cookie_names_to_forward()
+    return names[0] if names else None
 
 
 def _cache_key(
@@ -172,8 +184,18 @@ async def acting_headers(self, identity: ResolvedIdentity) -> dict[str, str] | N
                 extra={"sgp_user_id": identity.sgp_user_id},
             )
             return None
+        cookie_name = session_cookie_name()
+        if cookie_name is None:
+            # Cookie delegation is disabled, so build_delegation_headers would strip
+            # whatever we emit. Refuse rather than hand back headers that get
+            # silently dropped between here and the agent.
+            logger.warning(
+                "identity_link_cookie_delegation_disabled",
+                extra={"sgp_user_id": identity.sgp_user_id},
+            )
+            return None
         return {
-            HEADER_COOKIE: f"{SESSION_COOKIE_NAME}={credential}",
+            HEADER_COOKIE: f"{cookie_name}={credential}",
             HEADER_SELECTED_ACCOUNT_ID: identity.sgp_account_id,
         }
 

agentex/src/domain/services/link_nonce_service.py

@@ -40,7 +40,6 @@
 from __future__ import annotations
 
 import json
-import os
 import secrets
 from dataclasses import asdict, dataclass, field, replace
 from typing import Annotated, Any
@@ -53,7 +52,7 @@
 
 # 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"))
+_TTL_S = 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.
@@ -63,7 +62,7 @@
 # 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"))
+_MAX_SENDS = 2
 
 _KEY_PREFIX = "link_nonce:"
 # identity -> its one live token, so a second mention finds the first nonce instead

agentex/src/domain/use_cases/slack_gateway_use_case.py

@@ -141,7 +141,7 @@
 # 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"))
+_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
@@ -373,20 +373,20 @@ def _agent_text(messages) -> str | None:
 
 
 async def slack_user_profile(user_id: str) -> dict[str, str | None]:
-    """``{"display_name": …, "email": …}`` for a Slack user, best-effort.
+    """``{"display_name": …, "email": …, "error": …}`` for a Slack user.
 
-    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.
+    All three can be None. ``display_name`` only makes the confirmation page legible.
+    ``email`` requires the ``users:read.email`` scope, which may not be granted —
+    ``error`` is how a caller tells "Slack won't tell us" apart from "Slack told us
+    and there's no email", which the link route needs to decide whether it can
+    perform its identity check at all.
 
     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}
+        return {"display_name": None, "email": None, "error": "no_token"}
     try:
         async with httpx.AsyncClient(timeout=10) as client:
             resp = await client.get(
@@ -397,18 +397,19 @@ async def slack_user_profile(user_id: str) -> dict[str, str | None]:
         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}
+        return {"display_name": None, "email": None, "error": "request_failed"}
     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.
+        # missing_scope means users:read.email isn't granted — expected until the app
+        # is reinstalled, so info rather than warning.
         logger.info("[slack] users.info -> %s", body.get("error"))
-        return {"display_name": None, "email": None}
+        return {"display_name": None, "email": None, "error": body.get("error")}
     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"),
+        "error": None,
     }