"""
Auth - FAM-Core 登录校验（2026-08-22 新增）
背景：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），而不是
悄悄退回一个大家都知道的密码。
- 登录态：进程内 token 表 + HttpOnly cookie（fam_session），2 小时有效；
重启进程后需重新登录（可接受，见 config/auth 说明）。
- 拦截策略（app.before_request 全局生效）：
* 页面路径未登录 -> 302 重定向 /login
* /api/* 未登录 -> 401 JSON
- 白名单免登录：
* /login /api/login /api/logout /api/auth/check —— 登录流程本身
* /health —— 内部健康检查
* /api/ss/webhook —— Surveillance Station 推送无法携带登录态
* /assets/*、/favicon.ico —— SPA 静态资源
"""
import os
import secrets
import time
from flask import (Blueprint, Response, 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 小时
_sessions = {} # token -> 过期时间戳（进程内；重启需重新登录）
_warned_unconfigured = False # 只在第一次拒绝登录时打一条警告日志，别刷屏
def _check_credential(username: str, password: str) -> bool:
"""账号密码校验。凭据必须从环境变量读（.env 注入），不提供硬编码默认值——
这两个变量跟 NAS SSH 密码是同一个值，公网入口不能有"没配置就退回已知密码"
这种兜底，宁可直接拒绝所有登录（fail closed），等运维发现并配置好 .env。"""
global _warned_unconfigured
user = os.environ.get('FAM_AUTH_USER', '')
pwd = os.environ.get('FAM_AUTH_PASS', '')
if not user or not pwd:
if not _warned_unconfigured:
logger.error(
"FAM_AUTH_USER/FAM_AUTH_PASS 未配置，拒绝所有登录——"
"请在 .env 里设置这两个环境变量后重启 fam-core")
_warned_unconfigured = True
return False
return (username or '') == user and (password or '') == pwd
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
# ---------------------------------------------------------------------------
# 白名单（免登录）
# ---------------------------------------------------------------------------
_WHITELIST_EXACT = {
'/login', '/api/login', '/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():
if is_authed():
return redirect('/')
return Response(_LOGIN_HTML, mimetype='text/html')
@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/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()})
# ---------------------------------------------------------------------------
# 登录页（内嵌 HTML，深色风格与 fam-ui 一致；前端 SPA 无需改动）
# ---------------------------------------------------------------------------
_LOGIN_HTML = """
家庭监控 · 登录
"""