ericwyuan/sentinel-home-ai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(fam-core): 登录改接 auth-hub 统一登录（OAuth2 Authorization Code + PKCE / OIDC）

Browse Source 
移除本地 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>
This commit is contained in:
ericwyuan
2026-08-31 21:32:26 +08:00
parent 988c63a8f9
commit 8120d2ae6a
6 changed files with 399 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
fam-core/requirements.txt
View File

@@ -3,3 +3,5 @@ gunicorn>=21.2.0
PyMySQL>=1.1.0
PyYAML>=6.0
requests>=2.31.0
PyJWT>=2.8.0
cryptography>=42.0.0

259
fam-core/src/fam_core/auth.py
View File

@@ -1,20 +1,28 @@
"""
Auth - FAM-Core 登录校验（2026-08-22 新增）
Auth - FAM-Core 登录校验（2026-08-22 新增；2026-08-31 接入 auth-hub 统一登录改造）
背景：NAS :8000（fam-core + FAM-UI）通过 frp 暴露到外网后，需要先登录才能访问。
- 账号密码：环境变量 FAM_AUTH_USER / FAM_AUTH_PASS，由 NAS 的 start_core.sh
source .env 注入。2026-08-23 起不再提供硬编码默认值——这两个环境变量跟
NAS SSH 密码是同一个值，公网入口用同一串密码兜底、还写死在代码里是双重
风险；.env 没配置好这两个变量时，直接拒绝所有登录（fail closed），而不是
悄悄退回一个大家都知道的密码。
- 登录方式：不再自己校验账号密码，全权委托给独立部署的 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 说明）。
重启进程后需重新登录（可接受，见 config/auth 说明）。这一段跟接入 SSO
之前完全一样，只是"怎么发这个 cookie"变了，下游（`is_authed()`/白名单/
拦截逻辑）不用改。
- 拦截策略（app.before_request 全局生效）：
* 页面路径未登录 -> 302 重定向 /login
* /api/* 未登录 -> 401 JSON
- 白名单免登录：
* /login /api/login /api/logout /api/auth/check —— 登录流程本身
* /login /api/auth/callback /api/logout /api/auth/check —— 登录流程本身
* /health —— 内部健康检查
* /api/ss/webhook —— Surveillance Station 推送无法携带登录态
* /assets/*、/favicon.ico —— SPA 静态资源
@@ -22,9 +30,13 @@ Auth - FAM-Core 登录校验（2026-08-22 新增）
import os
import secrets
import time
from base64 import urlsafe_b64encode
from hashlib import sha256
from urllib.parse import urlencode
from flask import (Blueprint, Response, jsonify, make_response,
redirect, request)
import jwt
import requests
from flask import Blueprint, jsonify, make_response, redirect, request
from .logger import setup_logger
@@ -33,25 +45,43 @@ 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 _check_credential(username: str, password: str) -> bool:
"""账号密码校验。凭据必须从环境变量读（.env 注入），不提供硬编码默认值——
这两个变量跟 NAS SSH 密码是同一个值，公网入口不能有"没配置就退回已知密码"
这种兜底，宁可直接拒绝所有登录（fail closed），等运维发现并配置好 .env。"""
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
user = os.environ.get('FAM_AUTH_USER', '')
pwd = os.environ.get('FAM_AUTH_PASS', '')
if not user or not pwd:
cfg = _auth_hub_config()
if not all(cfg.values()):
if not _warned_unconfigured:
logger.error(
"FAM_AUTH_USER/FAM_AUTH_PASS 未配置，拒绝所有登录——"
"请在 .env 里设置这两个环境变量后重启 fam-core")
"AUTH_HUB_ISSUER/CLIENT_ID/CLIENT_SECRET/REDIRECT_URI 未配置齐全，"
"拒绝所有登录——请在 .env 里设置后重启 fam-core")
_warned_unconfigured = True
return False
return (username or '') == user and (password or '') == pwd
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:
@@ -67,11 +97,24 @@ def is_authed() -> bool:
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/login', '/api/logout', '/api/auth/check',
'/login', '/api/auth/callback', '/api/logout', '/api/auth/check',
'/health', '/favicon.ico',
}
_WHITELIST_PREFIX = ('/assets/',)
@@ -103,24 +146,85 @@ def init_auth(app):
# ---------------------------------------------------------------------------
@auth_bp.route('/login', methods=['GET'])
def login_page():
"""不再自己渲染登录表单，直接跳到 auth-hub 走 Authorization Code + PKCE。"""
if is_authed():
return redirect('/')
return Response(_LOGIN_HTML, mimetype='text/html')
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/login', methods=['POST'])
def login():
data = request.get_json(silent=True) or request.form
username = (data.get('username') or '').strip()
password = data.get('password') or ''
if not _check_credential(username, password):
return jsonify({"error": "账号或密码错误"}), 401
tok = secrets.token_hex(24)
_sessions[tok] = time.time() + _SESSION_TTL
resp = make_response(jsonify({"ok": True}))
resp.set_cookie('fam_session', tok, max_age=_SESSION_TTL,
httponly=True, samesite='Lax', path='/')
return resp
@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'])
@@ -136,86 +240,3 @@ def logout():
@auth_bp.route('/api/auth/check', methods=['GET'])
def check():
return jsonify({"authed": is_authed()})
# ---------------------------------------------------------------------------
# 登录页（内嵌 HTML，深色风格与 fam-ui 一致；前端 SPA 无需改动）
# ---------------------------------------------------------------------------
_LOGIN_HTML = """<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>家庭监控 · 登录</title>
<style>
* { margin:0; padding:0; box-sizing:border-box; }
body {
min-height:100vh; display:flex; align-items:center; justify-content:center;
background: radial-gradient(1200px 600px at 20% -10%, #1e293b 0%, #0b1120 55%, #0b1120 100%);
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", "PingFang SC", "Microsoft YaHei", sans-serif;
color:#e2e8f0;
}
.card {
width:360px; padding:40px 36px 32px; border-radius:16px;
background: rgba(15,23,42,.75); border:1px solid rgba(148,163,184,.18);
box-shadow: 0 24px 64px rgba(0,0,0,.45);
}
.logo { font-size:26px; font-weight:700; letter-spacing:.5px; margin-bottom:6px; }
.logo span { color:#38bdf8; }
.sub { font-size:13px; color:#94a3b8; margin-bottom:28px; }
label { display:block; font-size:12px; color:#94a3b8; margin:14px 0 6px; }
input {
width:100%; padding:10px 12px; border-radius:8px; font-size:14px; color:#e2e8f0;
background:#0f172a; border:1px solid #334155; outline:none; transition:border .15s;
}
input:focus { border-color:#38bdf8; }
button {
width:100%; margin-top:24px; padding:11px; border:none; border-radius:8px;
font-size:14px; font-weight:600; color:#0b1120; background:#38bdf8; cursor:pointer;
transition:background .15s, transform .05s;
}
button:hover { background:#0ea5e9; }
button:active { transform: scale(.98); }
.err { margin-top:14px; font-size:13px; color:#f87171; min-height:18px; text-align:center; }
.foot { margin-top:22px; text-align:center; font-size:11px; color:#475569; }
</style>
</head>
<body>
<div class="card">
<div class="logo">家庭监控 <span>FAM</span></div>
<div class="sub">请输入账号密码登录后访问</div>
<form id="f">
<label for="u">账号</label>
<input id="u" name="username" autocomplete="username" autofocus required>
<label for="p">密码</label>
<input id="p" name="password" type="password" autocomplete="current-password" required>
<button type="submit">登 录</button>
</form>
<div class="err" id="err"></div>
<div class="foot">Sentinel Home AI</div>
</div>
<script>
document.getElementById('f').addEventListener('submit', async (e) => {
e.preventDefault();
const err = document.getElementById('err');
err.textContent = '';
try {
const r = await fetch('/api/login', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({
username: document.getElementById('u').value.trim(),
password: document.getElementById('p').value
})
});
if (r.ok) { location.href = '/'; return; }
const d = await r.json().catch(() => ({}));
err.textContent = d.error || ('登录失败（' + r.status + '）');
} catch (ex) {
err.textContent = '网络错误，请重试';
}
});
</script>
</body>
</html>
"""

247
fam-core/tests/test_auth.py
View File

@@ -1,51 +1,61 @@
import os
import time
from types import SimpleNamespace
import jwt
import pytest
from fam_core import auth
@pytest.fixture(autouse=True)
def _clean_env_and_sessions(monkeypatch):
"""每个用例前清掉环境变量和进程内 session 表，用例之间不互相污染。"""
monkeypatch.delenv('FAM_AUTH_USER', raising=False)
monkeypatch.delenv('FAM_AUTH_PASS', raising=False)
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 test_check_credential_fails_closed_when_unconfigured():
"""核心诉求: .env 没配置 FAM_AUTH_USER/FAM_AUTH_PASS 时必须拒绝所有登录，
不能退回任何硬编码默认账号密码（这两个变量跟 NAS SSH 密码是同一个值，公网
入口不能有"没配置就用已知密码兜底"这种行为）。"""
assert auth._check_credential('ericwyuan', 'iLoveJava5') is False
assert auth._check_credential('anything', 'anything') is False
assert auth._check_credential('', '') is False
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')
def test_check_credential_succeeds_with_matching_configured_values(monkeypatch):
monkeypatch.setenv('FAM_AUTH_USER', 'testuser')
monkeypatch.setenv('FAM_AUTH_PASS', 'testpass')
assert auth._check_credential('testuser', 'testpass') is True
# ---------------------------------------------------------------------------
# 配置齐全性校验（fail closed）
# ---------------------------------------------------------------------------
def test_require_auth_hub_config_fails_closed_when_unconfigured():
assert auth._require_auth_hub_config() is None
def test_check_credential_rejects_wrong_password_when_configured(monkeypatch):
monkeypatch.setenv('FAM_AUTH_USER', 'testuser')
monkeypatch.setenv('FAM_AUTH_PASS', 'testpass')
assert auth._check_credential('testuser', 'wrongpass') is False
assert auth._check_credential('wronguser', 'testpass') is False
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_check_credential_fails_closed_when_only_one_var_set(monkeypatch):
"""只配了一半（比如账号忘配密码）也要 fail closed，不能退化成"密码随便"。"""
monkeypatch.setenv('FAM_AUTH_USER', 'testuser')
assert auth._check_credential('testuser', '') is False
assert auth._check_credential('testuser', 'anything') is False
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
@@ -91,8 +101,18 @@ def test_is_authed_false_without_cookie():
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/login', '/api/logout', '/api/auth/check',
for path in ('/login', '/api/auth/callback', '/api/logout', '/api/auth/check',
'/health', '/favicon.ico'):
assert auth._is_whitelisted(path) is True
@@ -104,3 +124,176 @@ def test_whitelist_assets_prefix():
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