feat: return per-caller OAuth token status from /oauth/status

[marekdano]

Summary

Wired TokenStorageService.get_token_info (already implemented, previously unused) into mcpgateway/routers/oauth_router.py.

GET /oauth/status/{gateway_id}

For authorization_code gateways, now includes a user_token_status field derived from the authenticated caller's own identity (never a client-supplied user param):

"user_token_status": {
  "status": "valid" | "near_expiry" | "expired" | "missing",
  "authorized": true,
  "scopes": ["read", "write"],
  "expires_at": "2026-01-01T00:00:00",
  "updated_at": "2026-01-01T00:00:00"
}

• No token values are ever returned.
• Existing config fields (client_id, scopes, authorization_url, etc.) are unchanged for backward compatibility.

GET /oauth/status?gateway_ids=a&gateway_ids=b&... (new)

Batch equivalent so a grid of cards issues one request instead of N.

• Reuses the single-gateway handler internally.
• Dedupes ids, caps the batch at 100.
• Silently omits ids that 404 or aren't visible to the caller, rather than failing the whole batch.

Tests

All 216 tests in test_oauth_router.py pass. make ruff interrogate pylint is clean (100% docstring coverage, 10.00/10 pylint).

Scope

Backend-only, no UI. Diff is scoped to 2 files (~91 insertions in the router, ~100 lines of tests). Per the issue, this unblocks
#5592, #3876, #3850, and the catalog card states in #5967.

Manual Test

Prereqs: make dev running, venv activated, secrets patched (make init-secrets-patch-env).

1. Get a bearer token for a real user

The server enforces strict "user must exist in DB". admin@example.com is the bootstrapped platform admin. Don't source .env directly - it contains JSON/regex values that break bash parsing. Pull just the one variable you need:

export JWT_SECRET_KEY=$(grep -E '^JWT_SECRET_KEY=' .env | head -1 | cut -d= -f2-)

export TOKEN=$(python -m mcpgateway.utils.create_jwt_token --username admin@example.com --exp 60 --secret "$JWT_SECRET_KEY" 2>/dev/null)

2. Create a test gateway directly in the DB

POST /gateways actively probes the URL and rolls back if it's not a real MCP server, so insert the row directly instead:

cat > /tmp/create_test_gateway.py <<'PYEOF'
from mcpgateway.db import SessionLocal, Gateway
from mcpgateway.utils.create_slug import slugify
import uuid

db = SessionLocal()
name = 'oauth-status-manual-test'
gw = Gateway(
    id=uuid.uuid4().hex, name=name, slug=slugify(name),
    url='https://mcp.example.com', capabilities={}, visibility='public',
    oauth_config={
        'grant_type': 'authorization_code', 'client_id': 'cid', 'client_secret': 'csecret',
        'authorization_url': 'https://idp.example.com/authorize',
        'token_url': 'https://idp.example.com/token',
        'redirect_uri': 'http://localhost:8000/oauth/callback', 'scopes': ['read'],
    },
)
db.add(gw); db.commit()
print('GATEWAY_ID=' + gw.id)
db.close()
PYEOF

python3 /tmp/create_test_gateway.py

Copy the printed id:

export GW_ID=<the id printed above>

3. Check status before authorizing — expect missing

curl -s http://localhost:8000/oauth/status/$GW_ID -H "Authorization: Bearer $TOKEN" | python3 -m json.tool

4. Seed a token to simulate a completed OAuth flow

cat > /tmp/seed_oauth_token.py <<'PYEOF'
import asyncio
import os
from mcpgateway.db import SessionLocal
from mcpgateway.services.token_backends.db_backend import DatabaseTokenBackend
from mcpgateway.config import settings

GW_ID = os.environ["GW_ID"]
EXPIRES_IN = int(os.environ.get("EXPIRES_IN", "3600"))  # 3600=valid, 60=near_expiry, -60=expired

