From ceb4404c6a350e907c462e12be3292a21311b372 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=96=86OvO?= <1261686748@qq.com> Date: Tue, 11 Aug 2026 22:07:55 +0800 Subject: [PATCH] New Database --- Dockerfile | 14 +- README.md | 175 +++++---- backend/alarm_mgr.py | 1 - backend/app.py | 92 +++-- backend/auth.py | 7 +- backend/calendar_mgr.py | 5 +- backend/countdown_mgr.py | 1 - backend/db_admin.py | 263 +++++++++++++ backend/events.py | 3 - backend/models.py | 473 +++++++++++++---------- backend/pg_compat.py | 252 ++++++++++++ backend/pomodoro_mgr.py | 3 - backend/routes.py | 186 +++++---- backend/scheduler.py | 131 ++++++- backend/time_util.py | 5 +- backend/weather_mgr.py | 6 +- backend/webauthn_mgr.py | 2 - docker-compose.yml | 50 ++- frontend/static/js/control.js | 146 +++++-- frontend/static/js/display.js | 112 +++++- frontend/templates/control.html | 18 +- frontend/templates/login.html | 4 - frontend/templates/login_control.html | 4 - frontend/templates/register_request.html | 151 -------- requirements.txt | 2 + run.py | 42 +- start.py | 83 +++- tests/test_multi_user.py | 362 ++++++++++++----- 28 files changed, 1825 insertions(+), 768 deletions(-) create mode 100644 backend/db_admin.py create mode 100644 backend/pg_compat.py delete mode 100644 frontend/templates/register_request.html diff --git a/Dockerfile b/Dockerfile index 0dbb89a..589b394 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,14 +1,13 @@ FROM python:3.11-slim -# System dependencies +# System dependencies (mirrors.aliyun.com for CN networks) RUN sed -i 's/deb.debian.org/mirrors.aliyun.com/g' /etc/apt/sources.list.d/debian.sources && \ apt-get update && apt-get install -y --no-install-recommends \ git \ - libgomp1 \ && rm -rf /var/lib/apt/lists/* -# Clone project -RUN git clone https://gitee.com/zhe-ovo/iot-dboard.git /app +# 复制项目代码（含 .git，供 compose 的 app_code 卷内 git pull 更新工作流使用） +COPY . /app WORKDIR /app # Install Python dependencies @@ -16,4 +15,9 @@ RUN pip install --no-cache-dir -r requirements.txt EXPOSE 5000 -CMD ["python", "start.py"] +# Gunicorn：2 个 worker 对应 2 逻辑核心（1 物理核心），线程模式； +# 2G 内存/2 CPU 限制由 docker-compose.yml 的 deploy.resources 配置。 +# 后台调度器通过 PostgreSQL advisory lock 保证只在其中一个 worker 内运行。 +CMD gunicorn --workers ${GUNICORN_WORKERS:-2} --threads ${GUNICORN_THREADS:-8} \ + --bind 0.0.0.0:5000 --timeout 120 \ + --access-logfile - --error-logfile - start:app diff --git a/README.md b/README.md index c5392eb..1c82443 100644 --- a/README.md +++ b/README.md @@ -11,12 +11,14 @@ - **控制端** (`/control`) — 完整配置面板 - **Root 管理员** (`id=0`) — 首个自动创建的账户，拥有全部功能与用户管理权限 +数据库使用 **PostgreSQL**（Docker 内置），生产以 **Gunicorn** 多 worker 运行。 + --- ## 功能 - **多用户**：Root 管理员可注册/编辑/删除用户；所有用户数据完全隔离，普通用户无管理权限 -- **两种注册方式**：管理员代注册（正式账户，可代配 TOTP/Passkey）；用户自助申请（管理员确认后生成临时账户） +- **账户创建**：仅管理员可创建 —— 管理员代注册（正式账户，可代配 TOTP/Passkey）或创建临时账户（用户自助填写登录方式） - **临时账户**：服务器持有临时 TOTP 密钥，管理员仅可见实时验证码；用户登录后被强制引导配置自己的登录方式，配置完成即正式生效 - **日历**：支持订阅 iCalendar (.ics) 链接及本地文件上传，自动同步日程 - **天气**：基于和风天气 (QWeather) API，含逐小时/分钟级预报 @@ -24,6 +26,7 @@ - **番茄钟**：定时提醒，自动继续，断网重连自动恢复 - **倒计时**：可自定义时长，到时提醒 - **自定义展示页**：支持多组自定义 HTML 模板，一键切换 +- **数据库管理（仅 Root）**：在 `/control → 设置 → 数据库管理` 导入旧版 SQLite 数据库文件或本系统导出的备份（自动识别格式）、导出现有数据库备份 --- @@ -41,7 +44,7 @@ **登录流程**： ``` -首次启动（无数据库）→ 自动创建 Root 管理员 (id=0)，控制台打印临时密码 +首次启动（空数据库）→ 自动创建 Root 管理员 (id=0)，控制台打印临时密码 ↓ 访问 /control → 输入 6 位数字（自动识别 TOTP / 临时账户 TOTP / Root 临时密码） ↓ @@ -58,16 +61,16 @@ Root 临时密码 → 强制跳转 /mfa-setup 配置 MFA **用户管理（仅 Root）**： - 管理员代注册：填写信息 → 正式账户，可随后代配置 TOTP（生成密钥+二维码）或 Passkey（注册仪式在当前浏览器完成） -- 用户自助申请：用户在 `/register-request` 填写信息 → 管理员确认 → 生成临时账户（服务器持有临时 TOTP 密钥，管理员仅看到实时验证码） -- 注册/编辑备注/删除用户均需**二次验证管理员身份**（Passkey 优先，无 Passkey 时用 TOTP） +- 创建临时账户：生成临时 TOTP（密钥仅存服务器），管理员仅看到实时验证码 +- 注册/编辑备注/删除用户均需**二次验证管理员身份**（TOTP 优先，无 TOTP 时用 Passkey） - 删除用户会级联删除其全部数据并使其会话立即失效 - 用户 ID 对用户透明（仅在备注默认为 ID 时可见一次），普通用户只能看到自己的数据 **安全防护**： - 登录频率限制（渐进式封禁：5次→5分钟、10次→15分钟、15次→24小时） - CSRF 防护（Origin/Referer 校验） -- 敏感操作二次验证（用户管理、MFA 配置需重新确认身份） -- **双 Token 认证**：短期 access token（`Authorization: Bearer`，15 分钟）+ 长期 refresh token（httpOnly cookie，30 天滑动续期、每次刷新轮换） +- 敏感操作二次验证（用户管理、MFA 配置、数据库管理需重新确认身份） +- **双 Token 认证**：短期 access token（`Authorization: Bearer`，15 分钟）+ 长期 refresh token（httpOnly cookie，30 天滑动续期、每次刷新轮换、并发重放安全） - Access token 快照会话版本：删除用户 / 踢设备 / 踢所有用户后**立即**使全部 access token 失效 - 显示端 SSE 通过一次性短时 ticket 鉴权（EventSource 无法带自定义 header） - 内容安全策略（CSP）限制脚本/样式来源 @@ -78,21 +81,30 @@ Root 临时密码 → 强制跳转 /mfa-setup 配置 MFA ## 快速开始 +### 环境要求 + +- Python 3.11+ +- PostgreSQL 14+（Docker 部署时无需单独安装，镜像内置） + ### 开发环境 ```bash -# 1. 安装依赖 +# 1. 准备 PostgreSQL（本机或 Docker） +# docker run -d --name pg -e POSTGRES_USER=iot -e POSTGRES_PASSWORD=iot \ +# -e POSTGRES_DB=iot_dboard -p 5432:5432 postgres:16-alpine + +# 2. 安装依赖 pip install -r requirements.txt -# 2. 直接运行（调试模式） -python start.py +# 3. 运行（开发模式） +DATABASE_URL="postgresql://iot:iot@localhost:5432/iot_dboard" python run.py -# 3. 访问 -# 显示端: http://localhost:5000 -# 控制端: http://localhost:5000/control +# 4. 访问 +# 显示端: http://localhost:5001 +# 控制端: http://localhost:5001/control ``` -### 生产部署 (推荐 Nginx + Certbot) +### 生产部署 (推荐 Nginx + Gunicorn) ``` 客户端 (HTTPS) @@ -102,10 +114,10 @@ Nginx (SSL termination + Certbot) │ proxy_pass http://127.0.0.1:5000 │ ▼ 5000 (HTTP) -Flask (127.0.0.1:5000) +Gunicorn (2 workers × 2 threads, start:app) │ ▼ -data.db (SQLite) +PostgreSQL (Docker 内置, 端口 5432) ``` #### 1. Nginx 配置 @@ -125,30 +137,22 @@ server { proxy_set_header Host $host; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; + proxy_read_timeout 300s; # SSE 长连接 } - # 可选：限制上传大小 - client_max_body_size 5M; -} - -# HTTP → HTTPS 强制跳转 -server { - listen 80; - server_name your-domain.com; - return 301 https://$host$request_uri; + client_max_body_size 200M; # 数据库导入需要较大体积 } ``` -#### 2. 启动 Flask +#### 2. 启动 Gunicorn ```bash -# 生产模式启动 -FLASK_HTTPS=yes python start.py - -# 可选：固定 SECRET_KEY（保持跨重启会话有效） -export FLASK_SECRET_KEY="$(openssl rand -hex 32)" +DATABASE_URL="postgresql://iot:iot@127.0.0.1:5432/iot_dboard" \ +FLASK_HTTPS=yes gunicorn --workers 2 --threads 2 --bind 127.0.0.1:5000 start:app ``` +后台调度器（闹钟/提醒/天气/番茄钟）通过 **PostgreSQL advisory lock** 保证只在其中一个 worker 内运行；若持有者异常退出，其他 worker 的 watchdog 会自动接管。 + #### 3. Certbot 申请证书 ```bash @@ -157,14 +161,12 @@ sudo certbot --nginx -d your-domain.com --- ---- - ## Docker 部署 ### 快速启动 ```bash -# 构建并启动（后台运行，首次会自动从 Gitee clone 代码） +# 构建并启动（自动创建 PostgreSQL 与 Web 服务） docker compose up -d # 查看启动日志（获取临时密码） @@ -195,36 +197,36 @@ docker compose up -d ```yaml environment: - - FLASK_BIND=0.0.0.0 + - DATABASE_URL=postgresql://iot:${POSTGRES_PASSWORD:-iot}@db:5432/iot_dboard - FLASK_SECRET_KEY=your-secret-here # 可选：固定密钥 - FLASK_HTTPS=yes # 如果前面有 HTTPS 反代 + - GUNICORN_WORKERS=2 # 默认 2（对应 1 物理核心 = 2 逻辑核心） + - GUNICORN_THREADS=2 +``` + +### 资源限制 + +`docker-compose.yml` 中已按 **1 物理核心（2 逻辑核心）+ 2G 内存** 配置： + +```yaml +deploy: + resources: + limits: + cpus: "2" + memory: 2G ``` ### 持久化数据 -- **数据库**：`./data/` 目录挂载到 `/app/data/`，数据库文件 `data.db` 自动在此创建 -- **项目代码**：命名卷 `app_code` 挂载到 `/app`，首次自动初始化为镜像内克隆的代码，之后 `git pull` 的修改永久保留数据库文件 `data.db` 会自动在该目录下创建，重启/升级后数据保留。 - -```bash -# 数据文件位置 -ls -la data/ -# data.db ← 数据库 -``` +- **PostgreSQL 数据**：命名卷 `pgdata`（`docker compose down` 不丢失） +- **旧数据库导入**：将旧 SQLite `.db` 文件放入 `./data/` 目录后，在 `/control → 设置 → 数据库管理` 上传导入（也支持直接选择文件上传） +- **项目代码**：命名卷 `app_code` 挂载到 `/app`，首次自动初始化为镜像内代码，之后 `git pull` 的修改永久保留 ### 搭配 Nginx 反代（生产） -使用宿主机上的 Nginx（或额外 Nginx 容器）做 SSL 终结。配置与前文 Nginx 部分相同，只需将 `proxy_pass` 指向 Docker 宿主机 IP:5000： +使用宿主机上的 Nginx（或额外 Nginx 容器）做 SSL 终结。配置与前文 Nginx 部分相同，只需将 `proxy_pass` 指向 Docker 宿主机 IP:5000。 -```nginx -location / { - proxy_pass http://宿主机IP:5000; - proxy_set_header X-Forwarded-For $remote_addr; - proxy_set_header X-Forwarded-Proto $scheme; - proxy_set_header Host $host; -} -``` - -> ⚠️ 注意：容器内 Flask 绑定 `0.0.0.0:5000`，宿主机可直接访问 `localhost:5000`。如需限制仅 Nginx 访问，在 `docker-compose.yml` 中将 `ports` 改为 `127.0.0.1:5000:5000`。 +> ⚠️ 注意：容器内 Gunicorn 绑定 `0.0.0.0:5000`，宿主机可直接访问 `localhost:5000`。如需限制仅 Nginx 访问，在 `docker-compose.yml` 中将 `ports` 改为 `127.0.0.1:5000:5000`。 --- @@ -232,27 +234,59 @@ location / { | 变量 | 默认值 | 说明 | |------|--------|------| +| `DATABASE_URL` | `postgresql://iot:iot@db:5432/iot_dboard` | PostgreSQL 连接串（Docker 内指向 db 服务） | | `FLASK_HTTPS` | 空 | 设为 `yes`/`true` 启用 refresh token cookie 的 Secure 属性（Nginx 反代时启用） | | `FLASK_SECRET_KEY` | 自动随机生成 | 保留兼容；认证已由 access/refresh 双 token 承载，不再依赖 Flask session | -| `FLASK_BIND` | `127.0.0.1` | 监听地址，Docker 部署需设为 `0.0.0.0` | -| `FLASK_DB_PATH` | `./data.db` | 数据库文件路径，Docker 挂载数据卷时需同步修改 | -| `FLASK_DEBUG` | 空 | 设为 `true` 开启 Flask 调试模式（**生产环境禁用**） | +| `FLASK_DISABLE_SCHEDULER` | 空 | 设为 `true` 禁用后台调度器（调试用） | +| `GUNICORN_WORKERS` | `2` | Gunicorn worker 数（对应逻辑核心数） | +| `GUNICORN_THREADS` | `2` | 每个 worker 的线程数 | + +--- + +## 数据库管理（仅 Root） + +> 位置：`/control → 设置 → 数据库管理`（仅 Root 用户可见） + +- **导出现有数据库**：下载自包含的 JSON 备份（含 schema 版本与全部数据，不含会话令牌；**含登录凭据，请妥善保管**） +- **导入旧版 SQLite 数据库**：上传旧版单用户 `.db` 文件，自动迁移（旧数据归属 Root、缺失字段用 DEFAULT 填充、旧版弃用表自动丢弃、旧版 PIN 不会迁移） +- **导入备份文件**：上传本系统导出的 JSON 备份（与 SQLite 文件自动识别） + +导入为**单事务原子操作**：任何一行失败都会整体回滚，当前数据不受影响；导入成功会清空会话（所有设备需重新登录）。 + +--- + +## 数据库迁移（版本化，升级无需重写数据） + +数据库内置 `meta` 表记录 **schema 版本**（程序内 `DB_VERSION`，当前为 **2**）。每次启动时比对： + +| 情况 | 行为 | +|------|------| +| 数据库版本 == 程序版本 | 直接使用，跳过升级 | +| 无版本字段或 数据库版本 < 程序版本 | 创建缺失的表/列，缺失字段一律用 SCHEMA DEFAULT 填充，并更新版本号 | +| 数据库版本 > 程序版本 | **报错并终止启动**（数据库比程序新，需先升级程序） | + +从旧版 SQLite 数据库迁移到 PostgreSQL **不依赖自动迁移**：由 Root 在 `/control → 设置 → 数据库管理` 手动导入旧 `.db` 文件完成。 + +运行测试：`python3 tests/test_multi_user.py`（需本地 PostgreSQL，可用环境变量 `TEST_PG_HOST/TEST_PG_USER/TEST_PG_PASS/TEST_PG_DB` 覆盖）。 --- ## 文件结构 ``` -├── run.py # 入口（端口 5001，仅开发） -├── start.py # 入口（端口 5000，推荐生产使用） +├── start.py # 生产入口（Gunicorn WSGI: start:app，含调度器单例） +├── run.py # 开发入口（Flask dev server，端口 5001） +├── Dockerfile # 镜像构建（Gunicorn） +├── docker-compose.yml # PostgreSQL + Web 服务编排 ├── requirements.txt ├── backend/ │ ├── app.py # Flask 应用工厂 + 安全配置 │ ├── auth.py # 双 Token 认证（TOTP/Passkey/恢复码/临时密码） -│ ├── webauthn_mgr.py # WebAuthn (Passkey) 注册与认证 -│ ├── jwt_util.py # Ed25519 JWT 签名（和风天气） -│ ├── models.py # SQLite 模型、schema 版本迁移、token 管理 +│ ├── pg_compat.py # sqlite3 兼容的 PostgreSQL 访问层（方言翻译） +│ ├── models.py # 数据模型、schema 版本迁移、token 管理 +│ ├── db_admin.py # 数据库导入/导出（SQLite/JSON、原子恢复） │ ├── routes.py # 全部 API 路由 + CSRF 防护 + SSE ticket +│ ├── scheduler.py # 后台调度器（advisory lock 单例 + watchdog） │ └── *_mgr.py # 各功能模块管理器 └── frontend/ ├── templates/ # HTML 模板 @@ -260,29 +294,18 @@ location / { ├── css/ # 样式 ├── js/ # JavaScript（纯原生，无框架） │ ├── token.js # 双 Token 客户端（access 存储/刷新/401 重试/SSE ticket） - │ ├── control.js # 控制端逻辑 - │ └── display.js # 显示端逻辑（账户时区渲染） + │ ├── control.js # 控制端逻辑（含数据库管理） + │ └── display.js # 显示端逻辑（账户时区渲染、本地到期兜底） └── ringtones/ # 自定义铃声 ``` --- -### 数据库迁移（版本化，升级无需重写数据） - -数据库内置 `meta` 表记录 **schema 版本**（程序内 `DB_VERSION`，当前为 1）。每次启动时比对： - -| 情况 | 行为 | -|------|------| -| 数据库版本 == 程序版本 | 直接使用，跳过重建 | -| 无版本字段（旧版数据库）或 数据库版本 < 程序版本 | 重建数据库并迁移（旧版单用户数据自动归属 Root (id=0)，新增字段一律用 DEFAULT 填充，`camera_config` 等已删除的表自动丢弃） | -| 数据库版本 > 程序版本 | **报错并终止启动**（数据库比程序新，需先升级程序） | - -旧版 4 位 PIN 不会被迁移。运行测试：`python3 tests/test_multi_user.py`。 - ## 注意 - **Passkey (WebAuthn) 需要 HTTPS**（localhost 除外） -- 旧版 4 位 PIN 已弃用，数据库迁移时旧 PIN 不会被迁移 +- 旧版 4 位 PIN 已弃用，SQLite 导入时旧 PIN 不会被迁移 +- 多 Gunicorn worker 下，显示端闹钟/番茄钟/倒计时提醒除 SSE 即时推送外，还有**本地到期检测兜底**，保证到点必定提醒 ### License MIT diff --git a/backend/alarm_mgr.py b/backend/alarm_mgr.py index 85ebb1d..db3f0e9 100644 --- a/backend/alarm_mgr.py +++ b/backend/alarm_mgr.py @@ -84,7 +84,6 @@ def delete_alarm(alarm_id: int, user_id: int = None) -> bool: def check_alarms(user_id: int = None) -> list: if user_id is None: user_id = _uid() - global _triggered_state db = get_db(current_app.config["DB_PATH"]) now = tz_now() diff --git a/backend/app.py b/backend/app.py index 54e6fa6..49d574a 100644 --- a/backend/app.py +++ b/backend/app.py @@ -1,16 +1,12 @@ """Flask application factory and SSE event system.""" -import json import os -import queue import secrets -import sqlite3 -import threading -from datetime import datetime, timedelta +from datetime import timedelta from backend.time_util import now as tz_now, strptime as tz_strptime from flask import Flask from flask_cors import CORS from werkzeug.middleware.proxy_fix import ProxyFix -from backend.models import init_db +from backend.models import init_db, get_db from backend.routes import register_routes from backend.events import EventBus from backend.weather_mgr import fetch_weather, fetch_hourly @@ -70,33 +66,69 @@ def create_app(db_path: str = "data.db") -> Flask: return app -def _startup_weather(app, db_path: str): - try: - conn = sqlite3.connect(db_path) - row = conn.execute( - "SELECT updated_at FROM weather_cache WHERE user_id=0 ORDER BY id DESC LIMIT 1" - ).fetchone() - conn.close() - except Exception: - row = None +# Startup weather fetch lock: only one Gunicorn worker may fetch at a time +# (others skip; the cache is then fresh so the regular scheduler job handles it). +_WEATHER_STARTUP_LOCK_KEY = 0x5745415448455253 # "WEATHERS" - should_fetch = True - if row and row[0]: + +def _startup_weather(app, db_path: str): + # Multi-worker guard: without this, every worker would hit the weather API + # concurrently on boot. Whoever wins the advisory lock runs the check/fetch. + lock_conn = None + got_lock = False + try: + import psycopg2 + lock_conn = psycopg2.connect(db_path, connect_timeout=5) + cur = lock_conn.cursor() + cur.execute("SELECT pg_try_advisory_lock(%s)", (_WEATHER_STARTUP_LOCK_KEY,)) + got_lock = bool(cur.fetchone()[0]) + cur.close() + except Exception: + if lock_conn is not None: + try: + lock_conn.close() + except Exception: + pass + return + if not got_lock: try: - updated = tz_strptime(row[0], "%Y-%m-%d %H:%M:%S") - age = tz_now() - updated - if age < timedelta(minutes=30): - should_fetch = False - app.logger.info( - f"Startup: weather cache is {int(age.total_seconds()/60)} min old, skip fetch" - ) + lock_conn.close() except Exception: pass + return - if should_fetch: + try: try: - fetch_weather() - fetch_hourly() - app.logger.info("Startup: weather fetched (cache stale or empty)") - except Exception as e: - app.logger.warning(f"Startup weather fetch failed: {e}") + conn = get_db(db_path) + row = conn.execute( + "SELECT updated_at FROM weather_cache WHERE user_id=0 ORDER BY id DESC LIMIT 1" + ).fetchone() + conn.close() + except Exception: + row = None + + should_fetch = True + if row and row["updated_at"]: + try: + updated = tz_strptime(row["updated_at"], "%Y-%m-%d %H:%M:%S") + age = tz_now() - updated + if age < timedelta(minutes=30): + should_fetch = False + app.logger.info( + f"Startup: weather cache is {int(age.total_seconds()/60)} min old, skip fetch" + ) + except Exception: + pass + + if should_fetch: + try: + fetch_weather() + fetch_hourly() + app.logger.info("Startup: weather fetched (cache stale or empty)") + except Exception as e: + app.logger.warning(f"Startup weather fetch failed: {e}") + finally: + try: + lock_conn.close() # releases the advisory lock + except Exception: + pass diff --git a/backend/auth.py b/backend/auth.py index 230c06f..e8883eb 100644 --- a/backend/auth.py +++ b/backend/auth.py @@ -20,11 +20,10 @@ import string import time from flask import g, redirect, url_for, request, jsonify, current_app from backend.models import ( - get_auth_config, update_auth_config, get_temp_password, set_temp_password, - clear_temp_password, get_recovery_codes, set_recovery_codes, has_mfa_configured, - user_exists, is_admin_user, get_db, + get_auth_config, get_temp_password, + is_admin_user, get_db, create_auth_token, find_access_token, - delete_auth_token, REFRESH_TOKEN_TTL, ACCESS_TOKEN_TTL, + delete_auth_token, REFRESH_TOKEN_TTL, ) # ─── Token / cookie constants ──────────────── diff --git a/backend/calendar_mgr.py b/backend/calendar_mgr.py index defb2c1..69a54c1 100644 --- a/backend/calendar_mgr.py +++ b/backend/calendar_mgr.py @@ -4,10 +4,7 @@ Every function is scoped to an explicit `user_id` so each user only ever sees/mutates their own data (defense in depth: rows are filtered by user_id even though they also join user-scoped groups/sources). """ -import os -import hashlib -import tempfile -from datetime import datetime, timedelta, date +from datetime import datetime, timedelta from backend.time_util import now as tz_now from flask import current_app from backend.models import get_db diff --git a/backend/countdown_mgr.py b/backend/countdown_mgr.py index 95f8586..79777a6 100644 --- a/backend/countdown_mgr.py +++ b/backend/countdown_mgr.py @@ -1,6 +1,5 @@ """Countdown timer management — duration-based, once-only.""" import time -from datetime import datetime, timedelta from backend.time_util import isoformat as tz_isoformat from flask import current_app from backend.models import get_db diff --git a/backend/db_admin.py b/backend/db_admin.py new file mode 100644 index 0000000..9430e0d --- /dev/null +++ b/backend/db_admin.py @@ -0,0 +1,263 @@ +"""Database administration: export / import / format detection. + +Only Root may call these (enforced at the route layer). All operations are +atomic: an import runs inside a single transaction — any failing row rolls the +whole import back, leaving the current database untouched. + +Formats: + - ``sqlite`` — a legacy SQLite database file (magic bytes ``SQLite format 3``). + Tables not present in the current SCHEMA are dropped (e.g. camera_config); + the legacy single-user `auth_config` row (keyed by id=1) is attributed to + Root (user 0); missing columns are filled with their SCHEMA DEFAULT. + - ``json`` — the self-contained backup produced by :func:`export_database` + (``{"format": "iot-dboard-backup", ...}``). +""" +import json +import os +import sqlite3 +import tempfile +from datetime import datetime, timezone + +from backend.models import ( + get_db, SCHEMA, DB_VERSION, _CREATE_TABLE_RE, _table_columns, + _read_db_version, _ensure_defaults, _fix_identity_sequences, +) + +# Tables whose contents are exported/imported (everything except the schema +# version row and transient session tokens). +_EXCLUDE_TABLES = {"meta", "auth_tokens"} + + +def _schema_table_names() -> list: + return [m.group(1) for m in _CREATE_TABLE_RE.finditer(SCHEMA)] + + +def detect_format(data: bytes) -> str: + """Detect the database file format: 'sqlite', 'json' or 'unknown'.""" + if data[:16] == b"SQLite format 3\x00": + return "sqlite" + try: + obj = json.loads(data.decode("utf-8")) + if isinstance(obj, dict) and obj.get("format") == "iot-dboard-backup": + return "json" + except Exception: + pass + return "unknown" + + +def export_database(db_path: str) -> dict: + """Snapshot every data table (except auth_tokens) into a JSON-serializable dict.""" + conn = get_db(db_path) + try: + tables = _schema_table_names() + out = { + "format": "iot-dboard-backup", + "backup_version": 1, + "db_version": _read_db_version(conn) or DB_VERSION, + "exported_at": datetime.now(timezone.utc).isoformat(), + "tables": {}, + } + for t in tables: + if t in _EXCLUDE_TABLES: + continue + cols = _table_columns(conn, t) + if not cols: + continue + rows = conn.execute( + f"SELECT {', '.join(cols)} FROM {t} ORDER BY 1" + ).fetchall() + out["tables"][t] = { + "columns": cols, + "rows": [[r[c] for c in cols] for r in rows], + } + return out + finally: + conn.close() + + +def export_database_json(db_path: str) -> str: + """Export as a compact JSON string (files are small; no pretty printing).""" + return json.dumps(export_database(db_path), ensure_ascii=False, default=str) + + +# ─── SQLite import ─────────────────────────── + +def _sqlite_tables(conn_old) -> list: + return [ + r["name"] for r in conn_old.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'" + ).fetchall() + ] + + +def _sqlite_rows(conn_old, table: str, cols: list) -> list: + if not cols: + return [] + col_str = ", ".join(cols) + return conn_old.execute(f"SELECT {col_str} FROM {table}").fetchall() + + +def _read_sqlite(data: bytes) -> dict: + """Read a legacy SQLite file into {table: {"columns": [...], "rows": [...]}}. + auth_config special case: legacy single-user row (id=1) belongs to Root.""" + tmp_path = None + try: + fd, tmp_path = tempfile.mkstemp(suffix=".db") + with os.fdopen(fd, "wb") as f: + f.write(data) + conn_old = sqlite3.connect(f"file:{tmp_path}?mode=ro", uri=True) + conn_old.row_factory = sqlite3.Row + try: + tables = {} + for t in _sqlite_tables(conn_old): + cols = [r["name"] for r in conn_old.execute(f"PRAGMA table_info({t})").fetchall()] + if not cols: + continue + rows = [] + for row in _sqlite_rows(conn_old, t, cols): + rows.append([row[c] for c in cols]) + tables[t] = {"columns": cols, "rows": rows} + return tables + finally: + conn_old.close() + finally: + if tmp_path and os.path.exists(tmp_path): + os.unlink(tmp_path) + + +def _import_table(conn, table: str, data: dict, where: str, skipped: list) -> int: + """Insert rows of one table into the target PG database (strict mode). + Returns the number of rows inserted.""" + src_cols = data.get("columns") or [] + rows = data.get("rows") or [] + if not src_cols or not rows: + return 0 + dst_cols = _table_columns(conn, table) + if not dst_cols: + skipped.append(table) # table not in schema — dropped (e.g. camera_config) + return 0 + + if table == "auth_config": + return _import_auth_config(conn, data, where) + + # Only common columns are copied; missing target columns get their DEFAULT. + common = [c for c in dst_cols if c in src_cols] + if not common: + skipped.append(table) + return 0 + col_str = ", ".join(common) + placeholders = ", ".join("?" * len(common)) + sql = f"INSERT INTO {table} ({col_str}) VALUES ({placeholders})" + n = 0 + for i, row in enumerate(rows): + vals = [] + for c in common: + vals.append(row[src_cols.index(c)]) + try: + conn.execute(sql, tuple(vals)) + n += 1 + except Exception as e: + raise ValueError( + f"{where}: 表 {table} 第 {i + 1} 行导入失败: {e}" + ) from e + return n + + +def _import_auth_config(conn, data: dict, where: str) -> int: + """Special case: the legacy auth_config table is keyed by `id` (single user, + id=1); the new table is keyed by `user_id`. The legacy row belongs to Root.""" + src_cols = data.get("columns") or [] + rows = data.get("rows") or [] + if not src_cols or not rows: + return 0 + dst_cols = _table_columns(conn, "auth_config") + n = 0 + for i, row in enumerate(rows): + d = {c: row[src_cols.index(c)] for c in src_cols} + if "user_id" in src_cols: + uid = d.get("user_id") or 0 + else: + old_id = d.get("id", 1) + uid = 0 if old_id == 1 else (old_id or 0) + common = [c for c in dst_cols if c in src_cols and c not in ("id", "user_id")] + if not common: + continue + col_str = ", ".join(common) + placeholders = ", ".join("?" * len(common)) + sql = f"INSERT OR IGNORE INTO auth_config (user_id, {col_str}) VALUES (?, {placeholders})" + try: + conn.execute(sql, (uid, *[d[c] for c in common])) + n += 1 + except Exception as e: + raise ValueError( + f"{where}: auth_config 第 {i + 1} 行导入失败: {e}" + ) from e + return n + + +# ─── Restore core (shared by sqlite + json imports) ───── + +def _restore(db_path: str, tables: dict, where: str) -> dict: + """Atomic restore: TRUNCATE every data table, insert the given rows, then + re-bootstrap defaults and stamp the schema version. Any failure rolls back + the whole import (the shim rolls back on error, keeping it atomic).""" + conn = get_db(db_path) + stats: dict[str, int] = {} + skipped: list = [] + try: + # 1. Wipe current data (TRUNCATE is transactional in PostgreSQL). + all_tables = _schema_table_names() + truncate_list = [t for t in all_tables if t != "meta"] + if truncate_list: + conn.execute("TRUNCATE TABLE " + ", ".join(truncate_list) + " CASCADE") + + # 2. Insert data (strict: first failing row aborts everything). + for t in all_tables: + if t in _EXCLUDE_TABLES: + continue + if t in tables: + stats[t] = _import_table(conn, t, tables[t], where, skipped) + + # 3. Re-bootstrap defaults (Root user, config rows) + stamp version. + _ensure_defaults(conn) + conn.execute("UPDATE meta SET version=? WHERE id=1", (DB_VERSION,)) + conn.execute("INSERT OR IGNORE INTO meta (id, version) VALUES (1, ?)", (DB_VERSION,)) + _fix_identity_sequences(conn) + conn.commit() + if skipped: + stats["_skipped_tables"] = skipped + return stats + except Exception: + try: + conn.rollback() + except Exception: + pass + raise + finally: + conn.close() + + +def import_sqlite(db_path: str, data: bytes) -> dict: + """Import a legacy SQLite database file (atomic).""" + tables = _read_sqlite(data) + return _restore(db_path, tables, "SQLite 导入") + + +def import_json(db_path: str, data: dict) -> dict: + """Import a JSON backup produced by export_database (atomic).""" + if not isinstance(data, dict) or data.get("format") != "iot-dboard-backup": + raise ValueError("不是有效的备份文件（缺少 iot-dboard-backup 标记）") + tables = data.get("tables") or {} + if not isinstance(tables, dict): + raise ValueError("备份文件结构无效：tables 缺失") + return _restore(db_path, tables, "备份导入") + + +def import_blob(db_path: str, data: bytes) -> dict: + """Import a file, auto-detecting sqlite vs json backup.""" + fmt = detect_format(data) + if fmt == "sqlite": + return import_sqlite(db_path, data) + if fmt == "json": + return import_json(db_path, json.loads(data.decode("utf-8"))) + raise ValueError("无法识别的数据库文件格式（仅支持 SQLite 数据库或本系统导出的备份 JSON）") diff --git a/backend/events.py b/backend/events.py index 5083061..15154cd 100644 --- a/backend/events.py +++ b/backend/events.py @@ -44,6 +44,3 @@ class EventBus: for item in dead: self._subscribers.remove(item) - def broadcast(self): - """Send heartbeat to all subscribers.""" - self.publish("heartbeat", {"ts": __import__("time").time()}) diff --git a/backend/models.py b/backend/models.py index ba1906c..e025ba1 100644 --- a/backend/models.py +++ b/backend/models.py @@ -1,23 +1,31 @@ -"""SQLite database models, initialization, and user-scoped helpers. +"""PostgreSQL database models, initialization, and user-scoped helpers. -Multi-user design: +Multi-user design (unchanged from the SQLite era): - The `users` table stores every account. Root (ID 0) is the administrator and is created automatically on first boot (fresh DB) or when migrating an existing single-user database (all pre-existing data becomes Root's data). - Every per-user table carries a `user_id INTEGER NOT NULL DEFAULT 0` column, - so legacy single-user rows are automatically attributed to Root on migration + so legacy single-user rows are automatically attributed to Root on import and any NEW column added in the future gets a DEFAULT (the migration fills missing columns with their SCHEMA defaults). - `auth_config` is keyed by `user_id` (legacy `id=1` row maps to user 0). + +The data-access layer talks PostgreSQL through ``backend.pg_compat``, a thin +sqlite3-compatible shim (``?`` placeholders, ``INSERT OR IGNORE``, +``datetime('now')``, ``cursor.lastrowid``, ``conn.total_changes`` are all +translated transparently). """ -import sqlite3 -import os +import re import sys import time +from backend.pg_compat import connect as _pg_connect +# ─── PostgreSQL schema ─────────────────────── +# All new columns must carry a DEFAULT (migration contract: missing columns are +# filled with their SCHEMA default when the DB is upgraded). SCHEMA = """ --- Global schema version (single row). All new columns must carry a DEFAULT. +-- Global schema version (single row). CREATE TABLE IF NOT EXISTS meta ( id INTEGER PRIMARY KEY DEFAULT 1, version INTEGER NOT NULL DEFAULT 1 @@ -26,7 +34,7 @@ CREATE TABLE IF NOT EXISTS meta ( -- Auth tokens (dual-token auth): access (short-lived, per-request) and -- refresh (long-lived, stored in an httpOnly cookie). Only hashes are stored. CREATE TABLE IF NOT EXISTS auth_tokens ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, token_hash TEXT NOT NULL, kind TEXT NOT NULL DEFAULT 'access' CHECK(kind IN ('access','refresh')), @@ -34,14 +42,14 @@ CREATE TABLE IF NOT EXISTS auth_tokens ( temp INTEGER NOT NULL DEFAULT 0, session_version INTEGER NOT NULL DEFAULT 0, reauth_at INTEGER NOT NULL DEFAULT 0, - created_at TEXT DEFAULT (datetime('now')), + created_at TEXT DEFAULT (to_char(now(), 'YYYY-MM-DD HH24:MI:SS')), expires_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_auth_tokens_hash ON auth_tokens(token_hash); CREATE INDEX IF NOT EXISTS idx_auth_tokens_user ON auth_tokens(user_id); CREATE TABLE IF NOT EXISTS users ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, remark TEXT NOT NULL DEFAULT '', role TEXT NOT NULL DEFAULT 'user' CHECK(role IN ('admin','user')), is_temp INTEGER NOT NULL DEFAULT 0, @@ -49,7 +57,7 @@ CREATE TABLE IF NOT EXISTS users ( status TEXT NOT NULL DEFAULT 'active' CHECK(status IN ('active','pending')), session_version INTEGER NOT NULL DEFAULT 0, theme TEXT NOT NULL DEFAULT 'dark' CHECK(theme IN ('dark','light','auto')), - created_at TEXT DEFAULT (datetime('now')) + created_at TEXT DEFAULT (to_char(now(), 'YYYY-MM-DD HH24:MI:SS')) ); CREATE TABLE IF NOT EXISTS auth_config ( @@ -64,16 +72,16 @@ CREATE TABLE IF NOT EXISTS auth_config ( ); CREATE TABLE IF NOT EXISTS calendar_groups ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, name TEXT NOT NULL, color TEXT DEFAULT '#4A90D9', enabled INTEGER DEFAULT 1, - created_at TEXT DEFAULT (datetime('now')) + created_at TEXT DEFAULT (to_char(now(), 'YYYY-MM-DD HH24:MI:SS')) ); CREATE TABLE IF NOT EXISTS calendar_sources ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, group_id INTEGER NOT NULL, type TEXT NOT NULL CHECK(type IN ('upload', 'subscription', 'manual')), @@ -85,7 +93,7 @@ CREATE TABLE IF NOT EXISTS calendar_sources ( ); CREATE TABLE IF NOT EXISTS calendar_events ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, source_id INTEGER NOT NULL, uid TEXT, @@ -99,7 +107,7 @@ CREATE TABLE IF NOT EXISTS calendar_events ( ); CREATE TABLE IF NOT EXISTS reminders ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, calendar_group_id INTEGER NOT NULL, minutes_before INTEGER DEFAULT 15, @@ -108,7 +116,7 @@ CREATE TABLE IF NOT EXISTS reminders ( ); CREATE TABLE IF NOT EXISTS weather_config ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, api_key TEXT DEFAULT '', city TEXT DEFAULT 'Beijing', @@ -124,35 +132,35 @@ CREATE TABLE IF NOT EXISTS weather_config ( ); CREATE TABLE IF NOT EXISTS weather_cache ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, data TEXT, - updated_at TEXT DEFAULT (datetime('now')) + updated_at TEXT DEFAULT (to_char(now(), 'YYYY-MM-DD HH24:MI:SS')) ); CREATE TABLE IF NOT EXISTS weather_hourly_cache ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, data TEXT, - fetched_at TEXT DEFAULT (datetime('now')) + fetched_at TEXT DEFAULT (to_char(now(), 'YYYY-MM-DD HH24:MI:SS')) ); CREATE TABLE IF NOT EXISTS weather_minutely_cache ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, data TEXT, fetched_at TEXT ); CREATE TABLE IF NOT EXISTS weather_warning_cache ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, data TEXT, fetched_at TEXT ); CREATE TABLE IF NOT EXISTS clock_config ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, city TEXT DEFAULT 'Beijing', timezone TEXT DEFAULT 'Asia/Shanghai', @@ -160,7 +168,7 @@ CREATE TABLE IF NOT EXISTS clock_config ( ); CREATE TABLE IF NOT EXISTS alarms ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, label TEXT DEFAULT '', time TEXT NOT NULL, @@ -171,7 +179,7 @@ CREATE TABLE IF NOT EXISTS alarms ( ); CREATE TABLE IF NOT EXISTS countdowns ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, label TEXT DEFAULT '', duration_seconds INTEGER DEFAULT 300, @@ -181,7 +189,7 @@ CREATE TABLE IF NOT EXISTS countdowns ( ); CREATE TABLE IF NOT EXISTS display_config ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, mode TEXT NOT NULL DEFAULT 'default' CHECK(mode IN ('default', 'custom')), custom_html TEXT DEFAULT '', @@ -191,7 +199,7 @@ CREATE TABLE IF NOT EXISTS display_config ( ); CREATE TABLE IF NOT EXISTS pomodoro_state ( - id INTEGER PRIMARY KEY AUTOINCREMENT, + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, user_id INTEGER NOT NULL DEFAULT 0, running INTEGER DEFAULT 0, phase TEXT DEFAULT 'work' CHECK(phase IN ('work', 'break')), @@ -213,26 +221,27 @@ CREATE UNIQUE INDEX IF NOT EXISTS idx_pomodoro_state_user ON pomodoro_state(user """ -# Tables whose rows are completely per-user (used for cascade delete). -USER_DATA_TABLES = [ - "auth_config", "calendar_groups", "calendar_sources", "calendar_events", - "reminders", "weather_config", "weather_cache", "weather_hourly_cache", - "weather_minutely_cache", "weather_warning_cache", "clock_config", - "alarms", "countdowns", "display_config", "pomodoro_state", +# Ordered list of all user-data tables (children first) for FK-safe deletion. +_DELETE_ORDER = [ + "calendar_events", "calendar_sources", "reminders", "calendar_groups", + "weather_cache", "weather_hourly_cache", "weather_minutely_cache", + "weather_warning_cache", "weather_config", "clock_config", + "alarms", "countdowns", "display_config", "pomodoro_state", "auth_config", "auth_tokens", ] -def get_db(db_path: str = "data.db") -> sqlite3.Connection: - conn = sqlite3.connect(db_path) - conn.row_factory = sqlite3.Row - conn.execute("PRAGMA journal_mode=WAL") - conn.execute("PRAGMA foreign_keys=ON") - return conn +def get_db(db_path: str = "data.db"): + """Open a PostgreSQL connection (sqlite3-compatible wrapper). + + `db_path` is a PostgreSQL connection string (kept as the historical + parameter name to minimise churn across the codebase). + """ + return _pg_connect(db_path) def _db_path(preferred: str = None) -> str: - """Get the actual database path, preferring current_app config if available.""" + """Get the actual database connection string, preferring current_app config.""" if preferred: return preferred try: @@ -330,16 +339,8 @@ def delete_user(user_id: int, db_path: str = None): if user_id == 0: return conn = get_db(_db_path(db_path)) - # Children first so FK constraints never reject a delete. - ordered = [ - "calendar_events", "calendar_sources", "reminders", "calendar_groups", - "weather_cache", "weather_hourly_cache", "weather_minutely_cache", - "weather_warning_cache", "weather_config", "clock_config", - "alarms", "countdowns", "display_config", "pomodoro_state", "auth_config", - "auth_tokens", - ] try: - for table in ordered: + for table in _DELETE_ORDER: conn.execute(f"DELETE FROM {table} WHERE user_id=?", (user_id,)) conn.execute("DELETE FROM users WHERE id=?", (user_id,)) conn.commit() @@ -826,6 +827,14 @@ def find_user_by_credential_id(credential_id: str, db_path: str = None) -> int | # ─── DB init / migration ───────────────────── +# Stores passwords that were just generated, for the startup script to print. +_EMITTED_PASSWORDS: list = [] + +# Current schema version supported by this program. Bump it whenever the +# SCHEMA changes so existing databases get upgraded on next start. +DB_VERSION = 2 + + def _ensure_defaults(conn): """Insert default rows for config tables and the Root user when missing. Generates a random 6-digit temp password when Root's auth_config is created @@ -843,6 +852,11 @@ def _ensure_defaults(conn): conn.execute( "INSERT OR IGNORE INTO users (id, remark, role, status) VALUES (0, 'Root', 'admin', 'active')" ) + # Keep the users identity sequence ahead of explicit id=0 insertion. + conn.execute( + "SELECT setval(pg_get_serial_sequence('users', 'id'), " + "GREATEST((SELECT COALESCE(MAX(id), 1) FROM users), 1))" + ) # weather_config for Root count = conn.execute("SELECT COUNT(*) FROM weather_config WHERE user_id=0").fetchone()[0] @@ -895,165 +909,224 @@ def _ensure_defaults(conn): conn.commit() -# Stores passwords that were just generated, for the startup script to print. -_EMITTED_PASSWORDS: list = [] - -# Current schema version supported by this program. Bump it whenever the -# SCHEMA changes so existing databases get rebuilt+migrated on next start. -DB_VERSION = 1 - - -def _table_columns(conn, table_name): - return [row["name"] for row in conn.execute(f"PRAGMA table_info({table_name})")] - - -def _migrate_auth_config(conn_new, conn_old): - """Special-case migration for auth_config: the legacy table is keyed by - `id` (single user, id=1), the new table is keyed by `user_id`. - The legacy single-user row belongs to Root (user 0). - """ - old_cols = _table_columns(conn_old, "auth_config") - new_cols = _table_columns(conn_new, "auth_config") - if not old_cols: - return - rows = conn_old.execute("SELECT * FROM auth_config").fetchall() - for row in rows: - d = dict(row) - if "user_id" in old_cols: - uid = d.get("user_id") or 0 - else: - # Legacy single-user DB: the id=1 row belongs to Root (user 0). - old_id = d.get("id", 1) - uid = 0 if old_id == 1 else (old_id or 0) - common = [c for c in new_cols if c in old_cols and c not in ("id", "user_id")] - if not common: - continue - col_str = ", ".join(common) - placeholders = ", ".join("?" * len(common)) - conn_new.execute( - f"INSERT OR IGNORE INTO auth_config (user_id, {col_str}) VALUES (?, {placeholders})", - (uid, *[d[c] for c in common]) - ) - - -def init_db(db_path: str = "data.db"): - """Initialize the database with schema-version aware migration. - - - No database file: create it from the latest SCHEMA and create Root. - - Existing database: - * no `meta.version` (legacy DB) → rebuild + migrate compatible data - * DB version < DB_VERSION → rebuild + migrate - * DB version == DB_VERSION → use as-is (fast path) - * DB version > DB_VERSION → abort with an error (DB is newer - than this program) - - Rebuild copies compatible data (legacy single-user data becomes Root's) - and fills missing columns with SCHEMA defaults. - """ - if not os.path.exists(db_path): - conn = get_db(db_path) - conn.executescript(SCHEMA) - _ensure_defaults(conn) - conn.execute("UPDATE meta SET version=? WHERE id=1", (DB_VERSION,)) - conn.commit() - conn.close() - return - - conn_probe = get_db(db_path) - old_version = _read_db_version(conn_probe) - conn_probe.close() - - if old_version is not None and old_version > DB_VERSION: - raise SystemExit( - f"[FATAL] 数据库 schema 版本 (v{old_version}) 高于当前程序支持的版本 " - f"(v{DB_VERSION})。请先升级程序，或删除/降级数据库后再启动。" - ) - - if old_version is not None and old_version == DB_VERSION: - # Same version — no rebuild needed. Just ensure defaults exist. - conn = get_db(db_path) - _ensure_defaults(conn) - conn.execute("INSERT OR IGNORE INTO meta (id, version) VALUES (1, ?)", (DB_VERSION,)) - conn.commit() - conn.close() - return - - # Migrate by rebuilding: create a fresh DB with the latest schema, - # copy compatible data from the old DB, then replace the old DB. - new_db_path = db_path + ".new" - backup_path = db_path + ".backup" - - # Clean up any stale temporary database from a previous failed run. - if os.path.exists(new_db_path): - os.remove(new_db_path) - - conn_new = get_db(new_db_path) - conn_new.executescript(SCHEMA) - - conn_old = get_db(db_path) - # Disable FK enforcement while copying; rows are copied table by table. - conn_old.execute("PRAGMA foreign_keys=OFF") - conn_new.execute("PRAGMA foreign_keys=OFF") - - tables = conn_old.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'" +def _table_columns(conn, table_name: str) -> list: + """Return the column names of a table (information_schema).""" + rows = conn.execute( + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema = current_schema() AND table_name = ? " + "ORDER BY ordinal_position", + (table_name,) ).fetchall() - - for (table_name,) in tables: - # Tables that no longer exist in the new schema are dropped (e.g. camera_config). - exists_new = conn_new.execute( - "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", - (table_name,) - ).fetchone() - if not exists_new: - continue - - if table_name == "auth_config": - _migrate_auth_config(conn_new, conn_old) - continue - - old_cols = _table_columns(conn_old, table_name) - new_cols = _table_columns(conn_new, table_name) - common_cols = [c for c in new_cols if c in old_cols] - if not common_cols: - continue - - col_str = ", ".join(common_cols) - placeholders = ", ".join("?" * len(common_cols)) - sql = f"INSERT OR IGNORE INTO {table_name} ({col_str}) VALUES ({placeholders})" - - for row in conn_old.execute(f"SELECT {col_str} FROM {table_name}"): - try: - conn_new.execute(sql, tuple(row)) - except Exception: - # Ignore rows that cannot be migrated (e.g. constraint violations). - pass - - conn_new.commit() - - # Now fill defaults for any tables that are still empty + Root bootstrap. - _ensure_defaults(conn_new) - - # Stamp the migrated database with the program's schema version. - conn_new.execute("UPDATE meta SET version=? WHERE id=1", (DB_VERSION,)) - conn_new.commit() - - conn_old.close() - conn_new.close() - - # Replace old DB with the new one, keeping a backup. - if os.path.exists(backup_path): - os.remove(backup_path) - os.rename(db_path, backup_path) - os.rename(new_db_path, db_path) + return [r["column_name"] for r in rows] def _read_db_version(conn) -> int | None: """Read the schema version from the `meta` table. - Returns None when the DB is legacy (no meta table / no version column). + Returns None when the DB is legacy (no meta table / no version row). """ try: row = conn.execute("SELECT version FROM meta WHERE id=1").fetchone() return int(row["version"]) if row else None except Exception: + conn._safe_rollback() return None + + +# ─── SCHEMA parsing for column-level migration ────── +_CREATE_TABLE_RE = re.compile( + r"CREATE TABLE IF NOT EXISTS\s+(\w+)\s*\((.*?)\)\s*;", + re.IGNORECASE | re.DOTALL, +) + + +def _split_top_level(text: str) -> list: + """Split `text` on top-level commas (ignoring commas inside parentheses + AND inside single-quoted strings, e.g. DEFAULT '1,2,3,4,5,6,7').""" + parts = [] + depth = 0 + cur = [] + in_str = False + i = 0 + while i < len(text): + ch = text[i] + if ch == "'": + # SQL 单引号转义用两个单引号表示 + if in_str and i + 1 < len(text) and text[i + 1] == "'": + cur.append("''") + i += 2 + continue + in_str = not in_str + cur.append(ch) + elif not in_str: + if ch == "(": + depth += 1 + elif ch == ")": + depth -= 1 + if ch == "," and depth == 0: + parts.append("".join(cur)) + cur = [] + i += 1 + continue + cur.append(ch) + else: + cur.append(ch) + i += 1 + if cur: + parts.append("".join(cur)) + return parts + + +def _strip_check(s: str) -> str: + """Remove every CHECK(...) clause (parenthesis-balanced).""" + out = s + while True: + idx = out.upper().find("CHECK(") + if idx == -1: + break + depth = 0 + end = -1 + for j in range(idx, len(out)): + if out[j] == "(": + depth += 1 + elif out[j] == ")": + depth -= 1 + if depth == 0: + end = j + break + if end == -1: + break + out = out[:idx] + out[end + 1:] + return out + + +def _migrate_missing_columns(conn): + """Add columns that exist in SCHEMA but are missing from existing tables. + New columns are created with their SCHEMA DEFAULT (migration contract). + Best-effort: a failing column does not block startup. + """ + for m in _CREATE_TABLE_RE.finditer(SCHEMA): + table = m.group(1) + try: + existing = set(_table_columns(conn, table)) + except Exception: + conn._safe_rollback() + continue + for col_def in _split_top_level(m.group(2)): + col_def = col_def.strip() + if not col_def: + continue + upper = col_def.upper() + if upper.startswith(("PRIMARY KEY", "UNIQUE", "CHECK", "FOREIGN KEY", "CONSTRAINT")): + continue + name = col_def.split()[0].strip('"') + if not name or name in existing: + continue + # Make the ADD COLUMN safe for rows that already exist: + # drop NOT NULL / PK / UNIQUE / IDENTITY; keep TYPE + DEFAULT. + safe = re.sub(r"\s+NOT\s+NULL\b", "", col_def, flags=re.IGNORECASE) + safe = re.sub(r"\s+PRIMARY\s+KEY\b", "", safe, flags=re.IGNORECASE) + safe = re.sub(r"\s+UNIQUE\b", "", safe, flags=re.IGNORECASE) + safe = re.sub(r"\s+GENERATED\s+BY\s+DEFAULT\s+AS\s+IDENTITY\b", "", safe, flags=re.IGNORECASE) + safe = _strip_check(safe).strip().rstrip(",") + try: + conn.execute(f"ALTER TABLE {table} ADD COLUMN IF NOT EXISTS {safe}") + except Exception: + # rollback done by the shim; the next column starts a fresh tx + pass + + +def _fix_identity_sequences(conn): + """Sync every IDENTITY sequence to max(id) (safe after explicit id inserts).""" + try: + rows = conn.execute( + "SELECT table_name, column_name FROM information_schema.columns " + "WHERE table_schema = current_schema() AND is_identity = 'YES'" + ).fetchall() + except Exception: + conn._safe_rollback() + return + for r in rows: + t, c = r["table_name"], r["column_name"] + try: + conn.execute( + f"SELECT setval(pg_get_serial_sequence('{t}', '{c}'), " + f"GREATEST((SELECT COALESCE(MAX({c}), 1) FROM {t}), 1))" + ) + except Exception: + conn._safe_rollback() + + +# Timestamp columns whose DEFAULT was generated by early builds with the wrong +# to_char pattern ('YYYY-MM-DD HH:MM:SS' — HH is 12-hour, MM is the month). +_TIMESTAMP_DEFAULT_FIX = [ + ("auth_tokens", "created_at"), + ("users", "created_at"), + ("calendar_groups", "created_at"), + ("weather_cache", "updated_at"), + ("weather_hourly_cache", "fetched_at"), +] + + +def _fix_timestamp_defaults(conn): + """Repair legacy DEFAULT expressions (idempotent, best-effort).""" + for t, c in _TIMESTAMP_DEFAULT_FIX: + try: + conn.execute( + f"ALTER TABLE {t} ALTER COLUMN {c} " + "SET DEFAULT to_char(now(), 'YYYY-MM-DD HH24:MI:SS')" + ) + except Exception: + conn._safe_rollback() + + +def init_db(db_path: str = "data.db"): + """Initialize the database with schema-version aware migration (PostgreSQL). + + - Fresh database (no `meta` row) -> create the latest SCHEMA + defaults. + - Existing database: + * DB version == DB_VERSION -> use as-is (fast path, defaults ensured) + * no version / DB version < DB_VERSION + -> create missing tables/columns; missing + columns are filled with their SCHEMA + DEFAULT; stamp the new version + * DB version > DB_VERSION -> abort with an error (DB is newer than + this program) + """ + conn = get_db(db_path) + try: + old_version = _read_db_version(conn) + + if old_version is not None and old_version > DB_VERSION: + raise SystemExit( + f"[FATAL] 数据库 schema 版本 (v{old_version}) 高于当前程序支持的版本 " + f"(v{DB_VERSION})。请先升级程序，或删除/降级数据库后再启动。" + ) + + if old_version is not None and old_version == DB_VERSION: + # Same version — no upgrade needed. Just ensure defaults exist. + _ensure_defaults(conn) + _fix_timestamp_defaults(conn) + conn.commit() + return + + # Create any missing tables (CREATE TABLE IF NOT EXISTS is idempotent), + # then add any columns missing from older databases. The DDL is + # committed first so a later best-effort column migration (which may + # roll back its own transaction) can never undo the schema creation. + conn.executescript(SCHEMA) + conn.commit() + _migrate_missing_columns(conn) + + # Bootstrap defaults (Root user, meta row, config rows). + _ensure_defaults(conn) + + # Repair legacy timestamp DEFAULT expressions (old to_char HH:MM pattern). + _fix_timestamp_defaults(conn) + + # Stamp the database with the program's schema version. + conn.execute("UPDATE meta SET version=? WHERE id=1", (DB_VERSION,)) + conn.execute("INSERT OR IGNORE INTO meta (id, version) VALUES (1, ?)", (DB_VERSION,)) + _fix_identity_sequences(conn) + conn.commit() + finally: + conn.close() diff --git a/backend/pg_compat.py b/backend/pg_compat.py new file mode 100644 index 0000000..35770ea --- /dev/null +++ b/backend/pg_compat.py @@ -0,0 +1,252 @@ +"""PostgreSQL compatibility layer for the legacy sqlite3-style data access. + +The rest of the codebase talks to the database through hand-written SQL that +was originally written for sqlite3 (``?`` placeholders, ``datetime('now')``, +``INSERT OR IGNORE``, ``cursor.lastrowid``, ``conn.total_changes`` ...). +This module exposes a sqlite3-like connection/cursor interface on top of +psycopg2 and transparently translates the dialect differences, so the +business-layer SQL keeps working with minimal changes. + +Translations applied inside :meth:`PgCursor.execute`: + - ``?`` placeholders -> psycopg2 ``%s`` placeholders + - ``datetime('now')`` -> ``to_char(now(), 'YYYY-MM-DD HH24:MI:SS')`` + (UTC, byte-identical format to before) + - ``INSERT OR IGNORE INTO`` -> ``INSERT INTO ... ON CONFLICT DO NOTHING`` + - ``BEGIN IMMEDIATE`` -> ``BEGIN`` + - ``INSERT`` (without RETURNING) -> appends ``RETURNING `` so that + ``cursor.lastrowid`` keeps working; the + primary key is looked up once from + information_schema and cached. + +Rows are returned as a dict subclass that supports both ``row["col"]`` and +``row[0]`` (positional) access to match the old ``sqlite3.Row`` behaviour. + +The connection timezone is pinned to UTC so ``now()`` output matches the +previous ``datetime('now')`` (UTC) semantics. +""" +import re +import threading + +import psycopg2 +import psycopg2.extras + +_DATETIME_NOW_RE = re.compile(r"datetime\('now'\)", re.IGNORECASE) +_INSERT_OR_IGNORE_RE = re.compile(r"INSERT\s+OR\s+IGNORE\s+INTO", re.IGNORECASE) +_INSERT_TABLE_RE = re.compile( + r"^\s*INSERT\s+(?:OR\s+IGNORE\s+)?INTO\s+([\"'`\w]+)", re.IGNORECASE +) +_BEGIN_IMMEDIATE_RE = re.compile(r"^\s*BEGIN\s+IMMEDIATE", re.IGNORECASE) + +_PK_CACHE: dict[str, str | None] = {} +_PK_LOCK = threading.Lock() + + +class _Row(dict): + """dict-like row supporting both column-name and positional access.""" + + __slots__ = ("_cols",) + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._cols = None + + def __getitem__(self, key): + if isinstance(key, int): + if self._cols is None or key < 0 or key >= len(self._cols): + raise IndexError(key) + return dict.__getitem__(self, self._cols[key]) + return dict.__getitem__(self, key) + + def get_by_index(self, i): + return self[i] + + +def _wrap_row(row): + """Convert a psycopg2 result row into a `_Row` (positional access enabled).""" + if row is None or isinstance(row, _Row): + return row + if isinstance(row, dict): + r = _Row(row) + r._cols = list(row.keys()) + return r + return row + + +def _primary_key(conn, table: str) -> str | None: + """Return the primary-key column of `table` (cached per process).""" + with _PK_LOCK: + if table in _PK_CACHE: + return _PK_CACHE[table] + pk: str | None = None + try: + cur = conn.cursor() + cur.execute( + """ + SELECT kcu.column_name + FROM information_schema.table_constraints tc + JOIN information_schema.key_column_usage kcu + ON tc.constraint_name = kcu.constraint_name + AND tc.table_schema = kcu.table_schema + WHERE tc.constraint_type = 'PRIMARY KEY' + AND tc.table_name = %s + AND tc.table_schema = current_schema() + ORDER BY kcu.ordinal_position + LIMIT 1 + """, + (table,), + ) + row = cur.fetchone() + pk = row["column_name"] if row else None + cur.close() + except Exception: + pk = None + with _PK_LOCK: + _PK_CACHE[table] = pk + return pk + + +def _translate_sql(sql: str) -> tuple[str, bool]: + """Return (translated_sql, has_insert_or_ignore).""" + sql = _DATETIME_NOW_RE.sub("to_char(now(), 'YYYY-MM-DD HH24:MI:SS')", sql) + sql = _BEGIN_IMMEDIATE_RE.sub("BEGIN", sql) + has_ignore = bool(_INSERT_OR_IGNORE_RE.search(sql)) + if has_ignore: + sql = _INSERT_OR_IGNORE_RE.sub("INSERT INTO", sql) + # `?` -> `%s`. The codebase SQL contains no literal `?` inside strings. + sql = sql.replace("?", "%s") + return sql, has_ignore + + +class PgCursor: + """A psycopg2 cursor wrapper exposing the sqlite3 cursor surface.""" + + def __init__(self, proxy: "PgConnection", cursor): + self._proxy = proxy + self._cursor = cursor + self._lastrowid = None + + def execute(self, sql: str, params=None) -> "PgCursor": + sql, has_ignore = _translate_sql(sql) + conn = self._proxy._conn + + m = _INSERT_TABLE_RE.match(sql) + table = m.group(1).strip('"`') if m else None + returning_pk = None + if table is not None and " RETURNING " not in sql.upper(): + returning_pk = _primary_key(conn, table) + + sql = sql.rstrip().rstrip(";") + if has_ignore: + sql += " ON CONFLICT DO NOTHING" + if returning_pk: + sql += f" RETURNING {returning_pk}" + + try: + self._cursor.execute(sql, params if params is not None else ()) + if returning_pk: + row = _wrap_row(self._cursor.fetchone()) + self._lastrowid = row[0] if row is not None else None + rc = self._cursor.rowcount + if rc and rc > 0: + self._proxy._total_changes += rc + except Exception: + self._proxy._safe_rollback() + raise + return self + + # ── sqlite3 cursor surface ─────────────────────────────── + @property + def lastrowid(self): + return self._lastrowid + + @property + def rowcount(self): + return self._cursor.rowcount + + def fetchone(self): + return _wrap_row(self._cursor.fetchone()) + + def fetchall(self): + return [_wrap_row(r) for r in self._cursor.fetchall()] + + def fetchmany(self, size=None): + return [_wrap_row(r) for r in self._cursor.fetchmany(size)] + + def close(self): + try: + self._cursor.close() + except Exception: + pass + + +class PgConnection: + """A sqlite3-like connection backed by psycopg2. + + Note: unlike a raw psycopg2 connection, a failed statement automatically + rolls the connection back (sqlite allows continuing after an error; this + keeps the connection usable under the same call pattern). + """ + + def __init__(self, dsn: str): + self._conn = psycopg2.connect(dsn, cursor_factory=psycopg2.extras.RealDictCursor) + self._conn.autocommit = False + self._total_changes = 0 + self._closed = False + cur = self._conn.cursor() + cur.execute("SET TIME ZONE 'UTC'") + cur.close() + + def execute(self, sql: str, params=None) -> PgCursor: + return self.cursor().execute(sql, params) + + def cursor(self) -> PgCursor: + return PgCursor(self, self._conn.cursor()) + + def executescript(self, sql: str): + """Run a multi-statement DDL script (SCHEMA). No translation.""" + cur = self._conn.cursor() + try: + cur.execute(sql) + except Exception: + self._safe_rollback() + raise + finally: + try: + cur.close() + except Exception: + pass + + def commit(self): + self._conn.commit() + + def rollback(self): + self._conn.rollback() + + def close(self): + if not self._closed: + try: + self._conn.rollback() + self._conn.close() + except Exception: + pass + self._closed = True + + def _safe_rollback(self): + try: + self._conn.rollback() + except Exception: + pass + + # ── sqlite3 connection surface ─────────────────────────── + @property + def total_changes(self): + return self._total_changes + + @property + def closed(self): + return self._closed + + +def connect(dsn: str) -> PgConnection: + """Open a PostgreSQL connection with the sqlite3-compatible wrapper.""" + return PgConnection(dsn) diff --git a/backend/pomodoro_mgr.py b/backend/pomodoro_mgr.py index 61af5cc..2ed7b27 100644 --- a/backend/pomodoro_mgr.py +++ b/backend/pomodoro_mgr.py @@ -2,10 +2,7 @@ State is stored per user (one row per user_id). """ -import json -import os import time -from datetime import datetime, timedelta from backend.time_util import now as tz_now from flask import current_app from backend.models import get_db diff --git a/backend/routes.py b/backend/routes.py index e69503a..33ffdd8 100644 --- a/backend/routes.py +++ b/backend/routes.py @@ -8,13 +8,11 @@ Multi-user rules enforced here: the MFA setup page until they configure (and verify) one real login method, at which point the temp TOTP is invalidated. """ -import json import os import queue import io import time as _time import secrets -from datetime import datetime from flask import ( Blueprint, request, jsonify, Response, render_template, current_app, stream_with_context, @@ -23,15 +21,15 @@ from flask import ( from backend import calendar_mgr, weather_mgr, alarm_mgr, pomodoro_mgr, countdown_mgr from backend.auth import ( login_required, login_or_temp_required, display_login_required, admin_required, - is_authenticated, is_temp_pending, is_admin, current_user_id, + is_authenticated, is_temp_pending, current_user_id, verify_login, verify_totp, verify_recovery_code, check_rate_limit, record_failure, record_success, - require_reauth, mark_reauth, generate_recovery_codes, generate_temp_password, + require_reauth, mark_reauth, generate_recovery_codes, load_auth_context, issue_tokens, set_refresh_cookie, clear_refresh_cookie, rotate_refresh, logout_current_device, load_page_context, REFRESH_COOKIE, ) from backend.models import ( - get_auth_config, update_auth_config, set_temp_password, clear_temp_password, + get_auth_config, update_auth_config, clear_temp_password, set_recovery_codes, get_recovery_codes, has_mfa_configured, get_display_config, set_display_mode, set_custom_html, set_custom_html_template, apply_custom_html, @@ -39,6 +37,10 @@ from backend.models import ( is_temp_account, is_admin_user, finalize_temp_account, find_user_by_credential_id, bump_all_session_versions, bump_user_session_version, revoke_user_tokens, revoke_all_tokens, + get_db, _read_db_version, +) +from backend.db_admin import ( + export_database_json, import_blob, detect_format, ) from backend.webauthn_mgr import ( begin_registration, complete_registration, @@ -64,6 +66,14 @@ def _load_token(): load_auth_context() +@api.before_request +def _adjust_upload_limit(): + """Database import accepts large legacy .db files; raise this request's own + body limit (per-request setter, no global race) instead of the global 5MB cap.""" + if request.path == "/api/admin/db/import" and request.method == "POST": + request.max_content_length = 200 * 1024 * 1024 + + @api.before_request def _check_origin(): """Protect state-changing requests against CSRF by verifying Origin/Referer. @@ -107,12 +117,6 @@ def login_page(): return render_template("login.html", next=next_url) -@api.route("/register-request") -def register_request_page(): - """Public page where a prospective user fills in their info (admin approves later).""" - return render_template("register_request.html") - - @api.route("/passkey") def passkey_page(): """Passkey bridge page: loaded inside the ASWebAuthenticationSession sheet @@ -492,39 +496,6 @@ def api_auth_status(): }) -# ════════════════════════════════════════════ -# USER SELF-SERVICE REQUEST -# ════════════════════════════════════════════ -# The user fills in their info; the admin confirms it and a temp account -# (with a server-held temp TOTP) is generated automatically. - -_REG_REQ_LIMIT: dict[str, list] = {} # ip -> [timestamps] - - -@api.route("/api/auth/register-request", methods=["POST"]) -def api_register_request(): - """Public: a prospective user submits their remark for admin approval. - Rate-limited per IP (5 requests per 10 minutes).""" - ip = request.remote_addr or "unknown" - now = _time.time() - stamps = [t for t in _REG_REQ_LIMIT.get(ip, []) if now - t < 600] - if len(stamps) >= 5: - _REG_REQ_LIMIT[ip] = stamps - return jsonify({"error": "申请过于频繁，请稍后再试"}), 429 - stamps.append(now) - _REG_REQ_LIMIT[ip] = stamps - - data = request.get_json() or {} - remark = (data.get("remark") or "").strip() - if not remark: - return jsonify({"error": "请填写你的称呼/备注"}), 400 - if len(remark) > 64: - return jsonify({"error": "备注过长"}), 400 - - create_user(remark=remark, role="user", is_temp=0, status="pending") - return jsonify({"ok": True, "message": "已提交申请，请等待管理员确认"}), 201 - - # ════════════════════════════════════════════ # MFA CONFIGURATION # ════════════════════════════════════════════ @@ -842,30 +813,6 @@ def admin_create_user(): return jsonify({"ok": True, "user": _user_payload(uid)}), 201 -@api.route("/api/admin/users/confirm", methods=["POST"]) -@admin_required -def admin_confirm_request(): - """Confirm a pending self-service request → temp account with temp TOTP.""" - if not require_reauth(): - return jsonify({"error": "需要重新验证身份", "reauth_required": True}), 403 - - data = request.get_json() or {} - request_id = data.get("request_id") - if request_id is None: - return jsonify({"error": "request_id required"}), 400 - - u = get_user(request_id) - if not u: - return jsonify({"error": "申请不存在"}), 404 - if u.get("status") != "pending": - return jsonify({"error": "该申请已处理"}), 400 - - import pyotp - secret = pyotp.random_base32() - update_user(request_id, is_temp=1, temp_totp_secret=secret, status="active") - return jsonify({"ok": True, "user": _user_payload(request_id)}) - - @api.route("/api/admin/users/", methods=["PUT"]) @admin_required def admin_update_user(user_id): @@ -1052,6 +999,109 @@ def admin_kick_all_users(): return jsonify({"ok": True}) +# ════════════════════════════════════════════ +# DATABASE ADMINISTRATION (Root) +# ════════════════════════════════════════════ +# Importing an old SQLite database, exporting a backup, or importing a backup +# replaces ALL current data — Root-only and re-auth required. + +def _require_root(): + """Database administration is Root-only (id 0).""" + if _cur() != 0: + return jsonify({"error": "仅 Root 用户可执行数据库管理操作"}), 403 + return None + + +@api.route("/api/admin/db/status", methods=["GET"]) +@admin_required +def admin_db_status(): + denied = _require_root() + if denied: + return denied + db_path = current_app.config.get("DB_PATH", "data.db") + conn = get_db(db_path) + try: + version = _read_db_version(conn) or 0 + server = conn.execute("SELECT version()").fetchone()[0] + rows = conn.execute( + "SELECT table_name FROM information_schema.tables " + "WHERE table_schema = current_schema() ORDER BY table_name" + ).fetchall() + tables = [r["table_name"] for r in rows] + finally: + conn.close() + return jsonify({ + "database": "postgresql", + "db_version": version, + "server": server, + "tables": tables, + "backup_format": "iot-dboard-backup", + }) + + +@api.route("/api/admin/db/export", methods=["GET"]) +@admin_required +def admin_db_export(): + denied = _require_root() + if denied: + return denied + if not require_reauth(): + return jsonify({"error": "需要重新验证身份", "reauth_required": True}), 403 + db_path = current_app.config.get("DB_PATH", "data.db") + try: + payload = export_database_json(db_path) + except Exception as e: + return jsonify({"error": f"导出失败: {e}"}), 500 + from datetime import datetime + fname = f"iot-dboard-backup-{datetime.now().strftime('%Y%m%d-%H%M%S')}.json" + return Response( + payload, + mimetype="application/json", + headers={"Content-Disposition": f'attachment; filename="{fname}"'}, + ) + + +@api.route("/api/admin/db/import", methods=["POST"]) +@admin_required +def admin_db_import(): + denied = _require_root() + if denied: + return denied + if not require_reauth(): + return jsonify({"error": "需要重新验证身份", "reauth_required": True}), 403 + + file = request.files.get("file") + if not file or not file.filename: + return jsonify({"error": "请选择要导入的数据库文件"}), 400 + data = file.read() + if not data: + return jsonify({"error": "文件为空"}), 400 + if len(data) > 200 * 1024 * 1024: + return jsonify({"error": "文件超过 200MB 限制"}), 413 + + db_path = current_app.config.get("DB_PATH", "data.db") + fmt = detect_format(data) + try: + stats = import_blob(db_path, data) + except ValueError as e: + return jsonify({"error": str(e)}), 400 + except Exception as e: + return jsonify({"error": f"导入失败（已回滚，当前数据未受影响）: {e}"}), 500 + + total = sum(v for k, v in (stats or {}).items() if isinstance(v, int)) + skipped = (stats or {}).get("_skipped_tables") or [] + msg = f"导入成功：{total} 行数据已恢复。所有设备已下线，请重新登录。" + if skipped: + msg += f" 以下表因结构与当前版本不兼容被跳过：{', '.join(skipped)}。" + return jsonify({ + "ok": True, + "format": fmt, + "stats": stats, + "message": msg, + "session_invalidated": True, + }) + + # ════════════════════════════════════════════ # DISPLAY PAGES # ════════════════════════════════════════════ diff --git a/backend/scheduler.py b/backend/scheduler.py index 4a9278f..08393e3 100644 --- a/backend/scheduler.py +++ b/backend/scheduler.py @@ -4,6 +4,8 @@ All user-scoped jobs iterate over every active user (with that user's timezone/context active) so each user's alarms, reminders, calendars, pomodoro, countdowns and weather are serviced independently. """ +import threading + from apscheduler.schedulers.background import BackgroundScheduler from backend.calendar_mgr import sync_all_subscriptions, get_upcoming_reminders from backend.weather_mgr import fetch_weather, fetch_hourly @@ -16,6 +18,129 @@ from backend.time_util import set_thread_user, clear_thread_user _scheduler: BackgroundScheduler | None = None +# ─── Scheduler single-instance guard ───────── +# With Gunicorn every worker imports the app; the background scheduler must run +# in exactly ONE process. A PostgreSQL session-level advisory lock decides which +# worker owns it (the lock is held by an open connection and released when that +# process exits, at which point another worker takes over). +_SCHEDULER_LOCK_KEY = 0x494F5444424F4152 # "IOTDBOAR" +_scheduler_lock_conn = None +_watchdog: "threading.Thread | None" = None +_WATCHDOG_INTERVAL = 30 # seconds + + +def _lock_connect(db_url: str): + """Open the lock connection with TCP keepalive + connect timeout. + Without keepalives a network blip could leave PostgreSQL holding the lock + for ~2h (default tcp keepalive), blocking every worker from taking over. + (libpq URI query params do NOT accept tcp_keepalives_* — pass them as + psycopg2 keyword args instead.)""" + import psycopg2 + return psycopg2.connect( + db_url, + connect_timeout=5, + keepalives_idle=30, + keepalives_interval=10, + keepalives_count=3, + ) + + +def try_acquire_scheduler_lock(db_url: str) -> bool: + """Try to become the scheduler owner. Returns True for the winner.""" + global _scheduler_lock_conn + conn = None + try: + conn = _lock_connect(db_url) + cur = conn.cursor() + cur.execute("SELECT pg_try_advisory_lock(%s)", (_SCHEDULER_LOCK_KEY,)) + ok = bool(cur.fetchone()[0]) + cur.close() + if ok: + _scheduler_lock_conn = conn + return True + conn.close() + return False + except Exception: + try: + if _scheduler_lock_conn: + _scheduler_lock_conn.close() + _scheduler_lock_conn = None + except Exception: + pass + try: + if conn is not None: + conn.close() # close the partially-used local connection + except Exception: + pass + return False + + +def release_scheduler_lock(): + """Release the advisory lock (also closes the holding connection).""" + global _scheduler_lock_conn + if _scheduler_lock_conn: + try: + _scheduler_lock_conn.close() + except Exception: + pass + _scheduler_lock_conn = None + + +def _watchdog_loop(app, db_url: str): + """Keeps the scheduler alive across worker restarts/freezes: + - lock owner: periodically checks the lock connection is healthy; + - non-owner: periodically retries to take over the lock. + """ + global _scheduler + import time + while True: + time.sleep(_WATCHDOG_INTERVAL) + try: + if _scheduler_lock_conn is not None: + # Owner: health check on the lock connection, and make sure the + # scheduler thread itself is still running (start failure or a + # dead thread must not hold the lock forever). + cur = _scheduler_lock_conn.cursor() + cur.execute("SELECT 1") + cur.fetchone() + cur.close() + if _scheduler is None or not _scheduler.running: + if _scheduler: + try: + stop_scheduler() + except Exception: + pass + start_scheduler(app) + app.logger.info("Scheduler restarted (lock held but not running)") + elif _scheduler is None: + # Non-owner: try to take over (previous owner died/froze). + if try_acquire_scheduler_lock(db_url): + start_scheduler(app) + app.logger.info("Scheduler started (watchdog takeover)") + except Exception: + # Lock connection became unhealthy → drop the lock and re-race. + release_scheduler_lock() + if _scheduler: + try: + stop_scheduler() + except Exception: + pass + try: + if try_acquire_scheduler_lock(db_url): + start_scheduler(app) + app.logger.info("Scheduler re-acquired lock after watchdog failure") + except Exception: + pass + + +def start_watchdog(app, db_url: str): + """Start the scheduler watchdog thread (idempotent).""" + global _watchdog + if _watchdog is not None and _watchdog.is_alive(): + return + _watchdog = threading.Thread(target=_watchdog_loop, args=(app, db_url), daemon=True) + _watchdog.start() + def _each_user(app, fn): """Run fn(user_id) for every active user inside the app context.""" @@ -82,5 +207,7 @@ def start_scheduler(app): def stop_scheduler(): global _scheduler if _scheduler: - _scheduler.shutdown(wait=False) - _scheduler = None + try: + _scheduler.shutdown(wait=False) + finally: + _scheduler = None diff --git a/backend/time_util.py b/backend/time_util.py index ba3ab76..e1a42f7 100644 --- a/backend/time_util.py +++ b/backend/time_util.py @@ -15,7 +15,6 @@ The timezone is cached per-request via Flask app config to avoid DB lookups. import time as _time import threading from datetime import datetime, timedelta, timezone, date -from typing import Union import pytz @@ -130,14 +129,14 @@ def set_tz(tz_str: str) -> bool: # Convert to an IANA-like name: pytz has FixedOffset but easiest is # just use a fixed-offset timezone import pytz - tz = pytz.FixedOffset(sign * hours * 60) # FixedOffset expects minutes + pytz.FixedOffset(sign * hours * 60) # FixedOffset expects minutes except (ValueError, IndexError): return False else: # IANA name — validate by trying to load try: import pytz - tz = pytz.timezone(normalized) + pytz.timezone(normalized) except Exception: return False diff --git a/backend/weather_mgr.py b/backend/weather_mgr.py index fe7830f..e15bcf3 100644 --- a/backend/weather_mgr.py +++ b/backend/weather_mgr.py @@ -2,7 +2,7 @@ import json import requests from datetime import datetime -from backend.time_util import now as tz_now, strftime as tz_strftime, today_str as tz_today_str, strptime as tz_strptime, localize as tz_localize +from backend.time_util import now as tz_now, strftime as tz_strftime, strptime as tz_strptime, localize as tz_localize from flask import current_app from backend.models import get_db from backend.jwt_util import sign_jwt, generate_key_pair @@ -353,10 +353,8 @@ def get_next_update_times(user_id: int = None) -> dict: h_updated = None h_row = db.execute("SELECT data FROM weather_hourly_cache WHERE user_id=?", (user_id,)).fetchone() if h_row and h_row["data"]: - h_data = json.loads(h_row["data"]) # Hourly data is a list; we store fetched_at separately, but for display - # use the SQLite fetched_at (converted) or just show the schedule - # Let's use the local-computed schedule as last-update signal + # use the local-computed schedule as last-update signal h_row2 = db.execute("SELECT fetched_at FROM weather_hourly_cache WHERE user_id=?", (user_id,)).fetchone() if h_row2 and h_row2[0]: # Keep the UTC string but note it's UTC — we'll just pass it diff --git a/backend/webauthn_mgr.py b/backend/webauthn_mgr.py index b72c76f..87e2794 100644 --- a/backend/webauthn_mgr.py +++ b/backend/webauthn_mgr.py @@ -108,7 +108,6 @@ def begin_registration(request, user_id: int, display_name: str = "", existing_credential_id: str = None) -> dict: """Start WebAuthn registration for a specific user. Returns options dict.""" rp_id = _get_rp_id(request) - origin = _get_origin(request) user_name = f"user_{user_id}" user_handle = user_name.encode("utf-8") @@ -200,7 +199,6 @@ def begin_authentication(request, allow_credential_ids: list = None) -> dict: ceremony to (used for re-authentication of the current user). """ rp_id = _get_rp_id(request) - origin = _get_origin(request) allow_credentials = [] if allow_credential_ids: diff --git a/docker-compose.yml b/docker-compose.yml index 9a15b34..20fd37a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,4 +1,29 @@ +# 部署拓扑：PostgreSQL 与 Web 应用都在 Docker 内。 +# - db：PostgreSQL 16 (alpine，镜像小)，数据持久化在 pgdata 卷 +# - app：Gunicorn 2 worker × 2 线程（对应 1 物理核心 / 2 逻辑核心），限制 2G 内存 services: + db: + image: postgres:16-alpine + container_name: cal-display-db + restart: unless-stopped + environment: + - POSTGRES_USER=iot + - POSTGRES_PASSWORD=${POSTGRES_PASSWORD:-iot} + - POSTGRES_DB=iot_dboard + - TZ=UTC + volumes: + - pgdata:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U iot -d iot_dboard"] + interval: 5s + timeout: 3s + retries: 12 + deploy: + resources: + limits: + cpus: "0.5" + memory: 512M + app: build: . container_name: cal-display @@ -6,15 +31,32 @@ services: ports: - "5000:5000" volumes: - # 持久化数据库 + # 挂载目录：可放置待导入的旧 SQLite db 文件（也支持 /control 设置页直接上传导入） - ./data:/app/data - # 持久化项目代码（首次自动初始化为镜像内的克隆内容，之后 git pull 生效） + # 持久化项目代码（首次自动初始化为镜像内代码，之后 git pull 生效） - app_code:/app environment: - - FLASK_BIND=0.0.0.0 - - FLASK_DB_PATH=/app/data/data.db + - DATABASE_URL=postgresql://iot:${POSTGRES_PASSWORD:-iot}@db:5432/iot_dboard - FLASK_SECRET_KEY=${FLASK_SECRET_KEY:-} - FLASK_HTTPS=${FLASK_HTTPS:-} + - GUNICORN_WORKERS=${GUNICORN_WORKERS:-2} + - GUNICORN_THREADS=${GUNICORN_THREADS:-8} + depends_on: + db: + condition: service_healthy + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:5000/api/auth/status', timeout=3)"] + interval: 30s + timeout: 5s + retries: 3 + start_period: 20s + deploy: + resources: + limits: + # 1 物理核心 = 2 逻辑核心 + 2G 内存 + cpus: "2" + memory: 2G volumes: + pgdata: app_code: diff --git a/frontend/static/js/control.js b/frontend/static/js/control.js index 01f8110..291c7c6 100644 --- a/frontend/static/js/control.js +++ b/frontend/static/js/control.js @@ -1060,10 +1060,115 @@ var tab = $('users-tab'); if (tab) tab.style.display = ''; loadUsers(); + // 数据库管理仅 Root (id 0) 可见 + if (status.user.id === 0) { + var dbSec = $('db-admin-section'); + if (dbSec) dbSec.style.display = ''; + loadDbStatus(); + } } }); } + // ─── Database administration (Root only) ─── + function loadDbStatus() { + var el = $('db-status'); + if (!el) return; + api('GET', '/api/admin/db/status', null, function (d) { + if (d && d.database) { + el.textContent = 'PostgreSQL · schema v' + d.db_version + ' · 表 ' + (d.tables ? d.tables.length : '?') + ' 张'; + el.style.color = '#2e7d32'; + } else { + el.textContent = (d && d.error) || '获取状态失败'; + el.style.color = '#E05555'; + } + }); + } + + function dbMsg(text, isErr) { + var el = $('db-admin-msg'); + if (!el) return; + el.textContent = text; + el.style.color = isErr ? '#E05555' : '#2e7d32'; + } + + function initDbAdmin() { + var exportBtn = $('db-export-btn'); + if (exportBtn) { + exportBtn.addEventListener('click', function () { + exportBtn.disabled = true; + dbMsg('正在导出...', false); + _reauthThen(function () { + // 用 fetch 携带 Authorization 下载备份文件 + var token = Auth.getAccess && Auth.getAccess(); + var xhr = new XMLHttpRequest(); + xhr.open('GET', '/api/admin/db/export'); + if (token) xhr.setRequestHeader('Authorization', 'Bearer ' + token); + xhr.responseType = 'blob'; + xhr.onload = function () { + exportBtn.disabled = false; + if (xhr.status === 200) { + var blob = xhr.response; + var disp = xhr.getResponseHeader('Content-Disposition') || ''; + var m = disp.match(/filename="?([^";]+)"?/); + var name = m ? m[1] : 'iot-dboard-backup.json'; + var a = document.createElement('a'); + a.href = URL.createObjectURL(blob); + a.download = name; + document.body.appendChild(a); + a.click(); + setTimeout(function () { URL.revokeObjectURL(a.href); a.remove(); }, 1000); + dbMsg('导出成功，文件已开始下载', false); + } else if (xhr.status === 403) { + dbMsg('需要重新验证身份，请重试', true); + } else { + dbMsg('导出失败 (HTTP ' + xhr.status + ')', true); + } + }; + xhr.onerror = function () { exportBtn.disabled = false; dbMsg('导出失败：网络错误', true); }; + xhr.send(); + }, function () { + exportBtn.disabled = false; + dbMsg('身份验证未通过，导出已取消', true); + }); + }); + } + + var form = $('db-import-form'); + if (form) { + form.addEventListener('submit', function (e) { + e.preventDefault(); + var fileInput = $('db-import-file'); + var file = fileInput && fileInput.files && fileInput.files[0]; + if (!file) { dbMsg('请先选择文件', true); return; } + var btn = $('db-import-btn'); + btn.disabled = true; + dbMsg('正在导入（大文件可能需要一些时间）...', false); + _reauthThen(function () { + var fd = new FormData(); + fd.append('file', file); + api('POST', '/api/admin/db/import', fd, function (res) { + btn.disabled = false; + if (res && res.ok) { + dbMsg(res.message || '导入成功', false); + // 导入会重建全部数据并清空会话（含当前设备），提示后跳转登录页 + setTimeout(function () { + try { localStorage.removeItem('dashboard_access_token'); } catch (e) {} + location.href = '/login?view=control&next=%2Fcontrol'; + }, 1800); + } else { + dbMsg((res && res.error) || '导入失败', true); + } + }); + }, function () { + btn.disabled = false; + dbMsg('身份验证未通过，导入已取消', true); + }); + }); + } + } + initDbAdmin(); + // ─── User Management (admin only) ───────── var _totpTimer = null; @@ -1137,30 +1242,13 @@ function loadUsers() { if (!$('users-tab') || $('users-tab').style.display === 'none') return; api('GET', '/api/admin/users', null, function (users) { - var pending = users.filter(function (u) { return u.status === 'pending'; }); - var active = users.filter(function (u) { return u.status === 'active'; }); - - // Pending requests - var pendHtml = ''; - if (pending.length > 0) { - pendHtml = '

