"""统一登录（2026-09-12 从 NAS fam-core 迁来）的单测。 除了把原 fam-core/tests/test_auth.py 的用例搬过来，额外盯死三件迁移时的关键行为： 回调失败不准再 302（防死循环复发）、服务端调 auth-hub 走内网地址但 iss 仍按公网校验、 会话 cookie 必须是无状态的（不依赖任何进程内状态）。 """ import time import jwt import pytest from flask import Flask from fam_edge import auth @pytest.fixture(autouse=True) def _clean_env_and_state(monkeypatch): for var in ('AUTH_HUB_ISSUER', 'AUTH_HUB_INTERNAL_BASE', 'AUTH_HUB_CLIENT_ID', 'AUTH_HUB_CLIENT_SECRET', 'AUTH_HUB_REDIRECT_URI', 'FAM_SESSION_SECRET'): monkeypatch.delenv(var, raising=False) auth._pending.clear() auth._warned_unconfigured = False auth._jwks_client = None yield auth._pending.clear() def _set_env(monkeypatch, internal_base=None): monkeypatch.setenv('AUTH_HUB_ISSUER', 'https://auth.example') monkeypatch.setenv('AUTH_HUB_CLIENT_ID', 'fam-core') monkeypatch.setenv('AUTH_HUB_CLIENT_SECRET', 'sekret') monkeypatch.setenv('AUTH_HUB_REDIRECT_URI', 'https://cam.example/api/auth/callback') monkeypatch.setenv('FAM_SESSION_SECRET', 'session-signing-secret') if internal_base: monkeypatch.setenv('AUTH_HUB_INTERNAL_BASE', internal_base) @pytest.fixture def client(): app = Flask(__name__) app.register_blueprint(auth.auth_bp) return app.test_client() def _cookie(secret='session-signing-secret', username='ericwyuan', ttl=3600): now = int(time.time()) return jwt.encode({'sub': '1', 'username': username, 'iat': now, 'exp': now + ttl}, secret, algorithm='HS256') # --------------------------------------------------------------------------- # 配置齐全性（fail closed） # --------------------------------------------------------------------------- def test_require_config_fails_closed_when_unconfigured(): assert auth._require_config() is None def test_require_config_fails_closed_without_session_secret(monkeypatch): _set_env(monkeypatch) monkeypatch.delenv('FAM_SESSION_SECRET') assert auth._require_config() is None def test_login_rejects_when_unconfigured(client): resp = client.get('/login') assert resp.status_code == 503 # 配置缺失也不能跳转，否则同样会跟 auth-hub 对跳 assert 'Location' not in resp.headers # --------------------------------------------------------------------------- # /login # --------------------------------------------------------------------------- def test_login_redirects_to_authorize_with_pkce(client, monkeypatch): _set_env(monkeypatch) resp = client.get('/login') assert resp.status_code == 302 location = resp.headers['Location'] assert location.startswith('https://auth.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_always_sends_browser_to_public_issuer(client, monkeypatch): """内网地址只给服务端自己用，浏览器必须跳公网——跳 127.0.0.1 用户当然打不开。""" _set_env(monkeypatch, internal_base='http://127.0.0.1:5300') resp = client.get('/login') assert resp.headers['Location'].startswith('https://auth.example/authorize?') def test_login_redirects_home_when_already_authed(client, monkeypatch): _set_env(monkeypatch) client.set_cookie('fam_session', _cookie()) resp = client.get('/login') assert resp.status_code == 302 assert resp.headers['Location'] == '/' # --------------------------------------------------------------------------- # 回调：失败分支一律错误页，绝不再跳 /login（旧实现死循环的根因） # --------------------------------------------------------------------------- def test_callback_unknown_state_shows_error_page_without_redirecting(client, monkeypatch): _set_env(monkeypatch) resp = client.get('/api/auth/callback?state=nope&code=abc') assert resp.status_code == 400 assert 'Location' not in resp.headers assert 'fam_session' not in resp.headers.get('Set-Cookie', '') def test_callback_idp_error_shows_error_page_without_redirecting(client, monkeypatch): _set_env(monkeypatch) resp = client.get('/api/auth/callback?error=access_denied&state=x') assert resp.status_code == 403 assert 'Location' not in resp.headers def test_callback_token_failure_shows_error_page_without_redirecting(client, monkeypatch): _set_env(monkeypatch) auth._pending['thestate'] = {'verifier': 'v', 'expires': time.time() + 600} class _Resp: status_code = 400 text = 'invalid_grant' monkeypatch.setattr(auth.requests, 'post', lambda *a, **k: _Resp()) resp = client.get('/api/auth/callback?state=thestate&code=abc') assert resp.status_code == 502 assert 'Location' not in resp.headers assert 'fam_session' not in resp.headers.get('Set-Cookie', '') # --------------------------------------------------------------------------- # 回调：成功路径 # --------------------------------------------------------------------------- def _stub_successful_exchange(monkeypatch, captured): class _Resp: status_code = 200 text = '' def json(self): return {'id_token': 'the-id-token', 'access_token': 'x'} def _post(url, **kwargs): captured['token_url'] = url return _Resp() monkeypatch.setattr(auth.requests, 'post', _post) class _FakeKey: key = 'unused' class _FakeJwks: uri = 'unused' def get_signing_key_from_jwt(self, token): return _FakeKey() def _jwks(uri): captured['jwks_uri'] = uri return _FakeJwks() monkeypatch.setattr(auth, '_get_jwks_client', _jwks) real_decode = auth.jwt.decode def _decode(token, key, **kw): # id_token 走 RS256：记下校验参数并返回固定 claims； # 会话 cookie 走 HS256：交给真正的实现，别把验签也 mock 掉 if kw.get('algorithms') == ['RS256']: captured['decode_kwargs'] = kw return {'sub': '1', 'preferred_username': 'ericwyuan'} return real_decode(token, key, **kw) monkeypatch.setattr(auth.jwt, 'decode', _decode) def test_callback_success_sets_cookie_and_goes_home(client, monkeypatch): _set_env(monkeypatch) auth._pending['thestate'] = {'verifier': 'v', 'expires': time.time() + 600} captured = {} _stub_successful_exchange(monkeypatch, captured) 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 'HttpOnly' in resp.headers['Set-Cookie'] assert 'thestate' not in auth._pending # 用过的 state 必须立刻作废 def test_callback_talks_to_internal_base_but_validates_public_issuer(client, monkeypatch): """换 token / 拉 JWKS 走本机，避免公网 TLS 那条链路上的 CA 和时钟坑； 但 id_token 里的 iss 是 auth-hub 配置的公网地址，校验必须按公网来。""" _set_env(monkeypatch, internal_base='http://127.0.0.1:5300') auth._pending['thestate'] = {'verifier': 'v', 'expires': time.time() + 600} captured = {} _stub_successful_exchange(monkeypatch, captured) client.get('/api/auth/callback?state=thestate&code=abc') assert captured['token_url'] == 'http://127.0.0.1:5300/token' assert captured['jwks_uri'] == 'http://127.0.0.1:5300/.well-known/jwks.json' assert captured['decode_kwargs']['issuer'] == 'https://auth.example' assert captured['decode_kwargs']['audience'] == 'fam-core' assert captured['decode_kwargs']['leeway'] == auth._LEEWAY # --------------------------------------------------------------------------- # 会话 cookie / check / verify / logout # --------------------------------------------------------------------------- def test_session_is_stateless(client, monkeypatch): """cookie 自带签名，服务端不存任何东西——换个进程、重启服务照样认。""" _set_env(monkeypatch) assert auth._pending == {} client.set_cookie('fam_session', _cookie()) assert client.get('/api/auth/verify').status_code == 204 def test_verify_rejects_missing_expired_and_forged_cookies(client, monkeypatch): _set_env(monkeypatch) assert client.get('/api/auth/verify').status_code == 401 client.set_cookie('fam_session', _cookie(ttl=-10)) assert client.get('/api/auth/verify').status_code == 401 client.set_cookie('fam_session', _cookie(secret='wrong-secret')) resp = client.get('/api/auth/verify') assert resp.status_code == 401 assert resp.get_json()['error'] == '未登录' def test_check_reports_auth_state(client, monkeypatch): _set_env(monkeypatch) assert client.get('/api/auth/check').get_json() == {'authed': False, 'username': ''} client.set_cookie('fam_session', _cookie(username='ericwyuan')) assert client.get('/api/auth/check').get_json() == {'authed': True, 'username': 'ericwyuan'} def test_logout_clears_cookie(client, monkeypatch): _set_env(monkeypatch) client.set_cookie('fam_session', _cookie()) resp = client.post('/api/logout') assert resp.status_code == 200 assert 'fam_session=;' in resp.headers['Set-Cookie']