async def seed():
    db = SessionLocal()
    backend = DatabaseTokenBackend(db, settings)
    await backend.store_tokens(
        gateway_id=GW_ID, team_id=None, user_id="idp-user-1",
        app_user_email="admin@example.com", access_token="fake-access",
        refresh_token=None, expires_in=EXPIRES_IN, scopes=["read"],
    )
    db.close()
    print(f"seeded token for gateway={GW_ID} expires_in={EXPIRES_IN}")

asyncio.run(seed())
PYEOF

python3 /tmp/seed_oauth_token.py

Re-check status — expect valid:

curl -s http://localhost:8000/oauth/status/$GW_ID -H "Authorization: Bearer $TOKEN" | python3 -m json.tool

Try the other states by overriding EXPIRES_IN:

EXPIRES_IN=60 python3 /tmp/seed_oauth_token.py     # near_expiry (<5 min left)

EXPIRES_IN=-60 python3 /tmp/seed_oauth_token.py    # expired

5. Batch endpoint — a real id plus a nonexistent one

curl -s "http://localhost:8000/oauth/status?gateway_ids=$GW_ID&gateway_ids=nonexistent-id" -H "Authorization: Bearer $TOKEN" | python3 -m json.tool

Only $GW_ID should appear in the response — nonexistent-id is silently omitted rather than failing the whole batch.

6. Per-user isolation — a second user must see missing

cat > /tmp/create_second_user.py <<'PYEOF'
from mcpgateway.db import SessionLocal, EmailUser
db = SessionLocal()
if not db.query(EmailUser).filter_by(email='other@example.com').first():
    db.add(EmailUser(email='other@example.com', password_hash='', full_name='Other User', is_admin=False))
    db.commit()
db.close()
PYEOF

python3 /tmp/create_second_user.py

export TOKEN2=$(python -m mcpgateway.utils.create_jwt_token --username other@example.com --exp 60 --secret "$JWT_SECRET_KEY" 2>/dev/null)

curl -s http://localhost:8000/oauth/status/$GW_ID -H "Authorization: Bearer $TOKEN2" | python3 -m json.tool

other@example.com should see "status": "missing" even though admin@example.com has a valid token on the same gateway.

Cleanup

curl -s -X DELETE http://localhost:8000/gateways/$GW_ID -H "Authorization: Bearer $TOKEN"

[msureshkumar88]

Nice work wiring TokenStorageService.get_token_info into the status endpoints — the scope match to #6459 is tight (per-caller status derived server-side, existing fields preserved, batch endpoint instead of N requests per card), and the parametrized tests for valid/near_expiry/expired/missing plus the "not shared across users" test are exactly the right things to lock down. No alembic migration needed here and none was added, which is correct since this only exposes an existing field.

A few things worth addressing before merge, all confined to the new code in oauth_router.py:

1. Batch endpoint N+1 queries (get_oauth_status_batch, ~line 1480-1487)

The loop calls get_oauth_status() once per id, so a 100-id batch does ~100 separate SELECT on Gateway plus ~100 separate token lookups — roughly 200-300 round trips for what the issue asks to be "batched per page load, no N+1 queries." Worth hoisting a single select(Gateway).where(Gateway.id.in_(deduped_ids)) and reusing one TokenStorageService instance across the loop rather than constructing/query-ing per id. (Note: this should stay a sequential loop, not asyncio.gather — the default database backend's get_token_info has no actual await inside it, and the shared synchronous Session isn't safe to use concurrently across coroutines.)

2. Batch endpoint swallows 5xx the same as 404/403 (~line 1483-1486)

get_oauth_status converts any unexpected exception into HTTPException(500, ...) (line 1437-1439), and the batch handler's except HTTPException: continue treats that identically to "not found" or "not accessible" — silently dropping the id with no log line. Could you split this so 403/404 are omitted as intended, but a 5xx gets logged (and maybe surfaced) rather than disappearing? Otherwise a backend fault during a batch call is indistinguishable from "gateway doesn't exist."

3. DB lookup failure reads as "not authorized" (db_backend.py:303-305 → _get_caller_token_status, line 1358-1359)

