feat(fam-edge): 统一登录从 NAS fam-core 迁入，回调失败不再对跳死循环
登录入口原先跑在 NAS 的 fam-core 里，NAS 或 frp 隧道一挂， smart-camera.zichuan.xyz/login 直接 502——连登录页都打不开。登录是入口， 不该依赖家里那台机器，整体搬到甲骨文（前端静态文件和 auth-hub 本来就在这台）。 跟旧实现的三处关键差异： - 换 token / 拉 JWKS 走 AUTH_HUB_INTERNAL_BASE（本机 :5300），不再跨公网 TLS。 旧链路上 PyJWKClient 用 urllib + 系统 CA（群晖易 CERTIFICATE_VERIFY_FAILED）、 两机时钟偏差会让 iat 显得来自未来，这两个坑一起消失；iss 校验和浏览器跳转 仍用公网 issuer - 会话改无状态 HS256 签名 cookie，服务重启不掉线（旧实现是进程内 token 表） - 回调失败渲染错误页，不再 302 回 /login。旧实现失败即跳 /login，而 auth-hub 只要还有会话就立刻再签一个 code 跳回来，两边对跳成死循环，浏览器只报 「重定向次数过多」，既看不到登录页也看不到原因 fam-core 侧删除 auth.py 与 test_auth.py，鉴权改由甲骨文 Caddy 的 forward_auth 前置拦截（打 /api/auth/verify）。 测试：fam-edge 新增 16 个用例，含「失败分支绝不 302」回归测试、 「服务端走内网但 iss 按公网校验」、「cookie 无状态」。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
ericwyuan
@@ -2,6 +2,9 @@ flask>=2.0.0
|
||||
gunicorn>=20.0.0
|
||||
requests>=2.28.0
|
||||
PyYAML>=6.0
|
||||
# 统一登录：验 auth-hub 的 id_token 签名（RS256）+ 签发会话 cookie（HS256）
|
||||
PyJWT>=2.8.0
|
||||
cryptography>=42.0.0
|
||||
# google-generativeai 和 openai 为可选依赖（代码用 requests 直接调 REST API）
|
||||
# 如需 SDK 方式调用，取消注释并在 Python 3.9+ 环境安装:
|
||||
# google-generativeai>=0.5.0
|
||||
|
||||
@@ -6,6 +6,8 @@ FAM-Edge 主应用 - Flask 单进程（新架构 v2.1）
|
||||
- VideoQueue（生产-消费队列：同步落地新视频入队，独立消费者云端分析，模型超时 ×2）
|
||||
- PersonService（人物汇总合并，定时）
|
||||
- DiskGuard（磁盘空间守护，定时检查，剩余空间过低时清理最旧的已完成素材）
|
||||
- Auth（统一登录：/login + OIDC 回调 + 给 Caddy forward_auth 用的会话校验，
|
||||
2026-09-12 从 NAS fam-core 迁入，原因见 auth.py 开头）
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
@@ -16,6 +18,7 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from .config_loader import load_config
|
||||
from .logger import setup_logger
|
||||
from .api_gateway.api_gateway import api_bp
|
||||
from .auth import auth_bp
|
||||
from . import state
|
||||
from .video_queue import VideoQueue
|
||||
from .person_service import PersonService
|
||||
@@ -25,6 +28,7 @@ logger = setup_logger('fam-edge.app')
|
||||
|
||||
app = Flask(__name__)
|
||||
app.register_blueprint(api_bp)
|
||||
app.register_blueprint(auth_bp)
|
||||
|
||||
|
||||
@app.route('/', methods=['GET'])
|
||||
|
||||
285
fam-edge/src/fam_edge/auth.py
Normal file
285
fam-edge/src/fam_edge/auth.py
Normal file
@@ -0,0 +1,285 @@
|
||||
"""
|
||||
Auth - 摄像头系统统一登录（2026-09-12 从 NAS fam-core 迁到甲骨文 fam-edge）
|
||||
|
||||
背景：登录流程原先跑在 NAS 的 fam-core 里，NAS 一挂或 frp 隧道一断，
|
||||
`smart-camera.zichuan.xyz/login` 直接 502——连登录页都进不去。登录是入口，
|
||||
不该依赖家里那台机器，所以整体搬到甲骨文：前端静态文件和 auth-hub 本来就在这台，
|
||||
搬完之后 NAS 只剩纯数据接口。
|
||||
|
||||
跟旧实现（fam-core/auth.py，已删除）的三处关键差异：
|
||||
|
||||
1. **服务端调用 auth-hub 走本机**：换 token 和拉 JWKS 用 `AUTH_HUB_INTERNAL_BASE`
|
||||
（默认 http://127.0.0.1:5300），不再走公网 TLS。旧链路是 NAS 跨公网访问
|
||||
https://auth.zichuan.xyz，踩过两个坑：`jwt.PyJWKClient` 用 urllib + 系统 CA
|
||||
（不像 requests 自带 certifi），群晖上很容易 CERTIFICATE_VERIFY_FAILED；
|
||||
以及两台机器时钟偏差会让 id_token 的 iat 看起来"来自未来"。
|
||||
但 `iss` 校验和浏览器跳转仍然用公网 `AUTH_HUB_ISSUER`——id_token 里的 iss
|
||||
是 auth-hub 自己配的公网地址，浏览器也只能跳公网地址。
|
||||
|
||||
2. **登录态改成无状态签名 cookie**：HS256 签发，密钥 `FAM_SESSION_SECRET`。
|
||||
旧实现是进程内 token 表，服务一重启所有人被踢下线；fam-edge 是视频分析进程，
|
||||
重启比 fam-core 更频繁，进程内会话在这里完全不成立。
|
||||
|
||||
3. **回调失败渲染错误页，不再 302 回 /login**：旧实现每个失败分支都跳 /login，
|
||||
而 auth-hub 侧只要还有会话，/authorize 会立刻再签发一个 code 跳回来，
|
||||
两边对跳成死循环——浏览器只报"重定向次数过多"，既看不到登录页也看不到原因。
|
||||
|
||||
拦截不在本模块做：Caddy 用 forward_auth 打到 `/api/auth/verify`，
|
||||
校验通过才把 `/api/*` 反代到 NAS 的 fam-core。
|
||||
"""
|
||||
import os
|
||||
import secrets
|
||||
import time
|
||||
from base64 import urlsafe_b64encode
|
||||
from hashlib import sha256
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import jwt
|
||||
import requests
|
||||
from flask import Blueprint, Response, jsonify, make_response, redirect, request
|
||||
|
||||
# 导入即加载 /opt/fam-edge/.env（config_loader 模块级 _load_env_file），
|
||||
# 下面所有 os.environ.get 才读得到 AUTH_HUB_* / FAM_SESSION_SECRET。
|
||||
from . import config_loader # noqa: F401
|
||||
from .logger import setup_logger
|
||||
|
||||
logger = setup_logger('fam-edge.auth')
|
||||
|
||||
auth_bp = Blueprint('auth', __name__)
|
||||
|
||||
COOKIE_NAME = 'fam_session'
|
||||
_SESSION_TTL = 2 * 3600 # 登录态有效期 2 小时
|
||||
_PENDING_TTL = 10 * 60 # /login -> /api/auth/callback 之间的等待上限
|
||||
_LEEWAY = 60 # 验 id_token 时容忍的跨机器时钟偏差（秒）
|
||||
|
||||
_pending = {} # state -> {verifier, expires}（进程内，PKCE 用）
|
||||
_warned_unconfigured = False # 只在第一次拒绝登录时打一条警告，别刷屏
|
||||
_jwks_client = None # jwt.PyJWKClient 单例，内建 JWKS 缓存
|
||||
|
||||
|
||||
def _config():
|
||||
"""接入参数全部来自环境变量（/opt/fam-edge/.env）。"""
|
||||
issuer = os.environ.get('AUTH_HUB_ISSUER', '').rstrip('/')
|
||||
return {
|
||||
'issuer': issuer,
|
||||
# 服务端直连 auth-hub 的地址；没配就退回公网 issuer（本地开发用）
|
||||
'internal_base': (os.environ.get('AUTH_HUB_INTERNAL_BASE', '') or issuer).rstrip('/'),
|
||||
'client_id': os.environ.get('AUTH_HUB_CLIENT_ID', ''),
|
||||
'client_secret': os.environ.get('AUTH_HUB_CLIENT_SECRET', ''),
|
||||
'redirect_uri': os.environ.get('AUTH_HUB_REDIRECT_URI', ''),
|
||||
'session_secret': os.environ.get('FAM_SESSION_SECRET', ''),
|
||||
}
|
||||
|
||||
|
||||
def _require_config():
|
||||
"""五个变量缺任何一个都拒绝登录（fail closed）。返回配置 dict 或 None。
|
||||
|
||||
本服务监听 0.0.0.0，没有"配置不全就退回某种默认放行"这种兜底。
|
||||
"""
|
||||
global _warned_unconfigured
|
||||
cfg = _config()
|
||||
if not all([cfg['issuer'], cfg['client_id'], cfg['client_secret'],
|
||||
cfg['redirect_uri'], cfg['session_secret']]):
|
||||
if not _warned_unconfigured:
|
||||
logger.error(
|
||||
"AUTH_HUB_ISSUER/CLIENT_ID/CLIENT_SECRET/REDIRECT_URI 或 "
|
||||
"FAM_SESSION_SECRET 未配置齐全，拒绝所有登录——"
|
||||
"请在 /opt/fam-edge/.env 里补齐后 systemctl restart fam-edge")
|
||||
_warned_unconfigured = True
|
||||
return None
|
||||
return cfg
|
||||
|
||||
|
||||
def _get_jwks_client(jwks_uri):
|
||||
global _jwks_client
|
||||
if _jwks_client is None or _jwks_client.uri != jwks_uri:
|
||||
_jwks_client = jwt.PyJWKClient(jwks_uri)
|
||||
return _jwks_client
|
||||
|
||||
|
||||
def _cleanup_pending():
|
||||
now = time.time()
|
||||
for state in [s for s, v in _pending.items() if v['expires'] < now]:
|
||||
_pending.pop(state, None)
|
||||
|
||||
|
||||
def _is_https():
|
||||
"""Caddy 终止 TLS 后回源是明文 HTTP，request.is_secure 恒为 False，
|
||||
所以 cookie 的 Secure 标志要看 X-Forwarded-Proto。"""
|
||||
return request.headers.get('X-Forwarded-Proto', '') == 'https' or request.is_secure
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 会话（无状态签名 cookie）
|
||||
# ---------------------------------------------------------------------------
|
||||
def _issue_session(secret, sub, username):
|
||||
now = int(time.time())
|
||||
return jwt.encode({'sub': sub, 'username': username,
|
||||
'iat': now, 'exp': now + _SESSION_TTL},
|
||||
secret, algorithm='HS256')
|
||||
|
||||
|
||||
def current_user():
|
||||
"""返回 cookie 里的用户信息（dict），未登录/过期/签名不对返回 None。"""
|
||||
cfg = _config()
|
||||
if not cfg['session_secret']:
|
||||
return None
|
||||
token = request.cookies.get(COOKIE_NAME, '')
|
||||
if not token:
|
||||
return None
|
||||
try:
|
||||
# 这里刻意不给 leeway：cookie 是本进程自签自验的，没有跨机器时钟偏差问题，
|
||||
# 加了只会让每个会话白白多活 60 秒。_LEEWAY 只用于 auth-hub 签发的 id_token。
|
||||
return jwt.decode(token, cfg['session_secret'], algorithms=['HS256'])
|
||||
except jwt.PyJWTError:
|
||||
return None
|
||||
|
||||
|
||||
def is_authed():
|
||||
return current_user() is not None
|
||||
|
||||
|
||||
def _error_page(reason, status=502):
|
||||
"""回调失败时给一个看得懂的页面。
|
||||
|
||||
这里刻意不做自动跳转：auth-hub 有会话时会立刻再发一个 code 回来，
|
||||
自动跳转等于把用户关进死循环。让用户自己点"重新登录"，最多再失败一次。
|
||||
"""
|
||||
html = (
|
||||
'<!doctype html><html lang="zh-CN"><head><meta charset="utf-8">'
|
||||
'<meta name="viewport" content="width=device-width,initial-scale=1">'
|
||||
'<title>登录失败</title></head>'
|
||||
'<body style="font-family:system-ui,-apple-system,sans-serif;background:#0b0e14;'
|
||||
'color:#e6ebf2;display:flex;min-height:100vh;align-items:center;justify-content:center;margin:0">'
|
||||
'<div style="max-width:34rem;padding:2rem">'
|
||||
'<h1 style="font-size:1.25rem;margin:0 0 .75rem">登录没能完成</h1>'
|
||||
'<p style="color:#9aa7b8;line-height:1.7;margin:0 0 1.5rem">原因：{reason}</p>'
|
||||
'<a href="/login" style="display:inline-block;background:#5b8cff;color:#fff;'
|
||||
'text-decoration:none;padding:.6rem 1.2rem;border-radius:.6rem">重新登录</a>'
|
||||
'</div></body></html>'
|
||||
).format(reason=reason)
|
||||
return Response(html, status=status, mimetype='text/html; charset=utf-8')
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 端点
|
||||
# ---------------------------------------------------------------------------
|
||||
@auth_bp.route('/login', methods=['GET'])
|
||||
def login():
|
||||
"""不渲染登录表单，直接跳 auth-hub 走 Authorization Code + PKCE。"""
|
||||
if is_authed():
|
||||
return redirect('/')
|
||||
|
||||
cfg = _require_config()
|
||||
if not cfg:
|
||||
return _error_page('本服务的登录参数没配齐（管理员看 fam-edge 日志）', 503)
|
||||
|
||||
_cleanup_pending()
|
||||
verifier = secrets.token_urlsafe(48)
|
||||
challenge = urlsafe_b64encode(
|
||||
sha256(verifier.encode('ascii')).digest()).rstrip(b'=').decode('ascii')
|
||||
state = secrets.token_hex(24)
|
||||
_pending[state] = {'verifier': verifier, 'expires': time.time() + _PENDING_TTL}
|
||||
|
||||
params = {
|
||||
'response_type': 'code',
|
||||
'client_id': cfg['client_id'],
|
||||
'redirect_uri': cfg['redirect_uri'],
|
||||
'scope': 'openid profile',
|
||||
'state': state,
|
||||
'code_challenge': challenge,
|
||||
'code_challenge_method': 'S256',
|
||||
}
|
||||
# 浏览器要跳的是公网 issuer，不是 internal_base
|
||||
return redirect("{}/authorize?{}".format(cfg['issuer'], urlencode(params)))
|
||||
|
||||
|
||||
@auth_bp.route('/api/auth/callback', methods=['GET'])
|
||||
def callback():
|
||||
cfg = _require_config()
|
||||
if not cfg:
|
||||
return _error_page('本服务的登录参数没配齐（管理员看 fam-edge 日志）', 503)
|
||||
|
||||
if request.args.get('error'):
|
||||
logger.warning("auth-hub 拒绝授权: {}".format(request.args.get('error')))
|
||||
return _error_page('auth-hub 拒绝了这次授权：{}'.format(request.args.get('error')), 403)
|
||||
|
||||
state = request.args.get('state', '')
|
||||
code = request.args.get('code', '')
|
||||
pending = _pending.pop(state, None)
|
||||
if not pending or pending['expires'] < time.time() or not code:
|
||||
logger.warning("回调 state 缺失/过期/重放，拒绝")
|
||||
return _error_page('这次登录请求已过期或被重复使用，请重新登录', 400)
|
||||
|
||||
try:
|
||||
resp = requests.post(
|
||||
"{}/token".format(cfg['internal_base']),
|
||||
data={
|
||||
'grant_type': 'authorization_code',
|
||||
'code': code,
|
||||
'redirect_uri': cfg['redirect_uri'],
|
||||
'client_id': cfg['client_id'],
|
||||
'client_secret': cfg['client_secret'],
|
||||
'code_verifier': pending['verifier'],
|
||||
}, timeout=(5, 15))
|
||||
except requests.RequestException as e:
|
||||
logger.error("auth-hub /token 请求失败: {}".format(e))
|
||||
return _error_page('连不上 auth-hub 的 /token（{}）'.format(e))
|
||||
|
||||
if resp.status_code != 200:
|
||||
logger.warning("auth-hub /token 拒绝: {} {}".format(resp.status_code, resp.text[:200]))
|
||||
return _error_page('auth-hub 拒绝换发 token（HTTP {}）'.format(resp.status_code))
|
||||
|
||||
id_token = (resp.json() or {}).get('id_token', '')
|
||||
try:
|
||||
jwks_client = _get_jwks_client(
|
||||
"{}/.well-known/jwks.json".format(cfg['internal_base']))
|
||||
signing_key = jwks_client.get_signing_key_from_jwt(id_token)
|
||||
claims = jwt.decode(id_token, signing_key.key, algorithms=['RS256'],
|
||||
audience=cfg['client_id'], issuer=cfg['issuer'],
|
||||
leeway=_LEEWAY)
|
||||
except jwt.PyJWTError as e:
|
||||
logger.warning("id_token 验签/校验失败: {}".format(e))
|
||||
return _error_page('id_token 校验失败（{}）'.format(e))
|
||||
|
||||
username = claims.get('preferred_username', '')
|
||||
token = _issue_session(cfg['session_secret'], claims.get('sub', ''), username)
|
||||
resp2 = make_response(redirect('/'))
|
||||
resp2.set_cookie(COOKIE_NAME, token, max_age=_SESSION_TTL, httponly=True,
|
||||
samesite='Lax', secure=_is_https(), path='/')
|
||||
logger.info("登录成功 username={}".format(username))
|
||||
return resp2
|
||||
|
||||
|
||||
@auth_bp.route('/api/logout', methods=['POST'])
|
||||
def logout():
|
||||
"""只清本地 cookie，不动 auth-hub 上的登录态（那是全站 SSO 会话）。
|
||||
|
||||
cookie 是无状态签名的，服务端没法单独作废某一张——真要立刻全员下线，
|
||||
换掉 FAM_SESSION_SECRET 重启即可。
|
||||
"""
|
||||
resp = make_response(jsonify({"ok": True}))
|
||||
resp.delete_cookie(COOKIE_NAME, path='/')
|
||||
return resp
|
||||
|
||||
|
||||
@auth_bp.route('/api/auth/check', methods=['GET'])
|
||||
def check():
|
||||
user = current_user()
|
||||
return jsonify({"authed": user is not None,
|
||||
"username": (user or {}).get('username', '')})
|
||||
|
||||
|
||||
@auth_bp.route('/api/auth/verify', methods=['GET', 'POST', 'PUT', 'DELETE', 'PATCH'])
|
||||
def verify():
|
||||
"""给 Caddy forward_auth 用：2xx 放行，401 拦截。
|
||||
|
||||
Caddy 会把原始请求的 cookie 一起带过来，401 的响应体会原样回给浏览器，
|
||||
前端 api.js 看到 401 就会自动跳 /login。
|
||||
"""
|
||||
user = current_user()
|
||||
if not user:
|
||||
return jsonify({"error": "未登录", "code": 401}), 401
|
||||
resp = make_response('', 204)
|
||||
resp.headers['X-Auth-User'] = user.get('username', '')
|
||||
return resp
|
||||
241
fam-edge/tests/test_auth.py
Normal file
241
fam-edge/tests/test_auth.py
Normal file
@@ -0,0 +1,241 @@
|
||||
"""统一登录（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']
|
||||
Reference in New Issue
Block a user