⏳ 待确认申请（用户自助填写信息）

'; - pending.forEach(function (u) { - pendHtml += '
' + - '
' + _esc(u.remark) + ' ' + _esc(u.created_at || '') + '
' + - '
' + - '' + - '' + - '
'; - }); - } - $('pending-requests').innerHTML = pendHtml; - - // Active users + // 统一渲染全部用户（含历史 pending 申请用户，标记后可删除） var html = '

👥 用户列表

'; - if (active.length === 0) html += '
暂无用户
'; - active.forEach(function (u) { + if (users.length === 0) html += '
暂无用户
'; + users.forEach(function (u) { var badges = []; if (u.role === 'admin') badges.push('管理员'); + if (u.status === 'pending') badges.push('待确认申请'); if (u.is_temp) badges.push('临时账户'); if (u.has_mfa) badges.push('已配置 MFA'); html += '
' + @@ -1183,23 +1271,11 @@ } function bindUserActions() { - document.querySelectorAll('#pending-requests [data-action], #user-list [data-action]').forEach(function (btn) { + document.querySelectorAll('#user-list [data-action]').forEach(function (btn) { btn.addEventListener('click', function () { var action = btn.dataset.action; var id = parseInt(btn.dataset.id, 10); - if (action === 'confirm-request') { - _reauthThen(function () { - api('POST', '/api/admin/users/confirm', { request_id: id }, function (res) { - if (res.ok) { toast('已生成临时账户'); loadUsers(); } else { toast(res.error || '操作失败'); } - }); - }); - } else if (action === 'delete-request') { - _reauthThen(function () { - api('DELETE', '/api/admin/users/' + id, null, function (res) { - if (res.ok) { toast('已拒绝该申请'); loadUsers(); } else { toast(res.error || '操作失败'); } - }); - }); - } else if (action === 'edit-remark') { + if (action === 'edit-remark') { _reauthThen(function () { showPrompt('编辑用户备注', '新的备注', '', function (newRemark) { if (newRemark === null || !newRemark.trim()) return; diff --git a/frontend/static/js/display.js b/frontend/static/js/display.js index 435a49b..a94096b 100644 --- a/frontend/static/js/display.js +++ b/frontend/static/js/display.js @@ -124,6 +124,7 @@ state.events = data.events || []; state.alarms = data.alarms || []; state.countdowns = data.countdowns || []; + state.pomodoro = data.pomodoro || state.pomodoro || {}; state.update_schedule = data.update_schedule || {}; state.minutely = data.minutely || null; state.warnings = data.warnings || null; @@ -145,8 +146,79 @@ }); } + // ─── Local due-check fallback ───────────── + // 多 Gunicorn worker 下，后台 scheduler 与显示端 SSE 连接可能落在不同进程 + // （进程内 EventBus 无法跨进程投递），SSE 提醒事件可能丢失。这里每秒本地 + // 检测一次，保证闹钟/倒计时/番茄钟到点必定提醒；SSE 正常时由 fireOnce 去重。 + state._firedLocal = {}; + function fireOnce(key) { + if (state._firedLocal[key]) return false; + state._firedLocal[key] = true; + return true; + } + + function localDueCheck() { + var nowSec = Date.now() / 1000; + var todayKey = new Date().toDateString(); + + // 1) 闹钟：按账户时区匹配 HH:MM + 星期 + if (state.timezone) { + try { + var parts = new Intl.DateTimeFormat('en-US', { + timeZone: state.timezone, hour: '2-digit', minute: '2-digit', + weekday: 'short', hourCycle: 'h23', + }).formatToParts(new Date()); + var hm = '', wd = ''; + parts.forEach(function (p) { + if (p.type === 'hour') hm = p.value; + else if (p.type === 'minute') hm += ':' + p.value; + else if (p.type === 'weekday') wd = p.value; + }); + var wdNum = { Sun: 7, Mon: 1, Tue: 2, Wed: 3, Thu: 4, Fri: 5, Sat: 6 }[wd]; + (state.alarms || []).forEach(function (a) { + if (!a.enabled || String(a.time) !== hm) return; + var days = String(a.days || '1,2,3,4,5,6,7').split(','); + if (days.indexOf(String(wdNum)) === -1) return; + if (fireOnce('alarm-' + a.id + '-' + todayKey)) { + showAlarmOverlay({ alarms: [a], ts: new Date().toISOString(), source: 'local' }); + } + }); + } catch (e) {} + } + + // 2) 倒计时：end_time（Unix 秒）到期 + (state.countdowns || []).forEach(function (cd) { + if (!cd.enabled || !cd.end_time) return; + if (parseFloat(cd.end_time) > nowSec) return; + if (fireOnce('cd-' + cd.id + '-' + (cd.end_time || ''))) { + showAlarmOverlay({ countdowns: [cd], ts: new Date().toISOString(), source: 'local' }); + switchPanel('countdown'); + } + }); + + // 3) 番茄钟：running 且 end_time（Unix 秒）到期 + var p = state.pomodoro || {}; + if (p.running && p.end_time && parseFloat(p.end_time) <= nowSec) { + if (fireOnce('pomo-' + p.end_time)) { + showPomodoroOverlay(); + } + } + } + + function startLocalDueCheck() { + if (!state._localCheckTimer) { + state._localCheckTimer = setInterval(localDueCheck, 1000); + } + } + function openSSE(url) { var es = new EventSource(url); + // SSE 与后台 scheduler 可能不在同一 Gunicorn worker（进程内事件总线无法跨 + // 进程投递）：本地到期检测 + 周期性重拉完整状态作为兜底，保证最终一致。 + startLocalDueCheck(); + if (!state._stateTimer) { + state._stateTimer = setInterval(fetchState, 15000); + } es.onmessage = function (e) { try { var msg = JSON.parse(e.data); @@ -174,7 +246,11 @@ function handleEvent(msg) { if (msg.type === 'alarm_triggered') { fetchState(); - showAlarmOverlay(msg.data); + var todayKey = new Date().toDateString(); + var anyNew = (msg.data.alarms || []).some(function (a) { + return fireOnce('alarm-' + a.id + '-' + todayKey); + }); + if (anyNew) showAlarmOverlay(msg.data); } else if (msg.type === 'pomodoro_updated') { state.pomodoro = msg.data || {}; renderPomodoro(); @@ -205,8 +281,13 @@ showCustomNotification(msg.data); } else if (msg.type === 'countdown_triggered') { fetchState(); - showAlarmOverlay(msg.data); // reuse alarm overlay UI - switchPanel('countdown'); + var anyCd = (msg.data.countdowns || []).some(function (cd) { + return fireOnce('cd-' + cd.id + '-' + (cd.end_time || '')); + }); + if (anyCd) { + showAlarmOverlay(msg.data); // reuse alarm overlay UI + switchPanel('countdown'); + } } else if (msg.type === 'display_mode_changed' || msg.type === 'force_refresh') { location.reload(); } else if (msg.type === 'theme_changed') { @@ -224,13 +305,14 @@ if (state.alarmActive) return; // already showing state.alarmActive = true; - // Build info text (supports both alarm and countdown data) + // Build info text (supports both alarm and countdown data). + // 用户可控文本（label/time/target_time）一律转义，防止存储型 XSS。 var info = ''; (data.alarms || []).forEach(function (a) { - info += a.time + (a.label ? ' ' + a.label : '') + '
'; + info += _esc(a.time) + (a.label ? ' ' + _esc(a.label) : '') + '
'; }); (data.countdowns || []).forEach(function (cd) { - info += cd.target_time + (cd.label ? ' ' + cd.label : '') + '
'; + info += _esc(cd.target_time) + (cd.label ? ' ' + _esc(cd.label) : '') + '
'; }); var overlay = $('alarm-overlay'); @@ -658,7 +740,7 @@ html += '
' + '
' + '
' + durStr + '
' + - (cd.label ? '
' + cd.label + '
' : '') + + (cd.label ? '
' + _esc(cd.label) + '
' : '') + '
' + '
' + remainStr + '
' + '
' + @@ -1297,9 +1379,9 @@ } if (m.error) { if (m.cooldown) { - content.innerHTML = '
' + m.error + '
'; + content.innerHTML = '
' + _esc(m.error) + '
'; } else { - content.innerHTML = '
' + m.error + '
'; + content.innerHTML = '
' + _esc(m.error) + '
'; } return; } @@ -1307,7 +1389,7 @@ var list = m.minutely || []; var summary = m.summary || ''; var html = ''; - if (summary) html += '
' + summary + '
'; + if (summary) html += '
' + _esc(summary) + '
'; if (list.length > 0) { html += '
'; list.forEach(function (item) { @@ -1355,11 +1437,11 @@ var fields = ''; for (var key in item) { if (item.hasOwnProperty(key) && item[key]) { - fields += '
' + key + ': ' + item[key] + '
'; + fields += '
' + _esc(key) + ': ' + _esc(item[key]) + '
'; } } html += '
' + - '
' + (item.title || item.type || '预警') + '
' + + '
' + _esc(item.title || item.type || '预警') + '
' + fields + '
'; }); @@ -1443,9 +1525,9 @@ var days = a.once ? '仅一次' : (a.days || '').split(',').map(function (d) { return dayNames[parseInt(d)]; }).join(' '); html += '
' + '
' + - '
' + a.time + '
' + - (a.label ? '
' + a.label + '
' : '') + - '
' + days + '
' + + '
' + _esc(a.time) + '
' + + (a.label ? '
' + _esc(a.label) + '
' : '') + + '
' + _esc(days) + '
' + '
' + '' + '' + diff --git a/frontend/templates/control.html b/frontend/templates/control.html index 0fad2d9..df64dba 100644 --- a/frontend/templates/control.html +++ b/frontend/templates/control.html @@ -261,7 +261,6 @@