DatabaseTokenBackend.get_token_info catches its own exceptions and returns None on failure, and _get_caller_token_status maps None to {"status": "missing", "authorized": False}. Pre-existing behavior in db_backend.py, but this PR is what first exposes it through a user-facing field that's meant to drive UI state — during a DB hiccup, a user who has authorized would see "missing" and be told to re-authorize. Worth at least a distinct status (or a log line) so a transient failure isn't indistinguishable from "never authorized."

Smaller, non-blocking:

• The if not gateway_ids: raise HTTPException(400, ...) branch (line 1473) looks unreachable via a real HTTP request — a fully-omitted gateway_ids query param triggers FastAPI's own 422 before the handler runs, and in production with should_expose_error_details() false that comes back as a generic message anyway. The unit test covers the function called directly, not the route, so this path isn't actually exercised end-to-end.
• Consider one TestClient-based test for the batch route (the file already has this pattern elsewhere) to catch exactly this kind of param-parsing mismatch.
• docs/docs/manage/oauth-troubleshooting.md:78 still describes /oauth/status/{gateway_id} as returning only "OAuth configuration status" — could use a follow-up mention of user_token_status and the new batch route.

Happy to take another pass once these land.

[vishu-bh]

Thanks for addressing the comments, PR looks good to merge once conflict is resolved

[vishu-bh]

LGTM 🚀

[msureshkumar88]

Nice work wiring TokenStorageService.get_token_info into the status endpoints — the per-caller isolation and the shared _build_oauth_status_payload helper keeping the two routes from drifting are both solid, and skipping an alembic migration is the right call here since OAuthToken already carries the per-user state.

Two things I think are worth another pass before merge:

1. The batch endpoint's N+1 fix doesn't hold for non-admin/narrowed-admin callers

get_oauth_status_batch does one bulk SELECT ... WHERE Gateway.id.in_(...), but the per-id _enforce_gateway_access call still goes through token_scoping_middleware._check_resource_team_ownership, which re-fetches each Gateway row from the DB by id for any caller that isn't a fully un-narrowed admin. I verified this with a quick test using realistic uuid4().hex gateway ids (the actual Gateway.id default) instead of "gateway123":

ids	db.execute calls
1 real hex id	2
2 real hex ids	3

So it's still effectively N+1 for the common case. The existing test_get_oauth_status_batch_single_gateway_query only passes because "gateway123" doesn't match _RESOURCE_PATTERNS' [a-f0-9\-]+ regex, so the ownership recheck never fires — a real hex id would fail that same assertion.

Issue #6459 calls out "Batched per page load, no N+1 queries" as an explicit requirement, so I think this needs a real fix rather than a follow-up: since the batch loop already has the loaded Gateway object, could _enforce_gateway_access (or the ownership check it delegates to) accept a preloaded gateway and skip the re-SELECT when one's supplied? Worth also swapping the test fixture id for a real hex string so this regresses loudly next time.

2. VaultTokenBackend.get_token_info doesn't get the same lookup-failure fix as the DB backend

This PR changes DatabaseTokenBackend.get_token_info to re-raise on failure instead of swallowing to None, specifically so a transient DB error is distinguishable from "never authorized" once it surfaces through _get_caller_token_status. VaultTokenBackend.get_token_info still catches VaultConnectionError/VaultAuthError and returns None, so on OAUTH_TOKEN_BACKEND=vault deployments a transient Vault outage would still report every caller as status: missing, authorized: false — the exact ambiguity this PR is fixing for the DB backend. Issue #6459 mentions get_token_info being implemented in both backends, so I think this should land in the same PR rather than as a gap.

Smaller/non-blocking notes, happy to see these follow up separately if preferred:

