diff --git a/fam-core/src/fam_core/auth.py b/fam-core/src/fam_core/auth.py
deleted file mode 100644
index 6318372..0000000
--- a/fam-core/src/fam_core/auth.py
+++ /dev/null
@@ -1,242 +0,0 @@
-"""
-Auth - FAM-Core 登录校验（2026-08-22 新增；2026-08-31 接入 auth-hub 统一登录改造）
-
-背景：NAS :8000（fam-core + FAM-UI）通过 frp 暴露到外网后，需要先登录才能访问。
-
-- 登录方式：不再自己校验账号密码，全权委托给独立部署的 auth-hub（OAuth2
- Authorization Code + PKCE / OIDC）。/login 直接 302 跳到 auth-hub 的
- /authorize，用户在 auth-hub 上登录后带 code 跳回本服务的
- /api/auth/callback，服务端拿 code 换 token、验完 id_token 签名后建立本地
- 会话——本服务自身不再保存、也不再校验任何密码。
-- 接入参数从环境变量读（.env 注入，由 start_core.sh source）：
- `AUTH_HUB_ISSUER` / `AUTH_HUB_CLIENT_ID` / `AUTH_HUB_CLIENT_SECRET` /
- `AUTH_HUB_REDIRECT_URI`（必须跟在 auth-hub 用 manage_clients 注册时登记的
- redirect_uri 逐字符一致，auth-hub 只做精确匹配）。四个变量任一没配置时直接
- 拒绝登录（fail closed）——跟之前账号密码时代的取舍一致：本服务监听
- 0.0.0.0，没有"配置不全就退回某种默认放行"这种兜底。
-- 登录态：进程内 token 表 + HttpOnly cookie（fam_session），2 小时有效；
- 重启进程后需重新登录（可接受，见 config/auth 说明）。这一段跟接入 SSO
- 之前完全一样，只是"怎么发这个 cookie"变了，下游（`is_authed()`/白名单/
- 拦截逻辑）不用改。
-- 拦截策略（app.before_request 全局生效）：
- * 页面路径未登录 -> 302 重定向 /login
- * /api/* 未登录 -> 401 JSON
-- 白名单免登录：
- * /login /api/auth/callback /api/logout /api/auth/check —— 登录流程本身
- * /health —— 内部健康检查
- * /api/ss/webhook —— Surveillance Station 推送无法携带登录态
- * /assets/*、/favicon.ico —— SPA 静态资源
-"""
-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, jsonify, make_response, redirect, request
-
-from .logger import setup_logger
-
-logger = setup_logger('fam-core.auth')
-
-auth_bp = Blueprint('auth', __name__)
-
-_SESSION_TTL = 2 * 3600 # cookie 有效期 2 小时
-_PENDING_TTL = 10 * 60 # /login -> /api/auth/callback 之间的等待上限
-_sessions = {} # token -> 过期时间戳（进程内；重启需重新登录）
-_pending = {} # state -> {verifier, expires}（进程内；PKCE 用）
-_warned_unconfigured = False # 只在第一次拒绝登录时打一条警告日志，别刷屏
-
-_jwks_client = None # jwt.PyJWKClient 单例，内建 JWKS 缓存
-
-
-def _auth_hub_config():
- """auth-hub 接入参数，四个变量都是必需的。"""
- return {
- 'issuer': os.environ.get('AUTH_HUB_ISSUER', ''),
- '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', ''),
- }
-
-
-def _require_auth_hub_config():
- """校验四个环境变量齐全，缺任何一个都拒绝（fail closed）。返回配置 dict 或 None。"""
- global _warned_unconfigured
- cfg = _auth_hub_config()
- if not all(cfg.values()):
- if not _warned_unconfigured:
- logger.error(
- "AUTH_HUB_ISSUER/CLIENT_ID/CLIENT_SECRET/REDIRECT_URI 未配置齐全，"
- "拒绝所有登录——请在 .env 里设置后重启 fam-core")
- _warned_unconfigured = True
- return None
- return cfg
-
-
-def _get_jwks_client(jwks_uri: str):
- global _jwks_client
- if _jwks_client is None or _jwks_client.uri != jwks_uri:
- _jwks_client = jwt.PyJWKClient(jwks_uri)
- return _jwks_client
-
-
-def is_authed() -> bool:
- tok = request.cookies.get('fam_session')
- if not tok:
- return False
- exp = _sessions.get(tok)
- if not exp:
- return False
- if time.time() > exp:
- _sessions.pop(tok, None)
- return False
- return True
-
-
-def _create_session() -> str:
- tok = secrets.token_hex(24)
- _sessions[tok] = time.time() + _SESSION_TTL
- return tok
-
-
-def _cleanup_pending():
- now = time.time()
- expired = [s for s, v in _pending.items() if v['expires'] < now]
- for s in expired:
- _pending.pop(s, None)
-
-
-# ---------------------------------------------------------------------------
-# 白名单（免登录）
-# ---------------------------------------------------------------------------
-_WHITELIST_EXACT = {
- '/login', '/api/auth/callback', '/api/logout', '/api/auth/check',
- '/health', '/favicon.ico',
-}
-_WHITELIST_PREFIX = ('/assets/',)
-
-
-def _is_whitelisted(path: str) -> bool:
- if path in _WHITELIST_EXACT:
- return True
- return any(path.startswith(p) for p in _WHITELIST_PREFIX)
-
-
-def init_auth(app):
- """注册全局登录拦截。需在注册完所有蓝图后调用。"""
-
- @app.before_request
- def _guard():
- path = request.path
- if _is_whitelisted(path):
- return None
- if is_authed():
- return None
- if path.startswith('/api/'):
- return jsonify({"error": "未登录", "code": 401}), 401
- return redirect('/login')
-
-
-# ---------------------------------------------------------------------------
-# 端点
-# ---------------------------------------------------------------------------
-@auth_bp.route('/login', methods=['GET'])
-def login_page():
- """不再自己渲染登录表单，直接跳到 auth-hub 走 Authorization Code + PKCE。"""
- if is_authed():
- return redirect('/')
-
- cfg = _require_auth_hub_config()
- if not cfg:
- return jsonify({"error": "登录服务未配置"}), 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',
- }
- return redirect(f"{cfg['issuer']}/authorize?{urlencode(params)}")
-
-
-@auth_bp.route('/api/auth/callback', methods=['GET'])
-def auth_callback():
- cfg = _require_auth_hub_config()
- if not cfg:
- return jsonify({"error": "登录服务未配置"}), 503
-
- if request.args.get('error'):
- logger.warning(f"auth-hub 登录被拒绝: {request.args.get('error')}")
- return redirect('/login')
-
- 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("auth-hub 回调 state 缺失/过期/重放，拒绝")
- return redirect('/login')
-
- try:
- resp = requests.post(
- f"{cfg['issuer']}/token",
- 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=(10, 15))
- except requests.RequestException as e:
- logger.error(f"auth-hub /token 请求失败: {e}")
- return redirect('/login')
-
- if resp.status_code != 200:
- logger.warning(f"auth-hub /token 拒绝: {resp.status_code} {resp.text[:200]}")
- return redirect('/login')
-
- id_token = (resp.json() or {}).get('id_token', '')
- try:
- jwks_client = _get_jwks_client(f"{cfg['issuer']}/.well-known/jwks.json")
- 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'])
- except jwt.PyJWTError as e:
- logger.warning(f"id_token 验签/校验失败: {e}")
- return redirect('/login')
-
- username = claims.get('preferred_username', '')
- tok = _create_session()
- resp2 = make_response(redirect('/'))
- resp2.set_cookie('fam_session', tok, max_age=_SESSION_TTL,
- httponly=True, samesite='Lax', path='/')
- logger.info(f"登录成功 username={username}")
- return resp2
-
-
-@auth_bp.route('/api/logout', methods=['POST'])
-def logout():
- tok = request.cookies.get('fam_session')
- if tok:
- _sessions.pop(tok, None)
- resp = make_response(jsonify({"ok": True}))
- resp.delete_cookie('fam_session', path='/')
- return resp
-
-
-@auth_bp.route('/api/auth/check', methods=['GET'])
-def check():
- return jsonify({"authed": is_authed()})
diff --git a/fam-core/tests/test_auth.py b/fam-core/tests/test_auth.py
deleted file mode 100644
index ab0afb6..0000000
--- a/fam-core/tests/test_auth.py
+++ /dev/null
@@ -1,299 +0,0 @@
-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
diff --git a/fam-edge/requirements.txt b/fam-edge/requirements.txt
index f883577..401fadd 100644
--- a/fam-edge/requirements.txt
+++ b/fam-edge/requirements.txt
@@ -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
diff --git a/fam-edge/src/fam_edge/app.py b/fam-edge/src/fam_edge/app.py
index 03129e1..2f57492 100644
--- a/fam-edge/src/fam_edge/app.py
+++ b/fam-edge/src/fam_edge/app.py
@@ -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'])
diff --git a/fam-edge/src/fam_edge/auth.py b/fam-edge/src/fam_edge/auth.py
new file mode 100644
index 0000000..780d851
--- /dev/null
+++ b/fam-edge/src/fam_edge/auth.py
@@ -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 = (
+ ''
+ ''
+ '登录失败'
+ ''
+ ''
+ '
登录没能完成
'
+ '
原因：{reason}
'
+ '
重新登录'
+ '