8120d2ae6a
移除本地 FAM_AUTH_USER/PASS 账号密码校验和内置登录表单，/login 改为 302 跳转 auth-hub /authorize，新增 /api/auth/callback 完成 code 换 token + id_token 签名验证（PyJWT + JWKS），验证通过后种回原有 fam_session cookie， is_authed()/全局登录拦截逻辑不变。接入参数走环境变量，未配置齐全 fail closed。本地起 auth-hub 开发实例 + 真实浏览器验证过完整登录/登出闭环。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
300 lines
9.7 KiB
Python
300 lines
9.7 KiB
Python
import time
|
||
from types import SimpleNamespace
|
||
|
||
import jwt
|
||
import pytest
|
||
|
||
from fam_core import auth
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _clean_env_and_state(monkeypatch):
|
||
"""每个用例前清掉环境变量和进程内状态，用例之间不互相污染。"""
|
||
for var in ('AUTH_HUB_ISSUER', 'AUTH_HUB_CLIENT_ID',
|
||
'AUTH_HUB_CLIENT_SECRET', 'AUTH_HUB_REDIRECT_URI'):
|
||
monkeypatch.delenv(var, raising=False)
|
||
auth._sessions.clear()
|
||
auth._pending.clear()
|
||
auth._warned_unconfigured = False
|
||
auth._jwks_client = None
|
||
yield
|
||
auth._sessions.clear()
|
||
auth._pending.clear()
|
||
|
||
|
||
def _set_auth_hub_env(monkeypatch):
|
||
monkeypatch.setenv('AUTH_HUB_ISSUER', 'http://auth-hub.example')
|
||
monkeypatch.setenv('AUTH_HUB_CLIENT_ID', 'fam-core')
|
||
monkeypatch.setenv('AUTH_HUB_CLIENT_SECRET', 'sekret')
|
||
monkeypatch.setenv('AUTH_HUB_REDIRECT_URI', 'http://fam.example/api/auth/callback')
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 配置齐全性校验（fail closed）
|
||
# ---------------------------------------------------------------------------
|
||
def test_require_auth_hub_config_fails_closed_when_unconfigured():
|
||
assert auth._require_auth_hub_config() is None
|
||
|
||
|
||
def test_require_auth_hub_config_fails_closed_when_partially_configured(monkeypatch):
|
||
monkeypatch.setenv('AUTH_HUB_ISSUER', 'http://auth-hub.example')
|
||
monkeypatch.setenv('AUTH_HUB_CLIENT_ID', 'fam-core')
|
||
assert auth._require_auth_hub_config() is None
|
||
|
||
|
||
def test_require_auth_hub_config_succeeds_when_fully_configured(monkeypatch):
|
||
_set_auth_hub_env(monkeypatch)
|
||
cfg = auth._require_auth_hub_config()
|
||
assert cfg == {
|
||
'issuer': 'http://auth-hub.example',
|
||
'client_id': 'fam-core',
|
||
'client_secret': 'sekret',
|
||
'redirect_uri': 'http://fam.example/api/auth/callback',
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 本地会话（fam_session cookie）
|
||
# ---------------------------------------------------------------------------
|
||
def test_session_ttl_is_two_hours():
|
||
assert auth._SESSION_TTL == 2 * 3600
|
||
|
||
|
||
def test_is_authed_true_within_ttl():
|
||
tok = 'sometoken'
|
||
auth._sessions[tok] = time.time() + 3600
|
||
|
||
class _FakeRequest:
|
||
cookies = {'fam_session': tok}
|
||
monkeypatch_request = auth.request
|
||
try:
|
||
auth.request = _FakeRequest()
|
||
assert auth.is_authed() is True
|
||
finally:
|
||
auth.request = monkeypatch_request
|
||
|
||
|
||
def test_is_authed_false_after_expiry():
|
||
tok = 'expiredtoken'
|
||
auth._sessions[tok] = time.time() - 1 # 已过期
|
||
|
||
class _FakeRequest:
|
||
cookies = {'fam_session': tok}
|
||
monkeypatch_request = auth.request
|
||
try:
|
||
auth.request = _FakeRequest()
|
||
assert auth.is_authed() is False
|
||
# 过期后应该从 session 表里清掉，不留垃圾
|
||
assert tok not in auth._sessions
|
||
finally:
|
||
auth.request = monkeypatch_request
|
||
|
||
|
||
def test_is_authed_false_without_cookie():
|
||
class _FakeRequest:
|
||
cookies = {}
|
||
monkeypatch_request = auth.request
|
||
try:
|
||
auth.request = _FakeRequest()
|
||
assert auth.is_authed() is False
|
||
finally:
|
||
auth.request = monkeypatch_request
|
||
|
||
|
||
def test_create_session_registers_token_with_ttl():
|
||
before = time.time()
|
||
tok = auth._create_session()
|
||
assert tok in auth._sessions
|
||
assert auth._sessions[tok] >= before + auth._SESSION_TTL
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 白名单
|
||
# ---------------------------------------------------------------------------
|
||
def test_whitelist_exact_paths():
|
||
for path in ('/login', '/api/auth/callback', '/api/logout', '/api/auth/check',
|
||
'/health', '/favicon.ico'):
|
||
assert auth._is_whitelisted(path) is True
|
||
|
||
|
||
def test_whitelist_assets_prefix():
|
||
assert auth._is_whitelisted('/assets/index-abc123.js') is True
|
||
|
||
|
||
def test_whitelist_rejects_protected_paths():
|
||
for path in ('/', '/api/ui/people', '/api/chat/ask', '/api/ss/status'):
|
||
assert auth._is_whitelisted(path) is False
|
||
|
||
|
||
def test_whitelist_no_longer_includes_removed_password_login_endpoint():
|
||
"""账号密码登录接口已随 SSO 改造下线，不应再出现在白名单里。"""
|
||
assert auth._is_whitelisted('/api/login') is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 通过 Flask test_client 打完整流程
|
||
# ---------------------------------------------------------------------------
|
||
@pytest.fixture
|
||
def app():
|
||
from flask import Flask, jsonify as _jsonify
|
||
app = Flask(__name__)
|
||
app.register_blueprint(auth.auth_bp)
|
||
|
||
@app.route('/some-protected-page')
|
||
def _protected_page():
|
||
return 'ok'
|
||
|
||
@app.route('/api/protected')
|
||
def _protected_api():
|
||
return _jsonify({"ok": True})
|
||
|
||
auth.init_auth(app)
|
||
return app
|
||
|
||
|
||
@pytest.fixture
|
||
def client(app):
|
||
return app.test_client()
|
||
|
||
|
||
def test_login_redirects_to_auth_hub_authorize_with_pkce(client, monkeypatch):
|
||
_set_auth_hub_env(monkeypatch)
|
||
resp = client.get('/login')
|
||
assert resp.status_code == 302
|
||
location = resp.headers['Location']
|
||
assert location.startswith('http://auth-hub.example/authorize?')
|
||
assert 'code_challenge=' in location
|
||
assert 'code_challenge_method=S256' in location
|
||
assert 'client_id=fam-core' in location
|
||
assert len(auth._pending) == 1
|
||
|
||
|
||
def test_login_rejects_when_unconfigured(client):
|
||
resp = client.get('/login')
|
||
assert resp.status_code == 503
|
||
|
||
|
||
def test_login_redirects_home_when_already_authed(client, monkeypatch):
|
||
_set_auth_hub_env(monkeypatch)
|
||
tok = auth._create_session()
|
||
client.set_cookie('fam_session', tok)
|
||
resp = client.get('/login')
|
||
assert resp.status_code == 302
|
||
assert resp.headers['Location'] == '/'
|
||
|
||
|
||
def test_callback_rejects_unknown_or_expired_state(client, monkeypatch):
|
||
_set_auth_hub_env(monkeypatch)
|
||
resp = client.get('/api/auth/callback?state=nope&code=abc')
|
||
assert resp.status_code == 302
|
||
assert resp.headers['Location'] == '/login'
|
||
assert 'fam_session' not in resp.headers.get('Set-Cookie', '')
|
||
|
||
|
||
def test_callback_rejects_idp_error(client, monkeypatch):
|
||
_set_auth_hub_env(monkeypatch)
|
||
resp = client.get('/api/auth/callback?error=access_denied&state=x')
|
||
assert resp.status_code == 302
|
||
assert resp.headers['Location'] == '/login'
|
||
|
||
|
||
def test_callback_exchanges_code_and_sets_session_cookie(client, monkeypatch):
|
||
_set_auth_hub_env(monkeypatch)
|
||
auth._pending['thestate'] = {'verifier': 'v', 'expires': time.time() + 600}
|
||
|
||
id_token = jwt.encode(
|
||
{'iss': 'http://auth-hub.example', 'aud': 'fam-core', 'sub': '1',
|
||
'preferred_username': 'ericwyuan', 'exp': time.time() + 300, 'iat': time.time()},
|
||
'unused', algorithm='HS256') # 签名算法在这里不重要，被下面的 mock 绕过验签
|
||
|
||
class _FakeResp:
|
||
status_code = 200
|
||
text = ''
|
||
def json(self):
|
||
return {'id_token': id_token, 'access_token': 'x', 'token_type': 'Bearer'}
|
||
|
||
monkeypatch.setattr(auth.requests, 'post', lambda *a, **k: _FakeResp())
|
||
|
||
class _FakeSigningKey:
|
||
key = 'unused'
|
||
|
||
class _FakeJwksClient:
|
||
uri = 'http://auth-hub.example/.well-known/jwks.json'
|
||
def get_signing_key_from_jwt(self, token):
|
||
return _FakeSigningKey()
|
||
|
||
monkeypatch.setattr(auth, '_get_jwks_client', lambda uri: _FakeJwksClient())
|
||
monkeypatch.setattr(auth.jwt, 'decode', lambda *a, **k: {
|
||
'preferred_username': 'ericwyuan', 'sub': '1'})
|
||
|
||
resp = client.get('/api/auth/callback?state=thestate&code=abc')
|
||
assert resp.status_code == 302
|
||
assert resp.headers['Location'] == '/'
|
||
assert 'fam_session=' in resp.headers['Set-Cookie']
|
||
assert 'thestate' not in auth._pending
|
||
assert len(auth._sessions) == 1
|
||
|
||
|
||
def test_callback_rejects_when_token_exchange_fails(client, monkeypatch):
|
||
_set_auth_hub_env(monkeypatch)
|
||
auth._pending['thestate'] = {'verifier': 'v', 'expires': time.time() + 600}
|
||
|
||
class _FakeResp:
|
||
status_code = 400
|
||
text = 'invalid_grant'
|
||
|
||
monkeypatch.setattr(auth.requests, 'post', lambda *a, **k: _FakeResp())
|
||
|
||
resp = client.get('/api/auth/callback?state=thestate&code=abc')
|
||
assert resp.status_code == 302
|
||
assert resp.headers['Location'] == '/login'
|
||
assert len(auth._sessions) == 0
|
||
|
||
|
||
def test_callback_rejects_when_id_token_verification_fails(client, monkeypatch):
|
||
_set_auth_hub_env(monkeypatch)
|
||
auth._pending['thestate'] = {'verifier': 'v', 'expires': time.time() + 600}
|
||
|
||
class _FakeResp:
|
||
status_code = 200
|
||
text = ''
|
||
def json(self):
|
||
return {'id_token': 'bogus'}
|
||
|
||
monkeypatch.setattr(auth.requests, 'post', lambda *a, **k: _FakeResp())
|
||
|
||
def _boom(uri):
|
||
raise jwt.PyJWTError("boom")
|
||
monkeypatch.setattr(auth, '_get_jwks_client', _boom)
|
||
|
||
resp = client.get('/api/auth/callback?state=thestate&code=abc')
|
||
assert resp.status_code == 302
|
||
assert resp.headers['Location'] == '/login'
|
||
assert len(auth._sessions) == 0
|
||
|
||
|
||
def test_logout_clears_session(client):
|
||
tok = auth._create_session()
|
||
client.set_cookie('fam_session', tok)
|
||
resp = client.post('/api/logout')
|
||
assert resp.status_code == 200
|
||
assert tok not in auth._sessions
|
||
|
||
|
||
def test_before_request_guard_redirects_unauthed_page_to_login(client):
|
||
resp = client.get('/some-protected-page')
|
||
assert resp.status_code == 302
|
||
assert resp.headers['Location'] == '/login'
|
||
|
||
|
||
def test_before_request_guard_401s_unauthed_api(client):
|
||
resp = client.get('/api/protected')
|
||
assert resp.status_code == 401
|
||
|
||
|
||
def test_before_request_guard_allows_authed_requests(client):
|
||
tok = auth._create_session()
|
||
client.set_cookie('fam_session', tok)
|
||
assert client.get('/some-protected-page').status_code == 200
|
||
assert client.get('/api/protected').status_code == 200
|