• Worth a batch test that exercises _enforce_gateway_access's team/private-visibility branch (all current fixtures use visibility="public"), and one that checks batch/single payload parity for the same gateway_id + caller.
• No batch coverage for a non-authorization_code grant type (confirming user_token_status stays omitted).
• Per AGENTS.md, PRs exercisable through a live gateway should include a black-box test under tests/live_gateway/ — currently there's none for either endpoint.
• The PR body doesn't reference #6459 anywhere, so merging won't auto-close the tracking issue (branch name does, but GitHub doesn't use that for auto-close).

[vishu-bh]

LGTM 🚀

[msureshkumar88]

Nice work here — this cleanly wires up logic that was already built and tested (get_token_info) rather than reinventing it, keeps the change backend-only as scoped, and the batch endpoint's fail-closed/omit-on-error design for inaccessible gateways is the right call. The preloaded_gateway fix to _check_resource_team_ownership is also a nice catch — it quietly fixes a redundant SELECT on all four existing OAuth endpoints, not just the new batch one. Test coverage on the router/middleware side is thorough (per-user isolation, N+1 regression with real hex ids, private-visibility owner/non-owner cases, batch edge cases).

A few things worth addressing before merge, plus some lower-priority suggestions.

Suggested before merge

No live/black-box test for either status endpoint. All coverage for user_token_status and the new batch endpoint is unit-level against mocked Session/TokenStorageService objects — nothing in tests/live_gateway, tests/integration, or the load tests exercises them against a running gateway. The PR description's manual-test section (create gateway, seed a token via DatabaseTokenBackend, curl the endpoint, verify per-user isolation) is already exactly this scenario — would it make sense to turn that into an automated tests/live_gateway test rather than leaving it as copy-paste curl instructions?

Worth considering (non-blocking)

Batch endpoint's per-gateway loop is fully sequential, and the two backends have different costs here:

• DatabaseTokenBackend.get_token_info issues a synchronous SELECT per gateway id — asyncio.gather wouldn't help (sync SQLAlchemy doesn't yield), but a single batched query (WHERE gateway_id IN (:ids) AND app_user_email = :email) would collapse N queries to 1.
• VaultTokenBackend.get_token_info opens a new httpx.AsyncClient per call (no connection reuse) and doesn't consult the existing _token_cache, so at the 100-id batch cap this is up to 100 uncached, sequential Vault round trips per catalog page load. Worth a look for anyone deploying with the Vault backend.

Two except Exception handlers in the batch loop log only str(exc), dropping the traceback (oauth_router.py, the access-check and payload-build handlers). Since these are the paths that would catch something unexpected rather than the anticipated Vault/DB errors, logger.exception(...) there would make a future incident easier to root-cause, with no change to the response.

Minor

Neither status endpoint declares a response_model (other routers in this codebase, e.g. tokens.py, do), so the user_token_status shape isn't visible in the generated OpenAPI schema. Given a few UI issues are blocked on this as a contract, might be worth a typed model in a follow-up — not something this PR needs to carry.

The get_token_info swallow-to-None → log-and-reraise change in base.py/db_backend.py/vault_backend.py is a broader behavior change than the issue asked for, though I traced it and it's safe: the only caller (_get_caller_token_status) catches it and maps back to the same "missing" response either way, and it doesn't break any of the existing tests for those backends. Just flagging it since it changes a shared method's contract for future callers, not just this one.

[msureshkumar88]

Thanks for wiring this up — get_token_info sitting unused was exactly the kind of gap that's easy to miss, and the batch endpoint design (dedupe, cap, silent omission of inaccessible ids) matches what a catalog grid actually needs. The preloaded_gateway reuse in _check_resource_team_ownership and the accompanying mismatch-fallback test are a nice touch. Test coverage on the router side is thorough — the 4-way status parametrize, cross-user isolation test, and the db.execute.call_count == 1 regression guard all land well. Good to see a live black-box test included too.

A few things worth resolving before merge:

Blocking

