feat(openapi): distinguish expired OAuth bearer from invalid token

Previously an expired OAuth bearer and an unknown/invalid one both
surfaced as an indistinguishable generic 401 (and an invalid token
actually leaked a 500), so a client could not tell "session expired,
re-authenticate" apart from "never authenticated."

The resolver now raises a distinct TokenExpiredError for expired DB
rows and records a separate `expired` negative-cache marker, so a
retry within the negative-cache TTL still reports expiry instead of
collapsing into a generic miss. The auth pipeline maps the two domain
errors to unified OpenApiError responses: SessionExpired (code
`token_expired`) and InvalidBearer (code `unauthorized`), both 401.
This also fixes the latent 500 on invalid bearers.

The `token_expired` code is synced through the contract codegen into
the generated types/zod, and the difyctl error mapper branches the
401 on it. The CLI `expired_token` taxonomy member (RFC 8628
device-flow code expiry) is merged into `token_expired`; the RFC 8628
wire value is unchanged.

Closes WTA-1062

api/controllers/openapi/_errors.py

@@ -34,6 +34,7 @@ class OpenApiErrorCode(StrEnum):
     # transport-generic (resolved from HTTP status for plain werkzeug raises)
     BAD_REQUEST = "bad_request"
     UNAUTHORIZED = "unauthorized"
+    TOKEN_EXPIRED = "token_expired"
     FORBIDDEN = "forbidden"
     NOT_FOUND = "not_found"
     METHOD_NOT_ALLOWED = "method_not_allowed"
@@ -223,6 +224,19 @@ class OpenApiErrorFormatter:
         return isinstance(part, (str, int)) and not isinstance(part, bool)
 
 
+class InvalidBearer(OpenApiError):  # noqa: N818
+    code = 401
+    error_code = OpenApiErrorCode.UNAUTHORIZED
+    description = "Invalid or unknown bearer token."
+
+
+class SessionExpired(OpenApiError):  # noqa: N818
+    code = 401
+    error_code = OpenApiErrorCode.TOKEN_EXPIRED
+    description = "Your session has expired."
+    hint = "Re-authenticate to continue (e.g. re-run your login command)."
+
+
 class FilenameNotExists(OpenApiError):  # noqa: N818
     code = 400
     error_code = OpenApiErrorCode.FILENAME_NOT_EXISTS

api/controllers/openapi/auth/pipeline.py

@@ -17,6 +17,7 @@ from flask_login import user_logged_in
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
 from controllers.openapi._audit import emit_wrong_surface
