fix(auth): token 登录将令牌放入 Authorization: Bearer 头
token 换登录调上游 getTokenInfo 时参数传错:原来发静态头 Authorization: Bearer admin + 令牌放 access_token cookie。上游实际 从 Authorization: Bearer <token> 头读取令牌。 - 鉴权头改为 "Bearer "+前端传入 token,删除 access_token cookie - service_authorization 配置项弃用(保留兼容,不再参与请求) - 新增回归测试 test_token_login_upstream_call.py - 同步更新 AUTH_LOGIN.md / CLAUDE.md / config.yaml 注释 / login_config 文档 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
247ff86f2f
commit
e1e9be8daa
7 files changed
+112
-20
No files matched your search
@@ -0,0 +1,89 @@
|
||||
"""Regression tests for the upstream call shape of token-exchange login.
|
||||
|
||||
The upstream ``getTokenInfo`` endpoint reads the bearer token from the
|
||||
``Authorization`` header — NOT from an ``access_token`` cookie. These tests pin
|
||||
that contract for ``_fetch_username_from_token``:
|
||||
|
||||
- ``Authorization`` is exactly ``Bearer <caller-token>`` (the token passed in),
|
||||
not the static ``service_authorization`` config value.
|
||||
- No ``access_token`` cookie is sent (the old, wrong placement).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import types
|
||||
|
||||
import pytest
|
||||
|
||||
from app.gateway.routers import auth as auth_module
|
||||
from app.gateway.routers.auth import _fetch_username_from_token
|
||||
|
||||
|
||||
class _FakeResponse:
|
||||
status_code = 200
|
||||
|
||||
def json(self) -> dict:
|
||||
return {"state": "200", "msg": "ok", "data": {"username": "b1admin"}}
|
||||
|
||||
|
||||
class _CapturingClient:
|
||||
"""Stand-in for ``httpx.AsyncClient`` that records the post() kwargs."""
|
||||
|
||||
captured: dict = {}
|
||||
|
||||
def __init__(self, *args, **kwargs) -> None:
|
||||
type(self).init_kwargs = kwargs
|
||||
|
||||
async def __aenter__(self) -> _CapturingClient:
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc) -> None:
|
||||
return None
|
||||
|
||||
async def post(self, url, **kwargs):
|
||||
type(self).captured = {"url": url, **kwargs}
|
||||
return _FakeResponse()
|
||||
|
||||
|
||||
def _patch_config(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
token_login = types.SimpleNamespace(
|
||||
token_info_url="https://upstream.example/getTokenInfo",
|
||||
service_authorization="Bearer admin", # must NOT be used anymore
|
||||
timeout_seconds=10.0,
|
||||
verify_ssl=False,
|
||||
ca_cert_path="",
|
||||
)
|
||||
fake_cfg = types.SimpleNamespace(auth_login=types.SimpleNamespace(token_login=token_login))
|
||||
monkeypatch.setattr(auth_module, "get_app_config", lambda: fake_cfg)
|
||||
monkeypatch.setattr(auth_module.httpx, "AsyncClient", _CapturingClient)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sends_caller_token_as_bearer_header(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_patch_config(monkeypatch)
|
||||
|
||||
username = await _fetch_username_from_token("UPSTREAM_TOK_123")
|
||||
|
||||
assert username == "b1admin"
|
||||
headers = _CapturingClient.captured["headers"]
|
||||
assert headers["Authorization"] == "Bearer UPSTREAM_TOK_123"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_does_not_use_static_service_authorization(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_patch_config(monkeypatch)
|
||||
|
||||
await _fetch_username_from_token("UPSTREAM_TOK_123")
|
||||
|
||||
headers = _CapturingClient.captured["headers"]
|
||||
assert headers["Authorization"] != "Bearer admin"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_access_token_cookie_is_sent(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_patch_config(monkeypatch)
|
||||
|
||||
await _fetch_username_from_token("UPSTREAM_TOK_123")
|
||||
|
||||
# The token must live in the header now; no cookie param should be passed.
|
||||
assert "cookies" not in _CapturingClient.captured
|
||||
Reference in new issue
Block a user