🚫 踢掉所有用户：所有用户的控制端与显示端立即全部下线（含当前设备），需二次验证身份。

-
@@ -315,6 +314,23 @@
包含当前控制端与所有显示端，全部立即下线。
+

时区设置

diff --git a/frontend/templates/login.html b/frontend/templates/login.html index b0327ca..97e55ba 100644 --- a/frontend/templates/login.html +++ b/frontend/templates/login.html @@ -206,10 +206,6 @@
- -
- - - -
-

📋 申请账号

-

填写你的称呼，提交后等待管理员确认。管理员确认后你会收到一个临时验证码，用它登录并配置你自己的登录方式（Passkey 或 TOTP）。

- - -
- ← 返回登录 -
- - - - diff --git a/requirements.txt b/requirements.txt index 826c8e5..93bf110 100644 --- a/requirements.txt +++ b/requirements.txt @@ -10,3 +10,5 @@ cryptography>=42.0 webauthn>=2.0 pyotp>=2.9 qrcode[pil]>=7.4 +psycopg2-binary>=2.9 +gunicorn>=21.2 diff --git a/run.py b/run.py index 58778d3..0e2c688 100644 --- a/run.py +++ b/run.py @@ -1,40 +1,34 @@ -"""Production entry point for the Calendar Display application.""" +"""Development entrypoint (Flask dev server, port 5001). + +For production use Gunicorn via `start:app` (see start.py and Dockerfile). +""" import os import sys os.chdir(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from backend.app import create_app -from backend.scheduler import start_scheduler, stop_scheduler -from backend.models import _EMITTED_PASSWORDS, get_auth_config +from backend.scheduler import start_scheduler, stop_scheduler, try_acquire_scheduler_lock -# Use environment variable to toggle debug, default False for production -DEBUG = os.environ.get("FLASK_DEBUG", "").lower() in ("1", "true", "yes") - -db_path = os.environ.get("FLASK_DB_PATH", os.path.join(os.path.dirname(__file__), "data.db")) -app = create_app(db_path=db_path) +db_url = os.environ.get("DATABASE_URL", + "postgresql://iot:iot@localhost:5432/iot_dboard") +app = create_app(db_path=db_url) if __name__ == "__main__": - # In debug mode, scheduler runs only in the reloader child process + DEBUG = os.environ.get("FLASK_DEBUG", "").lower() in ("1", "true", "yes") + + # In debug mode the reloader re-imports this module; only the child process + # should start the scheduler (advisory lock keeps it single anyway). is_reloader_child = os.environ.get("WERKZEUG_RUN_MAIN") == "true" - if not DEBUG or is_reloader_child: - start_scheduler(app) - - # Print temp password at startup - with app.app_context(): - if not _EMITTED_PASSWORDS: - cfg = get_auth_config() - pw = cfg.get("temp_password", "") - if pw and cfg.get("must_configure_mfa"): - print(f"\n{'='*50}", file=sys.stderr) - print(f" 🔐 临时密码: {pw}", file=sys.stderr) - print(f" ⚠️ 请使用此密码登录并配置双因素认证", file=sys.stderr) - print(f"{'='*50}\n", file=sys.stderr) + if try_acquire_scheduler_lock(db_url): + start_scheduler(app) try: bind_host = os.environ.get("FLASK_BIND", "127.0.0.1") - app.run(host=bind_host, port=5001, debug=DEBUG, threaded=True) + app.run(host=bind_host, port=int(os.environ.get("FLASK_PORT", "5001")), + debug=DEBUG, threaded=True) finally: if not DEBUG or is_reloader_child: - stop_scheduler() \ No newline at end of file + stop_scheduler() diff --git a/start.py b/start.py index c9db523..016152e 100644 --- a/start.py +++ b/start.py @@ -1,23 +1,72 @@ -import sys, os +"""Production WSGI entrypoint for Gunicorn (and plain `python start.py`). + +- Creates the Flask app and initializes the PostgreSQL schema (with the + schema-version migration gate). +- Starts the background scheduler ONLY in the process that wins the + PostgreSQL advisory lock, so multiple Gunicorn workers never duplicate + alarms / reminders / weather jobs. +- Prints the Root temp password (when one was generated or is still pending). + +Gunicorn usage (see Dockerfile / docker-compose.yml): + gunicorn --workers 2 --threads 2 --bind 0.0.0.0:5000 start:app +""" +import os +import sys + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + from backend.app import create_app -from backend.scheduler import start_scheduler +from backend.scheduler import ( + start_scheduler, stop_scheduler, try_acquire_scheduler_lock, start_watchdog, +) from backend.models import _EMITTED_PASSWORDS, get_auth_config -db_path = os.environ.get("FLASK_DB_PATH", os.path.join(os.path.dirname(__file__), "data.db")) -app = create_app(db_path=db_path) -start_scheduler(app) +# PostgreSQL connection string (kept as the historical DB_PATH config key). +db_url = os.environ.get("DATABASE_URL", + "postgresql://iot:iot@db:5432/iot_dboard") -# Print temp password if it wasn't already emitted during DB init -with app.app_context(): - if not _EMITTED_PASSWORDS: - cfg = get_auth_config() - pw = cfg.get("temp_password", "") - if pw and cfg.get("must_configure_mfa"): - print(f"\n{'='*50}", file=sys.stderr) - print(f" 🔐 临时密码 (重启后仍有效): {pw}", file=sys.stderr) - print(f" ⚠️ 请使用此密码登录并配置双因素认证", file=sys.stderr) - print(f"{'='*50}\n", file=sys.stderr) +app = create_app(db_path=db_url) -bind_host = os.environ.get("FLASK_BIND", "127.0.0.1") -app.run(host=bind_host, port=5000, debug=False, threaded=True) \ No newline at end of file + +def _print_temp_password(): + with app.app_context(): + if not _EMITTED_PASSWORDS: + cfg = get_auth_config() + pw = cfg.get("temp_password", "") + if pw and cfg.get("must_configure_mfa"): + print(f"\n{'='*50}", file=sys.stderr) + print(f" 🔐 临时密码: {pw}", file=sys.stderr) + print(f" ⚠️ 请使用此密码登录并配置双因素认证", file=sys.stderr) + print(f"{'='*50}\n", file=sys.stderr) + else: + for pw in _EMITTED_PASSWORDS: + print(f"\n{'='*50}", file=sys.stderr) + print(f" 🔐 Root 初始临时密码: {pw}", file=sys.stderr) + print(f" ⚠️ 请使用此密码登录 Root 并配置双因素认证", file=sys.stderr) + print(f"{'='*50}\n", file=sys.stderr) + + +_print_temp_password() + + +def _maybe_start_scheduler(): + """Start the scheduler if this process wins the advisory lock.""" + if os.environ.get("FLASK_DISABLE_SCHEDULER", "").lower() in ("1", "true", "yes"): + return + if try_acquire_scheduler_lock(db_url): + start_scheduler(app) + app.logger.info("Scheduler started (advisory lock acquired)") + + +_maybe_start_scheduler() +if os.environ.get("FLASK_DISABLE_SCHEDULER", "").lower() not in ("1", "true", "yes"): + start_watchdog(app, db_url) + + +if __name__ == "__main__": + import atexit + atexit.register(stop_scheduler) + DEBUG = os.environ.get("FLASK_DEBUG", "").lower() in ("1", "true", "yes") + bind_host = os.environ.get("FLASK_BIND", "127.0.0.1") + app.run(host=bind_host, port=int(os.environ.get("FLASK_PORT", "5000")), + debug=DEBUG, threaded=True) diff --git a/tests/test_multi_user.py b/tests/test_multi_user.py index f1e1c1a..62173e0 100644 --- a/tests/test_multi_user.py +++ b/tests/test_multi_user.py @@ -1,11 +1,18 @@ -"""Multi-user system test suite (dual-token auth). +"""Multi-user system test suite (dual-token auth) — PostgreSQL edition. Run: python3 tests/test_multi_user.py -Covers: Root bootstrap, legacy/new-DB migration + schema-version gate, +Requires a local PostgreSQL (default 127.0.0.1:5432, iot/iot@iot_dboard; +override with TEST_PG_HOST / TEST_PG_USER / TEST_PG_PASS / TEST_PG_DB). + +Each test uses its own isolated PostgreSQL schema; the schema is dropped and +recreated at startup. + +Covers: Root bootstrap, SQLite legacy import + schema-version gate, admin registration (both modes), temp-account lifecycle, data isolation, admin-only enforcement, re-auth (TOTP + passkey fallback), deletion, kick devices (self account + Root kick-all), access/refresh token lifecycle, -SSE ticket scoping, rate limits, CSRF, per-account timezone/theme. +SSE ticket scoping, rate limits, CSRF, per-account timezone/theme, +database administration (export / import / root-only / atomicity). """ import base64 import io @@ -16,10 +23,12 @@ import sqlite3 import sys import tempfile import time +import urllib.parse sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) os.environ.setdefault("FLASK_SECRET_KEY", "test-secret") +import psycopg2 import pyotp from backend.app import create_app @@ -30,9 +39,32 @@ from backend.models import ( create_user, update_auth_config as uac, ) from backend.events import EventBus +from backend import db_admin PASS, FAIL = 0, 0 +# ─── PostgreSQL test plumbing ───────────────── +PG_HOST = os.environ.get("TEST_PG_HOST", "127.0.0.1") +PG_USER = os.environ.get("TEST_PG_USER", "iot") +PG_PASS = os.environ.get("TEST_PG_PASS", "iot") +PG_DB = os.environ.get("TEST_PG_DB", "iot_dboard") + + +def _dsn(schema: str) -> str: + """DSN for one isolated schema (search_path pinned to it).""" + opts = urllib.parse.quote(f"-csearch_path={schema}") + return f"postgresql://{PG_USER}:{PG_PASS}@{PG_HOST}/{PG_DB}?options={opts}" + + +def reset_schema(schema: str): + conn = psycopg2.connect(f"postgresql://{PG_USER}:{PG_PASS}@{PG_HOST}/{PG_DB}") + conn.autocommit = True + cur = conn.cursor() + cur.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE') + cur.execute(f'CREATE SCHEMA "{schema}"') + cur.close() + conn.close() + class TClient: """Flask test client wrapper that attaches the access token to each request. @@ -73,14 +105,14 @@ class TClient: def _groups_for(db): - conn = sqlite3.connect(db) + conn = get_db(db) n = conn.execute("SELECT COUNT(*) FROM calendar_groups WHERE user_id=0").fetchone()[0] conn.close() return n def _count(db, table, user_id): - conn = sqlite3.connect(db) + conn = get_db(db) n = conn.execute(f"SELECT COUNT(*) FROM {table} WHERE user_id=?", (user_id,)).fetchone()[0] conn.close() return n @@ -96,8 +128,9 @@ def check(name, cond, detail=""): print(f" ❌ {name} {detail}") -def fresh_app(tmp): - db = os.path.join(tmp, "fresh.db") +def fresh_app(schema): + reset_schema(schema) + db = _dsn(schema) app = create_app(db_path=db) return app, db @@ -150,8 +183,7 @@ def b64url(hexstr): print("=" * 60) print("TEST 1: Fresh DB bootstrap → Root user + temp password + MFA setup") print("=" * 60) -tmp = tempfile.mkdtemp() -app, db = fresh_app(tmp) +app, db = fresh_app("t1") _EMITTED_PASSWORDS.clear() init_db(db) check("Root user exists with id 0", get_user(0, db) is not None) @@ -182,12 +214,24 @@ check("Wrong code rejected", r.status_code == 401) check("No token → 401", app.test_client().get("/api/theme").status_code == 401) check("Bad token → 401", app.test_client().get("/api/theme", headers={"Authorization": "Bearer nope"}).status_code == 401) +# datetime('now') 翻译正确性：PG to_char 必须用 HH24/MI（HH 是 12 小时制、MM 是月份） +import re as _re +from backend.pg_compat import _translate_sql +_tr, _ = _translate_sql("UPDATE t SET ts=datetime('now') WHERE id=?") +check("datetime('now') 翻译为 HH24:MI", "HH24:MI:SS" in _tr and "HH:MM:SS" not in _tr) +conn = get_db(db) +_trow = conn.execute("SELECT to_char(now(), 'YYYY-MM-DD HH24:MI:SS') AS t").fetchone() +check("to_char HH24:MI 输出格式正确", bool(_re.fullmatch(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", _trow["t"]))) +conn.close() + print() print("=" * 60) -print("TEST 2: Legacy single-user DB migration → data becomes Root's") +print("TEST 2: Legacy SQLite import → data becomes Root's (db_admin)") print("=" * 60) # Build a legacy single-user database (old schema: no users/user_id, pin-only auth) -legacy = os.path.join(tmp, "legacy.db") +legacy_fd, legacy = tempfile.mkstemp(suffix=".db") +with os.fdopen(legacy_fd, "wb"): + pass _lc = sqlite3.connect(legacy) _lc.executescript(""" CREATE TABLE calendar_groups (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, color TEXT DEFAULT '#4A90D9', enabled INTEGER DEFAULT 1, created_at TEXT DEFAULT (datetime('now'))); @@ -207,47 +251,52 @@ _lc.execute("INSERT INTO auth_config (id, pin) VALUES (1, '0411')") _lc.execute("INSERT INTO clock_config (city, timezone, is_primary) VALUES ('Beijing', 'Asia/Shanghai', 1)") _lc.commit() _lc.close() -_EMITTED_PASSWORDS.clear() -init_db(legacy) -check("Root created after legacy migration", get_user(0, legacy) is not None) -check("Legacy calendar groups → Root (user 0)", _groups_for(legacy) >= 3) -check("Legacy alarms migrated to user 0", _count(legacy, "alarms", 0) >= 1) -check("Root gets a fresh temp password (legacy had only PIN)", bool(get_temp_password(legacy, 0))) -conn = sqlite3.connect(legacy) -tables = [r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")] +legacy_bytes = open(legacy, "rb").read() +os.unlink(legacy) + +reset_schema("t2") +db2 = _dsn("t2") +init_db(db2) +check("Legacy file detected as sqlite", db_admin.detect_format(legacy_bytes) == "sqlite") +stats = db_admin.import_sqlite(db2, legacy_bytes) +check("Import succeeds", stats is not None and stats.get("calendar_groups") == 3) +check("Root created after legacy import", get_user(0, db2) is not None) +check("Legacy calendar groups → Root (user 0)", _groups_for(db2) >= 3) +check("Legacy alarms migrated to user 0", _count(db2, "alarms", 0) >= 1) +check("Root gets a fresh temp password (legacy had only PIN)", bool(get_temp_password(db2, 0))) +conn = get_db(db2) +tables = [r["table_name"] for r in conn.execute( + "SELECT table_name FROM information_schema.tables WHERE table_schema = current_schema()").fetchall()] conn.close() check("camera_config table dropped", "camera_config" not in tables) check("users table exists", "users" in tables) check("meta table exists", "meta" in tables) check("auth_tokens table exists", "auth_tokens" in tables) -check("Legacy DB stamped with program version", _read_db_version(get_db(legacy)) == DB_VERSION) -get_db(legacy).close() +check("Legacy DB stamped with program version", _read_db_version(get_db(db2)) == DB_VERSION) +get_db(db2).close() print() print("=" * 60) -print("TEST 3: Schema-version gate (same→skip, newer→abort, migrate→stamp)") +print("TEST 3: Schema-version gate (same→skip, newer→abort, lower→migrate+stamp)") print("=" * 60) -db3 = os.path.join(tmp, "roundtrip.db") +reset_schema("t3") +db3 = _dsn("t3") init_db(db3) -conn = sqlite3.connect(db3) -mtime1 = os.stat(db3).st_mtime_ns -conn.close() init_db(db3) -check("Same-version init skips rebuild", os.stat(db3).st_mtime_ns == mtime1) -conn = get_db(db3) -check("Version still matches", _read_db_version(conn) == DB_VERSION) -conn.close() +check("Same-version init idempotent", _read_db_version(get_db(db3)) == DB_VERSION) +get_db(db3).close() # user data preserved across a version-mismatch migration (simulate future bump) -db3b = os.path.join(tmp, "migrate_up.db") +db3b = _dsn("t3b") +reset_schema("t3b") init_db(db3b) uid = create_user("Round Trip", db_path=db3b) -conn = sqlite3.connect(db3b) +conn = get_db(db3b) conn.execute("INSERT INTO alarms (user_id, label, time) VALUES (?, 'test', '07:00')", (uid,)) conn.commit() conn.close() # simulate an old DB (version 0 → migration path) by rewriting meta -conn = sqlite3.connect(db3b) +conn = get_db(db3b) conn.execute("UPDATE meta SET version=0 WHERE id=1") conn.commit() conn.close() @@ -258,25 +307,61 @@ check("User data preserved through migration", _count(db3b, "alarms", uid) == 1) check("Migrated DB re-stamped to program version", _read_db_version(get_db(db3b)) == DB_VERSION) get_db(db3b).close() +# missing-column migration: drop a column, downgrade version, init adds it back +reset_schema("t3c") +db3c = _dsn("t3c") +init_db(db3c) +conn = get_db(db3c) +conn.execute("ALTER TABLE users DROP COLUMN theme") +conn.execute("UPDATE meta SET version=1 WHERE id=1") +conn.commit() +conn.close() +init_db(db3c) +conn = get_db(db3c) +theme_cols = conn.execute( + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema = current_schema() AND table_name='users' AND column_name='theme'").fetchall() +conn.close() +check("Missing column re-added by migration", len(theme_cols) == 1) +check("Migrated version stamped", _read_db_version(get_db(db3c)) == DB_VERSION) +get_db(db3c).close() + # a DB newer than the program → abort with SystemExit -newer = os.path.join(tmp, "newer.db") -init_db(newer) -conn = sqlite3.connect(newer) +reset_schema("t3d") +db3d = _dsn("t3d") +init_db(db3d) +conn = get_db(db3d) conn.execute(f"UPDATE meta SET version={DB_VERSION + 99} WHERE id=1") conn.commit() conn.close() try: - init_db(newer) + init_db(db3d) check("Newer DB aborts startup", False, "no SystemExit raised") except SystemExit: check("Newer DB aborts startup", True) +# legacy timestamp DEFAULT repair (old build used to_char HH:MM → wrong 12h/month) +reset_schema("t3e") +db3e = _dsn("t3e") +init_db(db3e) +conn = get_db(db3e) +conn.execute("ALTER TABLE users ALTER COLUMN created_at SET DEFAULT to_char(now(), 'YYYY-MM-DD HH:MM:SS')") +conn.commit() +conn.close() +init_db(db3e) # same version → fast path must still repair the DEFAULT +conn = get_db(db3e) +dflt = conn.execute( + "SELECT column_default FROM information_schema.columns " + "WHERE table_schema=current_schema() AND table_name='users' AND column_name='created_at'" +).fetchone()["column_default"] +conn.close() +check("Legacy timestamp DEFAULT repaired (HH24:MI)", "HH24:MI:SS" in (dflt or "")) + print() print("=" * 60) -print("TEST 4: MFA-configured DB migration → no new temp password") +print("TEST 4: MFA-configured DB → no new temp password") print("=" * 60) -db4 = os.path.join(tmp, "mfa.db") -app4 = create_app(db_path=db4) +app4, db4 = fresh_app("t4") c4 = root_client(app4, db4) _EMITTED_PASSWORDS.clear() init_db(db4) # same version → fast path, defaults intact @@ -288,8 +373,7 @@ print() print("=" * 60) print("TEST 5: Admin registration (admin-fills) + TOTP provisioning + login") print("=" * 60) -db5 = os.path.join(tmp, "adm.db") -app5 = create_app(db_path=db5) +app5, db5 = fresh_app("t5") root = root_client(app5, db5) root_secret = get_auth_config(db5, 0)["totp_secret"] # Reauth required first @@ -357,7 +441,7 @@ check("Account no longer temp", is_temp_account(temp_id, db5) is False) # security: the OLD temp access token must be dead immediately after finalize r = tu_client.post("/api/auth/mfa/complete") check("Old temp token dead after finalize (401)", r.status_code == 401) -conn = sqlite3.connect(db5) +conn = get_db(db5) n_temp = conn.execute("SELECT COUNT(*) FROM auth_tokens WHERE user_id=? AND temp=1", (temp_id,)).fetchone()[0] conn.close() check("No temp tokens remain after finalize", n_temp == 0) @@ -373,27 +457,7 @@ check("New TOTP login works", r.status_code == 200) print() print("=" * 60) -print("TEST 7: Self-service request → admin confirm → temp account") -print("=" * 60) -req_client = app5.test_client() -r = req_client.post("/api/auth/register-request", json={"remark": "Bob"}) -check("Public register request accepted", r.status_code == 201) -users_json = root.get("/api/admin/users").get_json() -pending = [u for u in users_json if u["status"] == "pending"] -check("Request appears as pending", len(pending) == 1 and pending[0]["remark"] == "Bob") -req_id = pending[0]["id"] -r = root.post("/api/admin/users/confirm", json={"request_id": req_id}) -check("Admin confirms request → temp account", r.status_code == 200) -check("Confirmed account is temp", is_temp_account(req_id, db5)) -r = root.post("/api/admin/users/confirm", json={"request_id": req_id}) -check("Double confirm rejected", r.status_code == 400) -bob_secret = get_temp_totp_secret(req_id, db5) -r = app5.test_client().post("/api/auth/login", json={"code": pyotp.TOTP(bob_secret).now(), "view": "control"}) -check("Bob can log in with temp TOTP", r.status_code == 200) - -print() -print("=" * 60) -print("TEST 8: Data isolation between users") +print("TEST 7: Data isolation between users") print("=" * 60) ua, _ = do_login(app5.test_client(), pyotp.TOTP(secret_a).now()) # User A creates data @@ -423,7 +487,7 @@ check("A sees own group events", len(r) == 1) print() print("=" * 60) -print("TEST 9: Edit remark + delete user + session invalidation") +print("TEST 8: Edit remark + delete user + session invalidation") print("=" * 60) # Fresh root session → no re-auth marker → admin ops must demand reauth. fresh_root, _ = do_login(app5.test_client(), pyotp.TOTP(root_secret).now()) @@ -450,7 +514,7 @@ check("Self-delete guard (root)", r.status_code == 400) print() print("=" * 60) -print("TEST 10: Kick devices (self account, all logged-in users) + Root kick-all-users") +print("TEST 9: Kick devices (self account, all logged-in users) + Root kick-all-users") print("=" * 60) # Two devices of the same account + one root device ua2, _ = do_login(app5.test_client(), pyotp.TOTP(secret_a).now()) @@ -474,10 +538,9 @@ check("Re-login after kick-all", r.status_code == 200) print() print("=" * 60) -print("TEST 11: Access/refresh token lifecycle (rotation, replay, refresh)") +print("TEST 10: Access/refresh token lifecycle (rotation, replay, refresh)") print("=" * 60) -db11 = os.path.join(tmp, "tok.db") -app11 = create_app(db_path=db11) +app11, db11 = fresh_app("t10") r11 = root_client(app11, db11) root11_secret = get_auth_config(db11, 0)["totp_secret"] # fresh login → access + refresh cookie @@ -503,7 +566,6 @@ r_again = rotate_refresh_token(_ref, db11) check("Replay of rotated refresh → None (single-use)", r_again is None) r = raw.post("/api/auth/refresh") check("Live refresh still works after rotation", r.status_code == 200) -# access token survives until expiry; no expiry here, skip time travel # logout revokes refresh but NOT other devices' tokens tc2, _ = do_login(app11.test_client(), pyotp.TOTP(root11_secret).now()) r = tc2.post("/api/auth/logout") @@ -516,7 +578,7 @@ check("Other device unaffected by logout", r.status_code == 200) print() print("=" * 60) -print("TEST 12: SSE ticket auth (one-time, scoped)") +print("TEST 11: SSE ticket auth (one-time, scoped)") print("=" * 60) # no ticket → 401 check("SSE without ticket → 401", app5.test_client().get("/api/events").status_code == 401) @@ -557,16 +619,7 @@ check("Global event reaches all subscribers", q_get(q2) is not None) print() print("=" * 60) -print("TEST 13: Rate limit on register-request") -print("=" * 60) -for i in range(5): - app5.test_client().post("/api/auth/register-request", json={"remark": f"spam{i}"}) -r = app5.test_client().post("/api/auth/register-request", json={"remark": "spam5"}) -check("6th register request rate-limited", r.status_code == 429) - -print() -print("=" * 60) -print("TEST 14: CSRF origin check on admin operations") +print("TEST 12: CSRF origin check on admin operations") print("=" * 60) root, _ = do_login(app5.test_client(), pyotp.TOTP(root_secret).now()) # re-login after kicks rr = reauth_totp(root, root_secret) @@ -585,7 +638,7 @@ check("Same-origin refresh allowed", r.status_code == 200) print() print("=" * 60) -print("TEST 15: Passkey flows (mocked ceremonies)") +print("TEST 13: Passkey flows (mocked ceremonies)") print("=" * 60) from backend import webauthn_mgr import backend.routes as routes_mod @@ -618,8 +671,7 @@ routes_mod.complete_authentication = lambda request, credential, stored_credenti "ok": True, "credential_id": _cred_hex(credential), } -db15 = os.path.join(tmp, "passkey.db") -app15 = create_app(db_path=db15) +app15, db15 = fresh_app("t13") root15 = root_client(app15, db15) root15_secret = get_auth_config(db15, 0)["totp_secret"] rr = reauth_totp(root15, root15_secret) @@ -672,14 +724,13 @@ routes_mod.complete_authentication = orig_complete_auth print() print("=" * 60) -print("TEST 16: Scheduler thread context (per-user timezone, no request context)") +print("TEST 14: Scheduler thread context (per-user timezone, no request context)") print("=" * 60) from backend.time_util import set_thread_user, clear_thread_user, get_tz_str, _ctx_user_id -db16 = os.path.join(tmp, "sched.db") -app16 = create_app(db_path=db16) +app16, db16 = fresh_app("t14") xid = create_user("Sched User", db_path=db16) -conn = sqlite3.connect(db16) +conn = get_db(db16) conn.execute("INSERT INTO clock_config (user_id, city, timezone, is_primary) VALUES (?, '', 'America/New_York', 1)", (xid,)) conn.commit() conn.close() @@ -711,7 +762,7 @@ check("Scheduler starts and stops cleanly", True) print() print("=" * 60) -print("TEST 17: Security hardening (SSRF, reauth gate, cascade delete, cross-user writes)") +print("TEST 15: Security hardening (SSRF, reauth gate, cascade delete, cross-user writes)") print("=" * 60) from backend.url_util import is_safe_public_url, is_safe_https_host check("SSRF: loopback URL rejected", is_safe_public_url("http://127.0.0.1/x.ics") is False) @@ -725,7 +776,7 @@ check("Host: public host accepted", is_safe_https_host("example.com") is True) root, _ = do_login(app5.test_client(), pyotp.TOTP(root_secret).now()) rr = reauth_totp(root, root_secret) ua, _ = do_login(app5.test_client(), pyotp.TOTP(secret_a).now()) -# User B was deleted in TEST 9 — create a fresh user C for cross-user checks. +# User B was deleted in TEST 8 — create a fresh user C for cross-user checks. rr = reauth_totp(root, root_secret) r = root.post("/api/admin/users", json={"remark": "User C", "mode": "admin_fills"}) uc_id = r.get_json()["user"]["id"] @@ -765,7 +816,7 @@ g = fc.post("/api/calendar/groups", json={"name": "g"}).get_json() fc.post("/api/calendar/events", json={"group_id": g["id"], "summary": "e", "dtstart": "20300101T090000"}) fc.post("/api/alarms", json={"label": "a", "time": "10:00"}) fc.post("/api/countdowns", json={"label": "c", "duration_seconds": 999}) -conn = sqlite3.connect(db5) +conn = get_db(db5) before = {t: conn.execute(f"SELECT COUNT(*) FROM {t} WHERE user_id=?", (fd_id,)).fetchone()[0] for t in ["calendar_groups", "calendar_sources", "calendar_events", "alarms", "countdowns", "auth_config", "pomodoro_state", "auth_tokens"]} conn.close() @@ -773,7 +824,7 @@ check("Cascade test user has data", sum(before.values()) >= 5) rr = reauth_totp(root, root_secret) r = root.delete(f"/api/admin/users/{fd_id}") check("Delete full-data user succeeds", r.status_code == 200) -conn = sqlite3.connect(db5) +conn = get_db(db5) after = {t: conn.execute(f"SELECT COUNT(*) FROM {t} WHERE user_id=?", (fd_id,)).fetchone()[0] for t in before} conn.close() check("All user data cascade-deleted (FK-safe)", sum(after.values()) == 0, str(after)) @@ -781,10 +832,9 @@ check("Deleted user's live tokens invalidated", fc.get("/api/alarms").status_cod print() print("=" * 60) -print("TEST 18: Passkey stash 一次性参数暂存（小窗传参协商）") +print("TEST 16: Passkey stash 一次性参数暂存（小窗传参协商）") print("=" * 60) -db18 = os.path.join(tmp, "stash.db") -app18 = create_app(db_path=db18) +app18, db18 = fresh_app("t16") c18 = app18.test_client() opts = {"challenge": "ch123", "rp": {"id": "d.example.com"}, "user": {"id": "u1", "name": "user_1"}} r = c18.post("/api/passkey/stash", json={"op": "creation", "options": opts}) @@ -829,10 +879,9 @@ check("读取参数后仍可 POST result（修复 410 回归）", r.status_code print() print("=" * 60) -print("TEST 19: 账户独立主题（dark/light/auto）") +print("TEST 17: 账户独立主题（dark/light/auto）") print("=" * 60) -db19 = os.path.join(tmp, "theme.db") -app19 = create_app(db_path=db19) +app19, db19 = fresh_app("t17") pw19 = get_temp_password(db19, 0) c19, _ = do_login(app19.test_client(), pw19) setup_totp(c19) @@ -861,10 +910,9 @@ check("未登录 GET /api/theme → 401", anon19.get("/api/theme").status_code = print() print("=" * 60) -print("TEST 20: 账户独立时区（display/state 返回 timezone + 时区设置生效）") +print("TEST 18: 账户独立时区（display/state 返回 timezone + 时区设置生效）") print("=" * 60) -db20 = os.path.join(tmp, "tz.db") -app20 = create_app(db_path=db20) +app20, db20 = fresh_app("t18") pw20 = get_temp_password(db20, 0) c20, _ = do_login(app20.test_client(), pw20) setup_totp(c20) @@ -884,11 +932,10 @@ r = c20.put("/api/time/config", json={"timezone": "Mars/Olympus"}) check("无效时区拒绝", r.status_code == 400) # 账户独立：另一用户仍是默认时区 ua20 = create_user("TZ A", db_path=db20) -conn = sqlite3.connect(db20) +conn = get_db(db20) conn.execute("UPDATE users SET status='active' WHERE id=?", (ua20,)) conn.close() # 给用户 A 配 TOTP 再登录 -from backend.auth import generate_temp_password uac(ua20, totp_secret="JBSWY3DPEHPK3PXP", totp_enabled=1, _db_path=db20) a20, _ = do_login(app20.test_client(), pyotp.TOTP("JBSWY3DPEHPK3PXP").now()) st = a20.get("/api/display/state").get_json() @@ -896,7 +943,7 @@ check("用户 B 时区保持默认", st.get("timezone") == "Asia/Shanghai") print() print("=" * 60) -print("TEST 21: 页面导航鉴权（refresh cookie，浏览器导航不带 Authorization）") +print("TEST 19: 页面导航鉴权（refresh cookie，浏览器导航不带 Authorization）") print("=" * 60) # 模拟真实浏览器：登录后导航页面（无 Authorization header，仅 cookie） raw_nav = app5.test_client() @@ -913,10 +960,14 @@ anon_nav = app5.test_client() check("未登录 /control → 302", anon_nav.get("/control").status_code == 302) check("未登录 / → 302", anon_nav.get("/").status_code == 302) check("未登录 /mfa-setup → 302", anon_nav.get("/mfa-setup").status_code == 302) -# temp 用户（Bob 仍是临时账户）导航 +# temp 用户导航（用 TEST 6 的 temp 账户，仍是临时账户 → 强制 MFA 设置页） +rr = reauth_totp(root, root_secret) +r = root.post("/api/admin/users", json={"remark": "Nav Temp", "mode": "user_fills"}) +nav_temp_id = r.get_json()["user"]["id"] +nav_temp_secret = get_temp_totp_secret(nav_temp_id, db5) bob_raw = app5.test_client() -r = bob_raw.post("/api/auth/login", json={"code": pyotp.TOTP(bob_secret).now(), "view": "control"}) -check("Bob temp 登录", r.status_code == 200) +r = bob_raw.post("/api/auth/login", json={"code": pyotp.TOTP(nav_temp_secret).now(), "view": "control"}) +check("temp 用户登录", r.status_code == 200) check("temp 用户 /mfa-setup → 200", bob_raw.get("/mfa-setup").status_code == 200) loc = bob_raw.get("/control").headers.get("Location", "") check("temp 用户 /control → 302 mfa-setup", "/mfa-setup" in loc) @@ -925,6 +976,103 @@ r = bob_raw.post("/api/auth/logout") check("登出 ok", r.status_code == 200) check("登出后 /mfa-setup → 302", bob_raw.get("/mfa-setup").status_code == 302) +print() +print("=" * 60) +print("TEST 20: 数据库管理（仅 Root、导出/导入 roundtrip、格式识别、原子性）") +print("=" * 60) +app21, db21 = fresh_app("t20") +root21 = root_client(app21, db21) +root21_secret = get_auth_config(db21, 0)["totp_secret"] + +# 造点数据 +c21 = app21.test_client() +r = c21.post("/api/auth/login", json={"code": pyotp.TOTP(root21_secret).now(), "view": "control"}) +root21 = TClient(c21, r.get_json()["access_token"]) +r = root21.post("/api/calendar/groups", json={"name": "DB Test"}) +check("Setup: group created", r.status_code == 201) +r = root21.post("/api/alarms", json={"label": "DB alarm", "time": "07:30"}) +check("Setup: alarm created", r.status_code == 201) + +# 非 root 用户（普通 admin 角色也拒绝——仅 root id 0） +rr = reauth_totp(root21, root21_secret) +r = root21.post("/api/admin/users", json={"remark": "DbUser", "mode": "admin_fills"}) +dbu_id = r.get_json()["user"]["id"] +r = root21.post(f"/api/admin/users/{dbu_id}/totp/begin") +dbu_secret = r.get_json()["secret"] +root21.post(f"/api/admin/users/{dbu_id}/totp/verify", json={"code": pyotp.TOTP(dbu_secret).now()}) +dbu, _ = do_login(app21.test_client(), pyotp.TOTP(dbu_secret).now()) +r = dbu.get("/api/admin/db/status") +check("非 root 访问 db 管理 → 403", r.status_code == 403) +r = dbu.get("/api/admin/db/export") +check("非 root 导出 → 403", r.status_code == 403) + +# 未二次验证 → 403（用全新的 root 会话，无 reauth 标记） +fresh_root21, _ = do_login(app21.test_client(), pyotp.TOTP(root21_secret).now()) +r = fresh_root21.get("/api/admin/db/export") +check("导出未 reauth → 403", r.status_code == 403 and r.get_json().get("reauth_required")) + +# 状态 +r = root21.get("/api/admin/db/status") +d = r.get_json() +check("db status 返回 PostgreSQL v" + str(DB_VERSION), d.get("database") == "postgresql" and d.get("db_version") == DB_VERSION) + +# 导出（reauth 后） +rr = reauth_totp(root21, root21_secret) +r = root21.get("/api/admin/db/export") +check("Root 导出成功", r.status_code == 200 and "json" in r.content_type) +backup = json.loads(r.data) +check("备份格式正确", backup.get("format") == "iot-dboard-backup") +check("备份含用户数据", len(backup["tables"]["calendar_groups"]["rows"]) >= 1 + and len(backup["tables"]["alarms"]["rows"]) >= 1) +check("备份不含会话令牌", "auth_tokens" not in backup["tables"]) + +# 格式识别 +check("SQLite magic 识别", db_admin.detect_format(b"SQLite format 3\x00" + b"x" * 8) == "sqlite") +check("JSON 备份识别", db_admin.detect_format(json.dumps({"format": "iot-dboard-backup"}).encode()) == "json") +check("未知格式识别", db_admin.detect_format(b"garbage") == "unknown") + +# 导入备份 roundtrip（导入会清空会话 → root 需重登） +r = root21.post("/api/admin/db/import", data={"file": (io.BytesIO(json.dumps(backup).encode()), "b.json")}, + content_type="multipart/form-data") +check("导入备份成功", r.status_code == 200 and r.get_json().get("ok"), str(r.get_json())) +check("导入后会话失效标记", r.get_json().get("session_invalidated") is True) +conn = get_db(db21) +n = conn.execute("SELECT COUNT(*) FROM calendar_groups WHERE name='DB Test'").fetchone()[0] +conn.close() +check("导入后数据恢复", n >= 1) +# 导入后旧 token 失效 +check("导入后旧会话失效 → 401", root21.get("/api/admin/db/status").status_code == 401) + +# 导入非法文件 → 400（不破坏数据） +root21b, _ = do_login(app21.test_client(), pyotp.TOTP(root21_secret).now()) +rr = reauth_totp(root21b, root21_secret) +r = root21b.post("/api/admin/db/import", data={"file": (io.BytesIO(b"junk"), "x.bin")}, + content_type="multipart/form-data") +check("导入非法文件 → 400", r.status_code == 400, str(r.get_json())) +conn = get_db(db21) +n = conn.execute("SELECT COUNT(*) FROM calendar_groups WHERE name='DB Test'").fetchone()[0] +conn.close() +check("非法导入未破坏数据", n >= 1) + +# 导入失败原子性：构造一个会导致失败的备份（缺少必需列 → 行插入失败回滚） +root21c, _ = do_login(app21.test_client(), pyotp.TOTP(root21_secret).now()) +rr = reauth_totp(root21c, root21_secret) +bad = { + "format": "iot-dboard-backup", + "db_version": DB_VERSION, + "tables": {"calendar_groups": {"columns": ["id", "user_id", "name", "color", "enabled", "created_at"], + "rows": [[1, 0, "OK Group", "#fff", 1, "2026-01-01 00:00:00"], + [2, 0, None, "#fff", 1, None]]}}, # name NOT NULL → None 会失败 +} +# 直接把 name=None 塞进去会因参数为 None 失败（psycopg2 拒绝 None 绑定到 NOT NULL？其实 None → NULL，NOT NULL 约束报错） +r = root21c.post("/api/admin/db/import", data={"file": (io.BytesIO(json.dumps(bad).encode()), "bad.json")}, + content_type="multipart/form-data") +check("含非法行的导入 → 400/500", r.status_code in (400, 500), str(r.get_json())) +conn = get_db(db21) +n = conn.execute("SELECT COUNT(*) FROM calendar_groups WHERE name='OK Group'").fetchone()[0] +conn.close() +check("失败导入整体回滚（原子性）", n == 0) + print() print("=" * 60) print(f"RESULT: {PASS} passed, {FAIL} failed")