+from controllers.openapi._errors import InvalidBearer, SessionExpired
 from controllers.openapi.auth.data import (
     AuthData,
     Edition,
@@ -28,7 +29,9 @@ from controllers.openapi.auth.data import (
 from controllers.openapi.auth.flow import When
 from libs.oauth_bearer import (
     AuthContext,
+    InvalidBearerError,
     Scope,
+    TokenExpiredError,
     TokenType,
     extract_bearer,
     get_authenticator,
@@ -217,7 +220,12 @@ class PipelineRouter:
         if not token:
             raise Unauthorized("bearer required")
 
-        identity = get_authenticator().authenticate(token)
+        try:
+            identity = get_authenticator().authenticate(token)
+        except TokenExpiredError:
+            raise SessionExpired()
+        except InvalidBearerError:
+            raise InvalidBearer()
 
         if allowed_token_types is not None and identity.token_type not in allowed_token_types:
             emit_wrong_surface(

api/libs/oauth_bearer.py

@@ -236,6 +236,16 @@ class TokenExpiredError(Exception):
     """Hard-expire bookkeeping is the resolver's job before raising."""
 
 
+class NegativeCache(StrEnum):
+    """Negative cache markers. ``EXPIRED`` is distinct from ``INVALID`` so a
+    retry inside ``NEGATIVE_TTL`` still reports expiry instead of collapsing
+    into a generic unknown-token miss.
+    """
+
+    INVALID = "invalid"
+    EXPIRED = "expired"
+
+
 # ============================================================================
 # Registry
 # ============================================================================
@@ -343,13 +353,15 @@ class OAuthAccessTokenResolver:
     def _cache_key(self, token_hash: str) -> str:
         return TOKEN_CACHE_KEY_FMT.format(hash=token_hash)
 
-    def cache_get(self, token_hash: str) -> ResolvedRow | None | Literal["invalid"]:
+    def cache_get(self, token_hash: str) -> ResolvedRow | None | NegativeCache:
         raw = self._redis.get(self._cache_key(token_hash))
         if raw is None:
             return None
         text = raw.decode() if isinstance(raw, (bytes, bytearray)) else raw
-        if text == "invalid":
-            return "invalid"
+        try:
+            return NegativeCache(text)
+        except ValueError:
+            pass
         try:
             return ResolvedRow.from_cache(json.loads(text))
         except (ValueError, KeyError):
@@ -363,8 +375,8 @@ class OAuthAccessTokenResolver:
             json.dumps(row.to_cache()),
         )
 
-    def cache_set_negative(self, token_hash: str) -> None:
-        self._redis.setex(self._cache_key(token_hash), self._negative_ttl, "invalid")
+    def cache_set_negative(self, token_hash: str, marker: NegativeCache = NegativeCache.INVALID) -> None:
+        self._redis.setex(self._cache_key(token_hash), self._negative_ttl, str(marker))
 
     def hard_expire(self, session: Session, row_id: uuid.UUID | str, token_hash: str) -> None:
         """Atomic CAS — only the worker that flips revoked_at emits audit;
@@ -385,7 +397,7 @@ class OAuthAccessTokenResolver:
                 extra={"audit": True, "token_id": str(row_id)},
             )
         self._redis.delete(self._cache_key(token_hash))
-        self.cache_set_negative(token_hash)
+        self.cache_set_negative(token_hash, NegativeCache.EXPIRED)
 
 
 class _VariantResolver:
@@ -395,9 +407,11 @@ class _VariantResolver:
 
     def resolve(self, token_hash: str) -> ResolvedRow | None:
         cached = self._parent.cache_get(token_hash)
-        if cached == "invalid":
+        if isinstance(cached, NegativeCache):
+            if cached is NegativeCache.EXPIRED:
+                raise TokenExpiredError("token_expired")
             return None
-        if cached is not None and not isinstance(cached, str):
+        if cached is not None:
             if not self._matches_variant(cached):
                 return None
             return cached
@@ -413,7 +427,7 @@ class _VariantResolver:
         now = datetime.now(UTC)
         if row.expires_at is not None and row.expires_at <= now:
             self._parent.hard_expire(session, row.id, token_hash)
-            return None
+            raise TokenExpiredError("token_expired")
 
         if not self._matches_variant_model(row):
             logger.error(
@@ -472,7 +486,7 @@ def record_layer0_verdict(token_hash: str, tenant_id: str, verdict: bool) -> Non
     if raw is None:
         return
     text = raw.decode() if isinstance(raw, (bytes, bytearray)) else raw
-    if text == "invalid":
+    if text in (NegativeCache.INVALID, NegativeCache.EXPIRED):
         return
     try:
         data = json.loads(text)
@@ -601,6 +615,8 @@ def validate_bearer(*, accept: frozenset[Accepts]) -> Callable[[Callable[_DP, _D
 
             try:
                 ctx = get_authenticator().authenticate(token)
+            except TokenExpiredError:
+                raise Unauthorized("token_expired")
             except InvalidBearerError as e:
                 raise Unauthorized(str(e))
 

api/tests/integration_tests/controllers/openapi/conftest.py

@@ -72,6 +72,7 @@ def mint_token(flask_app: Flask):
         prefix: str,
         subject_email: str,
         subject_issuer: str | None,
+        expires_at: datetime | None = None,
     ) -> OAuthAccessToken:
         with flask_app.app_context():
             row = OAuthAccessToken(
@@ -82,7 +83,7 @@ def mint_token(flask_app: Flask):
                 subject_issuer=subject_issuer,
                 client_id="difyctl",
                 device_label="test-device",
-                expires_at=datetime.now(UTC) + timedelta(hours=1),
+                expires_at=expires_at or (datetime.now(UTC) + timedelta(hours=1)),
             )
             db.session.add(row)
             db.session.commit()
@@ -111,6 +112,21 @@ def account_token(workspace_account, mint_token) -> str:
     return token
 
 
+@pytest.fixture
+def expired_account_token(workspace_account, mint_token) -> str:
+    account, _, _ = workspace_account
+    token = "dfoa_" + uuid.uuid4().hex
+    mint_token(
+        token,
+        account_id=account.id,
+        prefix="dfoa_",
+        subject_email=account.email,
+        subject_issuer="dify:account",
+        expires_at=datetime.now(UTC) - timedelta(minutes=1),
+    )
+    return token
+
+
 @pytest.fixture(autouse=True)
 def _flush_auth_redis(flask_app: Flask) -> Generator[None, None, None]:
     def _flush():

api/tests/integration_tests/controllers/openapi/test_auth.py

@@ -6,6 +6,7 @@ acceptance/rejection on app-scoped routes.
 
 from __future__ import annotations
 
+import uuid
 from collections.abc import Generator
 
 import pytest
@@ -16,6 +17,50 @@ from extensions.ext_database import db
 from models import App, Tenant
 
 
+def test_expired_token_returns_401_token_expired(
+    test_client: FlaskClient,
+    expired_account_token: str,
+) -> None:
+    """An expired bearer is distinguishable from an unknown one: 401 with the
+    domain code ``token_expired`` (+ actionable hint), not a generic 401 or 500."""
+    res = test_client.get(
+        "/openapi/v1/account",
+        headers={"Authorization": f"Bearer {expired_account_token}"},
+    )
+    assert res.status_code == 401
+    assert res.json["code"] == "token_expired"
+    assert res.json["hint"]
+
+
+def test_expired_token_replay_stays_token_expired(
+    test_client: FlaskClient,
+    expired_account_token: str,
+) -> None:
+    """The distinct ``expired`` negative-cache marker keeps the second hit (served
+    from cache, inside NEGATIVE_TTL) reporting ``token_expired`` rather than
+    collapsing into a generic unknown-token 401."""
+    headers = {"Authorization": f"Bearer {expired_account_token}"}
+    first = test_client.get("/openapi/v1/account", headers=headers)
+    second = test_client.get("/openapi/v1/account", headers=headers)
+    assert first.json["code"] == "token_expired"
+    assert second.status_code == 401
+    assert second.json["code"] == "token_expired"
+
+
+def test_unknown_token_returns_401_unauthorized_not_500(
+    test_client: FlaskClient,
+    workspace_account,
+) -> None:
+    """An unknown bearer is a clean 401 ``unauthorized`` — not the latent 500 the
+    pipeline used to leak for unmapped InvalidBearerError."""
+    res = test_client.get(
+        "/openapi/v1/account",
+        headers={"Authorization": "Bearer dfoa_" + uuid.uuid4().hex},
+    )
+    assert res.status_code == 401
+    assert res.json["code"] == "unauthorized"
+
+
 def test_info_accepts_account_bearer_with_apps_read_scope(
     test_client: FlaskClient,
     app_in_workspace: App,

api/tests/unit_tests/controllers/openapi/auth/test_pipeline.py

@@ -321,3 +321,56 @@ def test_guard_no_external_identity_when_subject_email_absent(app):
             view()
 
     assert received["data"].external_identity is None
+
+
+# --- auth-failure mapping (no raw 500 leak) ---
+
+
+def test_guard_expired_token_raises_session_expired_401(app):
+    from controllers.openapi._errors import OpenApiErrorCode, SessionExpired
+    from libs.oauth_bearer import TokenExpiredError
+
+    router = _make_router()
+
+    with app.test_request_context("/test", headers={"Authorization": "Bearer tok"}):
+        with (
+            patch("controllers.openapi.auth.pipeline.extract_bearer", return_value="tok"),
+            patch("controllers.openapi.auth.pipeline.get_authenticator") as mock_auth,
+            patch("controllers.openapi.auth.pipeline.current_edition", return_value=Edition.CE),
+        ):
+            mock_auth.return_value.authenticate.side_effect = TokenExpiredError("token_expired")
+
+            @router.guard(scope=Scope.FULL)
+            def view(*, auth_data):
+                pass
+
+            with pytest.raises(SessionExpired) as exc:
+                view()
+
+    assert exc.value.code == 401
+    assert exc.value.error_code == OpenApiErrorCode.TOKEN_EXPIRED
+
+
+def test_guard_invalid_token_raises_unified_401_not_500(app):
+    from controllers.openapi._errors import InvalidBearer, OpenApiErrorCode
+    from libs.oauth_bearer import InvalidBearerError
+
+    router = _make_router()
+
+    with app.test_request_context("/test", headers={"Authorization": "Bearer tok"}):
+        with (
+            patch("controllers.openapi.auth.pipeline.extract_bearer", return_value="tok"),
+            patch("controllers.openapi.auth.pipeline.get_authenticator") as mock_auth,
+            patch("controllers.openapi.auth.pipeline.current_edition", return_value=Edition.CE),
+        ):
+            mock_auth.return_value.authenticate.side_effect = InvalidBearerError("invalid_bearer")
+
+            @router.guard(scope=Scope.FULL)
+            def view(*, auth_data):
+                pass
+
+            with pytest.raises(InvalidBearer) as exc:
+                view()
+
+    assert exc.value.code == 401
+    assert exc.value.error_code == OpenApiErrorCode.UNAUTHORIZED

api/tests/unit_tests/controllers/openapi/test_error_contract.py

@@ -33,6 +33,7 @@ from controllers.openapi._errors import (
     OpenApiErrorCode,
     OpenApiErrorFormatter,
     RecipientSurfaceMismatch,
+    SessionExpired,
 )
 from controllers.service_api.app.error import (
     AppUnavailableError,
@@ -353,3 +354,20 @@ class TestErrorCodeEnumRegistration:
         schema = model.__schema__
         assert schema["type"] == "string"
         assert set(schema["enum"]) == {member.value for member in OpenApiErrorCode}
+
+
+class TestSessionExpired:
+    def test_session_expired_emits_token_expired_401_with_hint(self):
+        fmt = OpenApiErrorFormatter()
+        e = SessionExpired()
+        data = {"code": "unauthorized", "message": e.description, "status": 401}
+
+        wire = fmt.finalize(e, data, 401)
+
+        assert wire["code"] == OpenApiErrorCode.TOKEN_EXPIRED
+        assert wire["status"] == 401
+        assert wire["hint"]
+
+    def test_session_expired_code_is_401(self):
+        assert SessionExpired.code == 401
+        assert SessionExpired.error_code == OpenApiErrorCode.TOKEN_EXPIRED

api/tests/unit_tests/libs/test_oauth_bearer_token_expiry.py

@@ -0,0 +1,87 @@
+"""Resolver-level expiry signalling.
+
+An expired token must be distinguishable from an unknown/revoked one: the
+resolver raises ``TokenExpiredError`` for expiry and returns ``None`` for
+everything else. The signal survives the negative-cache window via a distinct
+``expired`` marker so a retry inside ``NEGATIVE_TTL`` still reports expiry.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from unittest.mock import MagicMock
+
+import pytest
+
+from libs.oauth_bearer import (
+    OAuthAccessTokenResolver,
+    TokenExpiredError,
+)
+
+
+def _row(expires_at: datetime):
+    row = MagicMock()
+    row.id = "11111111-1111-1111-1111-111111111111"
+    row.account_id = "22222222-2222-2222-2222-222222222222"
+    row.prefix = "dfoa_"
+    row.subject_email = None
+    row.subject_issuer = None
+    row.client_id = None
+    row.expires_at = expires_at
+    return row
+
+
+def _resolver(redis: MagicMock, db_row=None) -> OAuthAccessTokenResolver:
+    session = MagicMock()
+    session.query.return_value.filter.return_value.one_or_none.return_value = db_row
+    session.execute.return_value.rowcount = 1
+    return OAuthAccessTokenResolver(session_factory=lambda: session, redis_client=redis)
+
+
+def test_resolve_raises_token_expired_for_expired_db_row():
+    redis = MagicMock()
+    redis.get.return_value = None  # cache miss -> DB path
+    past = datetime.now(UTC) - timedelta(minutes=1)
+    resolver = _resolver(redis, db_row=_row(past))
+
+    with pytest.raises(TokenExpiredError):
+        resolver.for_account().resolve("expiredhash")
+
+
+def test_resolve_raises_token_expired_for_expired_cache_marker():
+    redis = MagicMock()
+    redis.get.return_value = b"expired"  # negative-cache replay
+    resolver = _resolver(redis, db_row=None)
+
+    with pytest.raises(TokenExpiredError):
+        resolver.for_account().resolve("expiredhash")
+
+
+def test_resolve_returns_none_for_invalid_cache_marker():
+    redis = MagicMock()
+    redis.get.return_value = b"invalid"
+    resolver = _resolver(redis, db_row=None)
+
+    assert resolver.for_account().resolve("revokedhash") is None
+
+
+def test_resolve_returns_none_for_unknown_token():
+    redis = MagicMock()
+    redis.get.return_value = None  # cache miss
+    resolver = _resolver(redis, db_row=None)  # no DB row
+
+    assert resolver.for_account().resolve("unknownhash") is None
+
+
+def test_hard_expire_caches_expired_marker_not_invalid():
+    redis = MagicMock()
+    redis.get.return_value = None
+    past = datetime.now(UTC) - timedelta(minutes=1)
+    resolver = _resolver(redis, db_row=_row(past))
+
+    with pytest.raises(TokenExpiredError):
+        resolver.for_account().resolve("expiredhash")
+
+    setex_values = [call.args[2] for call in redis.setex.call_args_list]
+    assert "expired" in setex_values
+    assert "invalid" not in setex_values

cli/src/commands/auth/login/device-flow.ts

@@ -89,7 +89,7 @@ async function pollWithRetry(
 
 function expired(): BaseError {
   return new BaseError({
-    code: ErrorCode.ExpiredToken,
+    code: ErrorCode.TokenExpired,
     message: 'code expired before authorization',
   })
 }

cli/src/errors/codes.test.ts

@@ -35,7 +35,6 @@ describe('error codes', () => {
     [ErrorCode.AuthExpired, ExitCode.Auth],
     [ErrorCode.TokenExpired, ExitCode.Auth],
     [ErrorCode.AccessDenied, ExitCode.Auth],
-    [ErrorCode.ExpiredToken, ExitCode.Auth],
     [ErrorCode.VersionSkew, ExitCode.VersionCompat],
     [ErrorCode.UnsupportedEndpoint, ExitCode.VersionCompat],
     [ErrorCode.ConfigSchemaUnsupported, ExitCode.VersionCompat],

cli/src/errors/codes.ts

@@ -3,7 +3,6 @@ export const ErrorCode = {
   AuthExpired: 'auth_expired',
   TokenExpired: 'token_expired',
   AccessDenied: 'access_denied',
-  ExpiredToken: 'expired_token',
   VersionSkew: 'version_skew',
   UnsupportedEndpoint: 'unsupported_endpoint',
   ConfigSchemaUnsupported: 'config_schema_unsupported',
@@ -40,7 +39,6 @@ const CODE_TO_EXIT: Readonly<Record<ErrorCodeValue, ExitCodeValue>> = {
   auth_expired: ExitCode.Auth,
   token_expired: ExitCode.Auth,
   access_denied: ExitCode.Auth,
-  expired_token: ExitCode.Auth,
   version_skew: ExitCode.VersionCompat,
   unsupported_endpoint: ExitCode.VersionCompat,
   config_schema_unsupported: ExitCode.VersionCompat,

cli/src/http/error-mapper.test.ts

@@ -33,7 +33,7 @@ describe('classifyResponse — canonical ErrorBody', () => {
     expect(err.code).toBe(ErrorCode.Server4xxOther)
   })
 
-  it('401 classifies by status as AuthExpired with CLI login hint', async () => {
+  it('401 unauthorized classifies as AuthExpired with CLI login hint', async () => {
     const err = await classified(401, {
       code: 'unauthorized',
       message: 'session expired or revoked',
@@ -44,6 +44,20 @@ describe('classifyResponse — canonical ErrorBody', () => {
     expect(err.hint).toBe('run \'difyctl auth login\' to sign in again')
   })
 
+  it('401 token_expired carries the structured TokenExpired code with the server message', async () => {
+    const err = await classified(401, {
+      code: 'token_expired',
+      message: 'Your session has expired.',
+      status: 401,
+      hint: 'Re-authenticate to continue (e.g. re-run your login command).',
+    })
+
+    expect(err.code).toBe(ErrorCode.TokenExpired)
+    expect(err.exit()).toBe(4)
+    expect(err.message).toBe('Your session has expired.')
+    expect(err.hint).toBe('run \'difyctl auth login\' to sign in again')
+  })
+
   it('unknown future server code is data, not behavior — status bucket decides', async () => {
     const err = await classified(409, {
       code: 'some_future_code',

cli/src/http/error-mapper.ts

@@ -1,6 +1,6 @@
 import type { ErrorBody } from '@dify/contracts/api/openapi/types.gen'
 import type { ErrorCodeValue } from '@/errors/codes'
-import { zErrorBody } from '@dify/contracts/api/openapi/zod.gen'
+import { zErrorBody, zOpenApiErrorCode } from '@dify/contracts/api/openapi/zod.gen'
 import { BaseError, HttpClientError, newError } from '@/errors/base'
 import { ErrorCode } from '@/errors/codes'
 import { redactBearer } from './sanitize'
@@ -24,6 +24,16 @@ const AUTH_EXPIRED_CLASS: StatusClass = {
   includeRaw: false,
 }
 
+// A 401 whose body carries the server's `token_expired` code is a known,
+// behavior-driving signal (not opaque data): the session lapsed rather than the
+// token being unknown/revoked, so wrappers get the distinct structured code.
+const TOKEN_EXPIRED_CLASS: StatusClass = {
+  code: ErrorCode.TokenExpired,
+  fallbackMessage: () => 'session expired',
+  hint: AUTH_LOGIN_HINT,
+  includeRaw: false,
+}
+
 const SERVER_5XX_CLASS: StatusClass = {
   code: ErrorCode.Server5xx,
   fallbackMessage: status => `server error (HTTP ${status})`,
@@ -50,9 +60,9 @@ const ACCESS_DENIED_CLASS: StatusClass = {
   includeRaw: false,
 }
 
-function statusClass(status: number): StatusClass {
+function statusClass(status: number, serverError?: ErrorBody): StatusClass {
   if (status === 401)
-    return AUTH_EXPIRED_CLASS
+    return serverError?.code === zOpenApiErrorCode.enum.token_expired ? TOKEN_EXPIRED_CLASS : AUTH_EXPIRED_CLASS
   if (status === 403)
     return ACCESS_DENIED_CLASS
   if (status === 429)
@@ -87,7 +97,7 @@ export async function classifyResponse(request: Request, response: Response): Pr
 
   const serverError = parseServerError(raw)
   const status = response.status
-  const c = statusClass(status)
+  const c = statusClass(status, serverError)
   return new HttpClientError({
     code: c.code,
     message: serverError?.message ?? c.fallbackMessage(status),

packages/contracts/generated/api/openapi/types.gen.ts

@@ -340,6 +340,7 @@ export type OpenApiErrorCode
     | 'rate_limit_error'
     | 'recipient_surface_mismatch'
     | 'request_entity_too_large'
+    | 'token_expired'
     | 'too_many_files'
     | 'too_many_requests'
     | 'unauthorized'

packages/contracts/generated/api/openapi/zod.gen.ts

@@ -423,6 +423,7 @@ export const zOpenApiErrorCode = z.enum([
   'rate_limit_error',
   'recipient_surface_mismatch',
   'request_entity_too_large',
+  'token_expired',
   'too_many_files',
   'too_many_requests',
   'unauthorized',