1. Batch endpoint is still N+1 on the token lookup. The Gateway fetch is batched (gateways_by_id = {...} via one IN (...) query), but _get_caller_token_status is still called once per gateway id inside the loop, each issuing its own SELECT against oauth_tokens (or a Vault round trip). The issue text calls this out explicitly: "Batched per page load, no N+1 queries." Would a bulk lookup work here — e.g. a get_token_info_bulk(gateway_ids, app_user_email) on AbstractTokenBackend with a default loop implementation, overridden in DatabaseTokenBackend with a single WHERE gateway_id IN (...) AND app_user_email = :email? That would close the gap for the default backend without touching Vault's per-item nature.

2. The None → raise change in the backends doesn't reach the caller. db_backend.get_token_info and vault_backend.get_token_info now raise instead of returning None on failure, with docstrings explaining this lets callers "distinguish never-authorized from lookup-failed." But the only caller (_get_caller_token_status in oauth_router.py) catches the exception and returns {"status": "missing", "authorized": False} — collapsing straight back to the same value the change was meant to avoid. Two ways to close this: either surface a distinct "status": "unknown" in the router (so a Vault outage doesn't look like "never authorized" to the UI — which matters, since a UI acting on that would prompt a fresh OAuth flow with an IdP), or drop the backend/base contract change and keep just the router-level log line. Either is fine, just want the two ends to agree.

3. No bound on a slow/unresponsive backend in the batch loop. With OAUTH_TOKEN_BACKEND=vault, each of up to 100 ids in a batch call goes through _vault_request's 3-attempt retry with exponential backoff (up to ~33s per id on a hard failure), sequentially, inside a single request. Worth adding a timeout around the batch as a whole, or a circuit-breaker so one backend failure doesn't retry 100 times?

Suggestions

4. Consider adding @require_permission("gateways.read") to both status endpoints (matching the existing /vault/authorize/{id} pattern) and a _PERMISSION_PATTERNS entry for /oauth/status — right now a scoped API token (as opposed to a session token) would get a 403 from TokenScopingMiddleware's default-deny, since the path isn't in the permission map. If the React catalog ends up using scoped tokens, the batch endpoint would be unreachable for it.
5. The if not gateway_ids: raise HTTPException(400) branch looks unreachable over HTTP — gateway_ids has no default=, so FastAPI returns 422 before the handler runs when the param is omitted. Worth giving it default=[] (making the 400 path real) or dropping the check and its test.
6. _get_caller_token_status's except-block uses logger.error(..., str(e)) (no traceback), while the batch loop uses logger.exception for the same kind of failure — worth aligning on logger.exception there too.
7. Live test suite could use a deny-path case (a private gateway owned by someone else → 403 / omitted from batch) alongside the existing per-user token isolation test, plus teardown for the second_user fixture (currently left in the DB after the run).

Minor

8. oauth-troubleshooting.md documents the new routes but not the user_token_status shape, the four status values, the 300s near-expiry threshold, or the 100-id cap / silent-omission behavior for inaccessible ids — might be worth a short addition for API consumers.
9. Could you add Closes #6459 to the PR description? The body currently references the issues this unblocks but not the one it resolves.

Happy to take another pass once these land — the shape of the change is right, just want the batch endpoint to actually deliver on the no-N+1 requirement from the issue.

[marekdano]

@msureshkumar88

Blocking findings fixed

1. Batch N+1 on token lookup — added get_token_info_bulk() to AbstractTokenBackend (default per-id loop, isolates failures) and TokenStorageService; DatabaseTokenBackend overrides it with a single WHERE gateway_id IN (...) query. The batch endpoint now does one bulk call instead of N.
2. None→raise not reaching the caller — added a "unknown" status (distinct from "missing") surfaced when a lookup fails, via a new _token_info_to_status_payload() helper shared by both endpoints.
3. No timeout bound — wrapped the batch's bulk lookup in asyncio.wait_for() (15s); on timeout, pending ids report "unknown" instead of hanging the request.

Suggestions addressed

