SmartUp DeveloperandClaude Opus 4.8
2262b6a564
- Parse JWT 'exp' claim to track token expiry (token_expires_at) - Proactively refresh Bearer tokens 1 hour before expiry in ensure_authenticated() - Refresh response without expires_in now falls back to JWT exp for expiry derivation - Support custom refresh_path via bearer auth_config whitelist - Unify 401 fallback: all bearer types (including Sub2API) now go through _refresh_bearer_token() instead of separate old path - Remove dead _is_sub2api_bearer() / _refresh_sub2api_bearer_token() - Add 2 new tests: refresh_path override + JWT exp backfill Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
415 lines
14 KiB
Python
415 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import time
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.services import upstream_client
|
|
from app.services.upstream_client import UpstreamClient
|
|
|
|
|
|
class FakeHttpClient:
|
|
def __init__(self, handler):
|
|
self.handler = handler
|
|
self.calls = []
|
|
|
|
def request(self, method: str, url: str, **kwargs):
|
|
request = httpx.Request(method, url)
|
|
self.calls.append({
|
|
"method": method,
|
|
"path": request.url.path,
|
|
"headers": kwargs.get("headers") or {},
|
|
"json": kwargs.get("json"),
|
|
})
|
|
return self.handler(request, kwargs)
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
|
|
def _install_fake_client(monkeypatch, handler) -> FakeHttpClient:
|
|
fake = FakeHttpClient(handler)
|
|
monkeypatch.setattr(upstream_client.httpx, "Client", lambda timeout=None: fake)
|
|
return fake
|
|
|
|
|
|
def _response(request: httpx.Request, status_code: int, payload=None) -> httpx.Response:
|
|
return httpx.Response(status_code, json=payload, request=request)
|
|
|
|
|
|
def _jwt_token(exp_timestamp: int) -> str:
|
|
"""Build a fake JWT with the given exp claim."""
|
|
payload_b64 = base64.urlsafe_b64encode(
|
|
json.dumps({"exp": exp_timestamp}).encode()
|
|
).rstrip(b"=").decode()
|
|
return f"header.{payload_b64}.sig"
|
|
|
|
|
|
# ── Existing Sub2API bearer tests (unchanged) ─────────────────────────────
|
|
|
|
|
|
def test_sub2api_bearer_refreshes_on_401_and_retries(monkeypatch):
|
|
updates: list[dict] = []
|
|
|
|
def handler(request: httpx.Request, kwargs):
|
|
if request.url.path == "/api/v1/groups/available":
|
|
auth = kwargs["headers"].get("Authorization")
|
|
if auth == "Bearer old-access":
|
|
return _response(request, 401, {"detail": "expired"})
|
|
assert auth == "Bearer new-access"
|
|
return _response(request, 200, [{"id": "g1", "name": "G1"}])
|
|
if request.url.path == "/api/v1/auth/refresh":
|
|
assert kwargs["json"] == {"refresh_token": "old-refresh"}
|
|
return _response(request, 200, {
|
|
"access_token": "new-access",
|
|
"refresh_token": "new-refresh",
|
|
"expires_in": 3600,
|
|
})
|
|
raise AssertionError(f"unexpected request: {request.method} {request.url}")
|
|
|
|
fake = _install_fake_client(monkeypatch, handler)
|
|
client = UpstreamClient(
|
|
base_url="http://sub2api.local",
|
|
api_prefix="/api/v1",
|
|
auth_type="bearer",
|
|
auth_config={"token": "old-access", "refresh_token": "old-refresh"},
|
|
on_auth_config_update=updates.append,
|
|
)
|
|
|
|
groups = client.get_available_groups("/groups/available")
|
|
|
|
assert groups == [{"id": "g1", "name": "G1"}]
|
|
assert [(c["method"], c["path"]) for c in fake.calls] == [
|
|
("GET", "/api/v1/groups/available"),
|
|
("POST", "/api/v1/auth/refresh"),
|
|
("GET", "/api/v1/groups/available"),
|
|
]
|
|
assert updates[-1]["token"] == "new-access"
|
|
assert updates[-1]["refresh_token"] == "new-refresh"
|
|
assert updates[-1]["expires_in"] == 3600
|
|
|
|
|
|
def test_sub2api_bearer_without_refresh_token_does_not_refresh(monkeypatch):
|
|
def handler(request: httpx.Request, kwargs):
|
|
assert request.url.path == "/api/v1/groups/available"
|
|
return _response(request, 401, {"detail": "expired"})
|
|
|
|
fake = _install_fake_client(monkeypatch, handler)
|
|
client = UpstreamClient(
|
|
base_url="http://sub2api.local",
|
|
api_prefix="/api/v1",
|
|
auth_type="bearer",
|
|
auth_config={"token": "old-access"},
|
|
)
|
|
|
|
with pytest.raises(httpx.HTTPStatusError):
|
|
client.get_available_groups("/groups/available")
|
|
|
|
assert [(c["method"], c["path"]) for c in fake.calls] == [
|
|
("GET", "/api/v1/groups/available"),
|
|
]
|
|
|
|
|
|
# ── Updated: non-Sub2API bearer now refreshes with refresh_token ──────────
|
|
|
|
|
|
def test_bearer_refreshes_on_401_with_default_refresh_path(monkeypatch):
|
|
"""Non-sub2api bearer with refresh_token refreshes on 401 (new behaviour)."""
|
|
updates: list[dict] = []
|
|
|
|
def handler(request: httpx.Request, kwargs):
|
|
if request.url.path == "/groups/available":
|
|
auth = kwargs["headers"].get("Authorization")
|
|
if auth == "Bearer old-access":
|
|
return _response(request, 401, {"detail": "expired"})
|
|
assert auth == "Bearer new-access"
|
|
return _response(request, 200, [{"id": "g1", "name": "G1"}])
|
|
if request.url.path == "/api/user/refresh":
|
|
assert kwargs["json"] == {"refresh_token": "old-refresh"}
|
|
return _response(request, 200, {
|
|
"access_token": "new-access",
|
|
"refresh_token": "new-refresh",
|
|
"expires_in": 3600,
|
|
})
|
|
raise AssertionError(f"unexpected: {request.method} {request.url}")
|
|
|
|
fake = _install_fake_client(monkeypatch, handler)
|
|
client = UpstreamClient(
|
|
base_url="http://plain-bearer.local",
|
|
api_prefix="",
|
|
auth_type="bearer",
|
|
auth_config={"token": "old-access", "refresh_token": "old-refresh"},
|
|
on_auth_config_update=updates.append,
|
|
)
|
|
|
|
groups = client.get_available_groups("/groups/available")
|
|
|
|
assert groups == [{"id": "g1", "name": "G1"}]
|
|
assert [(c["method"], c["path"]) for c in fake.calls] == [
|
|
("GET", "/groups/available"),
|
|
("POST", "/api/user/refresh"),
|
|
("GET", "/groups/available"),
|
|
]
|
|
assert updates[-1]["token"] == "new-access"
|
|
|
|
|
|
def test_bearer_without_refresh_token_still_fails_on_401(monkeypatch):
|
|
"""Bearer without refresh_token does NOT retry on 401."""
|
|
def handler(request: httpx.Request, kwargs):
|
|
assert request.url.path == "/groups/available"
|
|
return _response(request, 401, {"detail": "expired"})
|
|
|
|
fake = _install_fake_client(monkeypatch, handler)
|
|
client = UpstreamClient(
|
|
base_url="http://plain-bearer.local",
|
|
api_prefix="",
|
|
auth_type="bearer",
|
|
auth_config={"token": "old-access"},
|
|
)
|
|
|
|
with pytest.raises(httpx.HTTPStatusError):
|
|
client.get_available_groups("/groups/available")
|
|
|
|
assert [(c["method"], c["path"]) for c in fake.calls] == [
|
|
("GET", "/groups/available"),
|
|
]
|
|
|
|
|
|
# ── New: proactive refresh tests ─────────────────────────────────────────
|
|
|
|
|
|
def test_bearer_proactive_refresh_when_expiry_approaching(monkeypatch):
|
|
"""ensure_authenticated proactively refreshes token expiring within 1 h."""
|
|
updates: list[dict] = []
|
|
|
|
def handler(request: httpx.Request, kwargs):
|
|
if request.url.path == "/api/user/refresh":
|
|
assert kwargs["json"] == {"refresh_token": "the-refresh"}
|
|
return _response(request, 200, {
|
|
"access_token": "new-token",
|
|
"refresh_token": "new-refresh",
|
|
"expires_in": 7200,
|
|
})
|
|
raise AssertionError(f"unexpected: {request.method} {request.url}")
|
|
|
|
_install_fake_client(monkeypatch, handler)
|
|
client = UpstreamClient(
|
|
base_url="http://bearer.local",
|
|
api_prefix="",
|
|
auth_type="bearer",
|
|
auth_config={
|
|
"token": "old-token",
|
|
"refresh_token": "the-refresh",
|
|
"token_expires_at": int(time.time()) + 1800, # 30 min left
|
|
},
|
|
on_auth_config_update=updates.append,
|
|
)
|
|
|
|
client.ensure_authenticated()
|
|
|
|
assert updates[-1]["token"] == "new-token"
|
|
assert updates[-1]["refresh_token"] == "new-refresh"
|
|
|
|
|
|
def test_bearer_proactive_skips_when_token_fresh(monkeypatch):
|
|
"""ensure_authenticated does NOT refresh token that is far from expiry."""
|
|
|
|
def handler(request: httpx.Request, kwargs):
|
|
raise AssertionError(f"unexpected request: {request.method} {request.url}")
|
|
|
|
_install_fake_client(monkeypatch, handler)
|
|
client = UpstreamClient(
|
|
base_url="http://bearer.local",
|
|
api_prefix="",
|
|
auth_type="bearer",
|
|
auth_config={
|
|
"token": "still-fresh",
|
|
"refresh_token": "the-refresh",
|
|
"token_expires_at": int(time.time()) + 86400 * 7, # 7 days
|
|
},
|
|
)
|
|
|
|
# Should not make any HTTP request
|
|
client.ensure_authenticated()
|
|
|
|
|
|
def test_bearer_proactive_parses_jwt_exp_when_missing(monkeypatch):
|
|
"""Missing token_expires_at is derived from JWT exp claim."""
|
|
future_exp = int(time.time()) + 86400 * 7 # 7 days → far away, no refresh
|
|
fake_jwt = _jwt_token(future_exp)
|
|
|
|
def handler(request: httpx.Request, kwargs):
|
|
raise AssertionError(f"unexpected request: {request.method} {request.url}")
|
|
|
|
_install_fake_client(monkeypatch, handler)
|
|
client = UpstreamClient(
|
|
base_url="http://bearer.local",
|
|
api_prefix="",
|
|
auth_type="bearer",
|
|
auth_config={"token": fake_jwt, "refresh_token": "the-refresh"},
|
|
)
|
|
|
|
client.ensure_authenticated()
|
|
|
|
assert client.auth_config.get("token_expires_at") == future_exp
|
|
|
|
|
|
def test_bearer_proactive_refresh_from_jwt_exp(monkeypatch):
|
|
"""When token_expires_at is missing but JWT exp is within 1h, refresh."""
|
|
updates: list[dict] = []
|
|
|
|
def handler(request: httpx.Request, kwargs):
|
|
if request.url.path == "/api/user/refresh":
|
|
return _response(request, 200, {
|
|
"access_token": "refreshed-token",
|
|
"expires_in": 7200,
|
|
})
|
|
raise AssertionError(f"unexpected: {request.method} {request.url}")
|
|
|
|
_install_fake_client(monkeypatch, handler)
|
|
near_exp = int(time.time()) + 600 # 10 min left
|
|
client = UpstreamClient(
|
|
base_url="http://bearer.local",
|
|
api_prefix="",
|
|
auth_type="bearer",
|
|
auth_config={"token": _jwt_token(near_exp), "refresh_token": "the-refresh"},
|
|
on_auth_config_update=updates.append,
|
|
)
|
|
|
|
client.ensure_authenticated()
|
|
|
|
assert updates[-1]["token"] == "refreshed-token"
|
|
|
|
|
|
def test_bearer_refresh_uses_custom_refresh_path(monkeypatch):
|
|
"""refresh_path in auth_config overrides the default refresh endpoint."""
|
|
updates: list[dict] = []
|
|
|
|
def handler(request: httpx.Request, kwargs):
|
|
if request.url.path == "/custom/refresh/endpoint":
|
|
assert kwargs["json"] == {"refresh_token": "my-refresh"}
|
|
return _response(request, 200, {
|
|
"access_token": "custom-refreshed",
|
|
"expires_in": 7200,
|
|
})
|
|
raise AssertionError(f"unexpected: {request.method} {request.url}")
|
|
|
|
_install_fake_client(monkeypatch, handler)
|
|
# Token is within 1h → triggers proactive refresh
|
|
near_exp = int(time.time()) + 1800
|
|
client = UpstreamClient(
|
|
base_url="http://bearer.local",
|
|
api_prefix="", # default would be /api/user/refresh
|
|
auth_type="bearer",
|
|
auth_config={
|
|
"token": "stale-token",
|
|
"refresh_token": "my-refresh",
|
|
"refresh_path": "/custom/refresh/endpoint",
|
|
"token_expires_at": near_exp,
|
|
},
|
|
on_auth_config_update=updates.append,
|
|
)
|
|
|
|
client.ensure_authenticated()
|
|
|
|
assert updates[-1]["token"] == "custom-refreshed"
|
|
|
|
|
|
def test_bearer_refresh_backfills_token_expires_at_from_jwt(monkeypatch):
|
|
"""When refresh response has no expires_in, derive token_expires_at from JWT exp."""
|
|
updates: list[dict] = []
|
|
future_exp = int(time.time()) + 3600 # 1 h from now
|
|
new_access_token = _jwt_token(future_exp)
|
|
|
|
def handler(request: httpx.Request, kwargs):
|
|
if request.url.path == "/api/user/refresh":
|
|
assert kwargs["json"] == {"refresh_token": "my-refresh"}
|
|
# Response deliberately omits expires_in
|
|
return _response(request, 200, {
|
|
"access_token": new_access_token,
|
|
"refresh_token": "new-refresh",
|
|
})
|
|
raise AssertionError(f"unexpected: {request.method} {request.url}")
|
|
|
|
_install_fake_client(monkeypatch, handler)
|
|
# Token is within 1h → triggers proactive refresh
|
|
near_exp = int(time.time()) + 1800
|
|
client = UpstreamClient(
|
|
base_url="http://bearer.local",
|
|
api_prefix="",
|
|
auth_type="bearer",
|
|
auth_config={
|
|
"token": "stale-token",
|
|
"refresh_token": "my-refresh",
|
|
"token_expires_at": near_exp,
|
|
},
|
|
on_auth_config_update=updates.append,
|
|
)
|
|
|
|
client.ensure_authenticated()
|
|
|
|
assert updates[-1]["token"] == new_access_token
|
|
assert updates[-1]["token_expires_at"] == future_exp
|
|
# expires_in was NOT in the response → should not appear in the update
|
|
assert "expires_in" not in updates[-1]
|
|
|
|
|
|
# ── Existing login_password tests (unchanged) ─────────────────────────────
|
|
|
|
|
|
def test_login_password_saves_sub2api_refresh_token(monkeypatch):
|
|
updates: list[dict] = []
|
|
|
|
def handler(request: httpx.Request, kwargs):
|
|
assert request.url.path == "/api/v1/auth/login"
|
|
assert kwargs["json"] == {"email": "user@example.com", "password": "secret"}
|
|
return _response(request, 200, {
|
|
"access_token": "login-access",
|
|
"refresh_token": "login-refresh",
|
|
"expires_in": 1800,
|
|
})
|
|
|
|
_install_fake_client(monkeypatch, handler)
|
|
client = UpstreamClient(
|
|
base_url="http://sub2api.local",
|
|
api_prefix="/api/v1",
|
|
auth_type="login_password",
|
|
auth_config={"email": "user@example.com", "password": "secret"},
|
|
on_auth_config_update=updates.append,
|
|
)
|
|
|
|
client.login()
|
|
|
|
assert client._headers()["Authorization"] == "Bearer login-access"
|
|
assert updates[-1]["token"] == "login-access"
|
|
assert updates[-1]["refresh_token"] == "login-refresh"
|
|
assert updates[-1]["expires_in"] == 1800
|
|
|
|
|
|
def test_login_password_empty_prefix_defaults_to_user_api_login(monkeypatch):
|
|
def handler(request: httpx.Request, kwargs):
|
|
assert request.url.path == "/api/user/login"
|
|
assert kwargs["json"] == {"username": "admin", "password": "secret"}
|
|
return _response(request, 200, {
|
|
"token": "nox-access",
|
|
"user": {"id": 9},
|
|
})
|
|
|
|
_install_fake_client(monkeypatch, handler)
|
|
client = UpstreamClient(
|
|
base_url="http://nox.local",
|
|
api_prefix="",
|
|
auth_type="login_password",
|
|
auth_config={"email": "admin", "password": "secret"},
|
|
)
|
|
|
|
client.login()
|
|
|
|
headers = client._headers()
|
|
assert headers["Authorization"] == "Bearer nox-access"
|
|
assert headers["Nox-Api-User"] == "9"
|
|
assert headers["New-Api-User"] == "9"
|