4. Added a _PERMISSION_PATTERNS entry (gateways.read) for /oauth/status* in the middleware, matching the /vault/authorize/{id} pattern — closes the scoped-API-token 403 gap. (Skipped also adding a router-level @require_permission decorator — the in-handler _enforce_gateway_access already does per-gateway RBAC-equivalent authorization, and adding the decorator would have required extensive PermissionService/DB mocking rework across ~24 existing tests for a Layer-1 gap the middleware entry alone closes. Happy to add it too if wanted.)
5. gateway_ids now has a real default (None→[]), making the 400 branch reachable over HTTP.
6. _get_caller_token_status's except-block now uses logger.exception.
7. Added a live-test deny-path case (private gateway, non-owner → 403 / omitted from batch) plus teardown for the second_user fixture.

Minor

8. Documented user_token_status shape, all 5 status values, the 300s threshold, and the 100-id cap/silent-omission in oauth-troubleshooting.md.
9. Closes #6459 is ready to add to the PR description (issue confirmed open and matching).

Tests

Added/updated unit tests across test_oauth_router.py, test_token_scoping.py, test_token_storage_service.py, test_db_backend.py, test_base.py — all passing (281 tests). make ruff, make interrogate (100%), and make pylint are clean on the touched files (the 2 remaining pylint warnings pre-exist on main, unrelated to this diff). make detect-secrets-scan ran clean (0 unaudited/live findings) and refreshed .secrets.baseline.

[msureshkumar88]

Thanks for the thorough follow-up on the batching, backend-error handling, documentation, and live coverage. One blocking authorization gap remains.

Blocking — preserve Layer-2 RBAC on the OAuth status routes

GET /oauth/status/{gateway_id} and GET /oauth/status authenticate the caller and run gateway visibility/ownership checks, but neither is decorated with @require_permission(Permissions.GATEWAYS_READ). The new _PERMISSION_PATTERNS entry is necessary for scoped API tokens, but it is only Layer 1: an empty token scope intentionally passes through to route-level RBAC, and these handlers do not perform that check.

This lets an authenticated principal without gateways.read receive OAuth configuration and its caller-scoped token state for a visible gateway, contrary to the project’s two-layer authorization invariant (CWE-863). The current live isolation test is a useful regression setup: on a clean stack its second user has teams=[] and no role, yet receives 200 for the public gateway.

Could we add @require_permission(Permissions.GATEWAYS_READ) to both status routes, retain the token-scope mapping, and add deny-path coverage for a caller with no gateways.read permission? The batch route’s expected response for a caller lacking that permission should be explicitly asserted as well.

[msureshkumar88]

Thanks for adding the Layer-2 gateways.read checks to both OAuth status routes — that closes the authorization gap identified in the previous review, and the new unit deny-path assertions cover the decorator itself.

One blocking inconsistency remains in the live black-box coverage:

Blocking — update the live-test user for the new RBAC requirement

second_user is inserted directly as an EmailUser without any UserRole, and its JWT uses teams=[]. Under the real permission path this user has no gateways.read permission, so the new decorator returns 403 before the handlers reach the per-user isolation or private-gateway filtering logic.

That conflicts with these current expectations:

• test_per_user_isolation expects 200 and status: missing.
• test_private_gateway_denies_non_owner expects 403, but now exercises missing RBAC permission rather than private ownership.
• test_batch_endpoint_omits_private_gateway_for_non_owner expects 200 with omission, but receives 403 at the route-level gate.

Could the fixture grant this user an appropriate role containing gateways.read for the isolation and visibility scenarios, while retaining separate no-permission coverage that explicitly expects 403 for both routes? The PR body's manual per-user-isolation steps should be updated similarly, since they also create a role-less user and currently expect status: missing.

I verified the behavior with the real PermissionService against an isolated database: a directly inserted user has zero roles and check_permission(..., Permissions.GATEWAYS_READ, token_teams=[], check_any_team=True) returns false.

[marekdano]

Verification: I started colima, brought up the compose Postgres service, and ran the gateway locally against it. All 10 live tests pass; the full 216-test test_oauth_router.py unit suite and make ruff interrogate pylint are still clean.