workbuddy-portal/workbuddy_portal/config.py

# -*- coding: utf-8 -*-
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 Wang Chuanli

"""基础配置。

刻意保持「薄」：凡是运行期要改的东西（cookie、调度周期、采集参数）都放数据库
settings 表，由后台页面维护；这里只放路径、密钥、默认值这类启动期常量。
"""
import json
import os
import secrets

# ---------------- 项目标识（单一来源）----------------
# 仓库名 / 镜像名 / compose 服务名 / 界面品牌都从这里取，避免多处硬编码走样。
PROJECT_NAME = "workbuddy-portal"          # 技术标识：目录、仓库、镜像名
PROJECT_TITLE = "WorkBuddy Portal"         # 界面品牌
PROJECT_DESC = "WorkBuddy 积分用量采集 / 存储 / 呈现一体化门户"

# 项目根（workbuddy-portal/）
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

# 数据 / 日志目录可用环境变量覆盖（容器里把卷挂到别处时不必改代码）。
DATA_DIR = os.environ.get("WB_DATA_DIR") or os.path.join(BASE_DIR, "data")
LOG_DIR = os.environ.get("WB_LOG_DIR") or os.path.join(BASE_DIR, "logs")
SQLITE_PATH = os.environ.get("WB_DB") or os.path.join(DATA_DIR, "usage.sqlite")
EXPORT_DIR = os.path.join(DATA_DIR, "exports")
APP_LOG = os.path.join(LOG_DIR, "app.log")
INSTANCE_FILE = os.path.join(DATA_DIR, "instance.json")

# 备份落点。**刻意与 DATA_DIR 分开**：
#   * 容器里 DATA_DIR 挂的是数据卷，`docker compose down -v` 会连卷一起删；
#     备份若躺在同一个卷里，就等于「正本与副本同时消失」—— 备份的意义没了。
#   * 备份里含 settings 的凭证密文与 users 的口令散列，必须能单独控制权限、
#     单独挂卷、单独排除出镜像（见 .dockerignore 的 backups/）。
BACKUP_DIR = os.environ.get("WB_BACKUP_DIR") or os.path.join(BASE_DIR, "backups")

# 旧版脚本项目的存档（迁移用；--migrate-csv 默认读这里）。
# v1.3.0 起旧版被收进工作区级的 legacy-v1/ 目录，所以第一个候选是新位置，
# 后面两个保留以兼容「还没挪走」的部署。
LEGACY_CSV_CANDIDATES = [
    os.path.join(os.path.dirname(BASE_DIR), "legacy-v1", "data", "usage_records.csv"),
    os.path.join(os.path.dirname(BASE_DIR), "data", "usage_records.csv"),
    os.path.join(BASE_DIR, "data", "usage_records.csv"),
]

# ---------------- 云端接口 ----------------
API_BASE = "https://www.workbuddy.cn"
API_PATH = "/billing/meter/get-user-request-usage"

# ---------------- 采集参数默认值（可被 settings 表覆盖）----------------
# settings 表是 (user_id, key) 复合主键：user_id=0 表示**实例级**，
# 其余表示**个人级**（每个账号一份，互不可见）。见下方的 GLOBAL_KEYS。
DEFAULTS = {
    # ---- 实例级：连接的是哪个云端 ----
    "api_base": API_BASE,
    "api_path": API_PATH,
    # ---- 实例级：开放注册与防攻击策略 ----
    "allow_register": "1",            # 是否开放自助注册
    "register_max_per_ip": "3",       # 同一 IP 每天最多注册几个账号
    "captcha_policy": "always",       # always | adaptive | off（见 CAPTCHA_POLICIES）
    "captcha_length": "4",            # 验证码字符数 4~6
    # ---- 实例级：采集调度（全实例统一，见 GLOBAL_KEYS）----
    "schedule_enabled": "1",
    "schedule_times": "09:00,17:00",   # 每天固定时刻（逗号分隔，本地时区）
    "catch_up": "1",                    # 启动时补跑当天已错过且未执行的槽位
    "catch_up_grace_hours": "12",       # 超过该小时数就不再补跑
    "max_schedule_slots_per_day": "6",  # 每天最多几个时刻（挡住「填 200 个时刻」）
    "collect_min_interval_seconds": "60",  # 同一账号两次手动采集的最小间隔
    "collect_max_range_days": "31",     # 单次采集的最长跨度（硬顶 31 天 = 1 个月）
    # ---- 实例级：自动备份 ----
    "backup_enabled": "1",              # 是否开启自动备份
    "backup_interval_hours": "24",      # 备份周期（小时）
    "backup_keep": "7",                 # 保留最近几份，超出的自动删除最旧的
    # ---- 实例级：采集参数 ----
    "page_size": "200",
    "rewind_minutes": "2",          # 断点回退分钟数
    "drift_tolerance_minutes": "5",  # 云端比本地早超过该值才告警
    "max_prompt": "2048",           # 0 表示不截断
    "verify_days": "0",             # 每次采集后做整日完整性校验的天数
    "timeout": "30",
    "ssl_verify": "1",              # 校验云端 HTTPS 证书（cookie 是凭证，不该裸奔）
    # ---- 个人级：凭证（每个账号自己的，Cookie 静态加密后入库）----
    "cookie": "",
    "user_agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                   "(KHTML, like Gecko) Chrome/153.0.0.0 Safari/537.36"),
}

# ---------------- 配置的作用域与写权限（改这里之前先读完整段）----------------
#
# 三级回落：**个人级(user_id=n) -> 实例级(user_id=0) -> config.DEFAULTS**。
# 但「谁能写哪个键」和「键存在哪一级」是两件事，本项目把两者对齐成一条规则：
#
#   普通用户只能写 USER_EDITABLE_KEYS（本人凭证）；
#   其余所有键都归管理员，且一律存在**实例级**（user_id=0）。
#
# 为什么采集参数 / 调度也要放到实例级，而不是「个人级但只有管理员能写」：
#   如果它们只在管理员自己的 user_id 下，其它账号读取时会回落到
#   DEFAULTS，管理员改的值对别人**完全不生效** —— 那才是真正的坑。
#   统一放实例级，语义是「一台部署一套采集与调度策略」，读起来也简单。
GLOBAL_KEYS = {
    # 云端接口
    "api_base", "api_path",
    # 开放注册与防攻击策略
    "allow_register", "register_max_per_ip", "captcha_policy", "captcha_length",
    # 采集调度（v1.3.0 起为实例级：普通用户只读，不能设置频率）
    "schedule_enabled", "schedule_times", "catch_up", "catch_up_grace_hours",
    "max_schedule_slots_per_day",
    # 任务频率与采集跨度（v1.4.0 起：对外提供服务时必须能限流，
    # 否则一个注册账号就能拿 /api/collect 把云端与线程池打满）
    "collect_min_interval_seconds", "collect_max_range_days",
    # 自动备份（谁掌握备份谁就掌握全库数据，所以归管理员）
    "backup_enabled", "backup_interval_hours", "backup_keep",
    # 采集参数（同理：允许普通用户调 page_size/关 ssl_verify 都是越权）
    "page_size", "rewind_minutes", "drift_tolerance_minutes", "max_prompt",
    "verify_days", "timeout", "ssl_verify",
}

# 普通用户**唯一**可写的两个键：本人账号的云端凭证。
#   cookie      —— 静态加密后入库，页面/接口只回掩码
#   user_agent  —— 必须与拿 Cookie 的那次请求同源，所以和 Cookie 归在一起
# 这两个键是个人级、且不参与实例级回落（见 db.NO_FALLBACK_KEYS）：
# 回落等于「用别人的身份采集」，是最严重的一类越权。
USER_EDITABLE_KEYS = {"cookie", "user_agent"}


def is_user_key(key):
    """是否属于「个人凭证」类配置。"""
    return key in USER_EDITABLE_KEYS


def writable_by(key, is_admin):
    """当前角色能否写这个键 —— 前后端与测试都走这一个判断，避免两处规则漂移。"""
    if key in USER_EDITABLE_KEYS:
        return True
    return bool(is_admin)

# 验证码策略
CAPTCHA_POLICIES = {
    "always":   "始终要求（默认，最安全）",
    "adaptive": "仅在同一来源连续失败 2 次后要求",
    "off":      "关闭（仅当前面有可信网关做鉴权时才考虑）",
}

# 页面展示用：哪些键属于「敏感」，在界面上做掩码
SECRET_KEYS = {"cookie"}
# 需要静态加密后再入库的键（明文只存在于内存与请求体里）
ENCRYPTED_KEYS = {"cookie"}

# 内部簿记键前缀：调度槽位标记等，**不属于用户可配置项**，
# 不在 /api/settings 里回传，也不允许通过接口写入。
INTERNAL_PREFIXES = ("slot:",)


def is_internal_key(key):
    return any(str(key).startswith(p) for p in INTERNAL_PREFIXES)


def is_global_key(key):
    """实例级键：所有账号共用一份，只有管理员可写。"""
    return key in GLOBAL_KEYS


# ---------------- 设置项校验表 ----------------
# 这些键必须能安全地转成数字：后台页面是自由文本框，用户敲错一个字符
# 就会让采集在 int() 处抛 ValueError（历史 bug），所以写入时校验、读取时兜底。
# 值 = (最小值, 最大值, 单位说明)
NUM_SETTINGS = {
    "page_size":               (20, 1000, "条/页"),
    "rewind_minutes":          (0, 120, "分钟"),
    "drift_tolerance_minutes": (0, 720, "分钟"),
    "max_prompt":              (0, 20000, "字符"),
    "verify_days":             (0, 90, "天"),
    "timeout":                 (5, 300, "秒"),
    "catch_up_grace_hours":    (1, 168, "小时"),
    "captcha_length":          (4, 6, "个字符"),
    "register_max_per_ip":     (1, 50, "个/天"),
    "max_schedule_slots_per_day":    (1, 12, "个/天"),
    "collect_min_interval_seconds":  (0, 3600, "秒"),
    "collect_max_range_days":        (1, 31, "天"),
    "backup_interval_hours":         (1, 720, "小时"),
    "backup_keep":                   (1, 100, "份"),
}
BOOL_SETTINGS = {"schedule_enabled", "catch_up", "allow_register", "backup_enabled"}

# 采集跨度的**硬顶**：无论 settings 里被改成什么（含历史脏数据、手工改库），
# 代码层一律按这个上限夹一次。写进配置只是给管理员一个更严的旋钮，
# 不是「改大就能突破」——上限必须由代码兜底，不能只靠校验。
COLLECT_MAX_RANGE_DAYS_HARD = 31
# 每日调度时刻的硬顶（同上）
SCHEDULE_SLOTS_HARD_MAX = 12

_TRUE = ("1", "true", "yes", "on", "是", "启用")


def normalize_setting(key, raw, conn=None):
    """校验并规范化单个设置值。

    返回 (value, error)：
      * value 为可直接写入 settings 表的字符串；error 非空时 value 为 None。
      * 未知键（不在 DEFAULTS 里）直接拒绝，避免接口被用来写任意键。
      * `conn` 可选：个别键的上限本身是可配置的（如每日时刻数受
        `max_schedule_slots_per_day` 约束），有连接时才查得到。
        不传时只做代码层的硬顶校验，所以离线调用不会因此失败。
    """
    if key not in DEFAULTS:
        return None, "未知配置项：%s" % key
    if raw is None:
        return "", None

    if key in BOOL_SETTINGS:
        v = str(raw).strip().lower()
        if v in _TRUE:
            return "1", None
        if v in ("0", "false", "no", "off", "否", "停用", ""):
            return "0", None
        return None, "%s 只能是 0/1" % key

    if key in NUM_SETTINGS:
        lo, hi, unit = NUM_SETTINGS[key]
        try:
            n = int(float(str(raw).strip()))
        except (TypeError, ValueError):
            return None, "%s 必须是数字（%s）" % (key, unit)
        if not (lo <= n <= hi):
            return None, "%s 需在 %d ~ %d %s 之间" % (key, lo, hi, unit)
        return str(n), None

    if key == "schedule_times":
        from . import scheduler      # 局部导入避免循环依赖
        parsed = scheduler.parse_times(raw)
        if not parsed:
            return None, "每日时刻格式不对，正确写法如 09:00,17:00"
        if len(parsed) > SCHEDULE_SLOTS_HARD_MAX:
            return None, "每日时刻最多 %d 个（当前填了 %d 个）" % (
                SCHEDULE_SLOTS_HARD_MAX, len(parsed))
        # 可配置的更严上限：时刻数量直接决定调度器的采集频次，
        # 是「一个账号能不能把云端与线程池打满」的开关，所以要有刹车。
        if conn is not None:
            from . import db as _db
            cap = _db.get_int(conn, "max_schedule_slots_per_day", 6)
            cap = max(1, min(SCHEDULE_SLOTS_HARD_MAX, cap))
            if len(parsed) > cap:
                return None, ("每日时刻最多 %d 个（当前填了 %d 个）。"
                              "如需更多，请先把「每日调度时刻上限」调大。"
                              % (cap, len(parsed)))
        return ",".join(parsed), None

    if key == "captcha_policy":
        v = str(raw).strip().lower()
        if v not in CAPTCHA_POLICIES:
            return None, "验证码策略只能是 %s" % " / ".join(sorted(CAPTCHA_POLICIES))
        return v, None

    if key in ("api_base", "api_path"):
        v = str(raw).strip()
        if not v:
            return None, "%s 不能为空" % key
        if key == "api_base":
            if not v.startswith(("http://", "https://")):
                return None, "接口基址需以 http:// 或 https:// 开头"
            # 云元数据地址永远不该是「云端接口」：它是 SSRF 拿云上临时凭证
            # 最经典的一跳，而且没有任何合法的采集场景需要它。
            host = v.split("//", 1)[1].split("/", 1)[0].split(":")[0].lower()
            if host in ("169.254.169.254", "metadata.google.internal", "[fd00:ec2::254]"):
                return None, "接口基址不能指向云元数据地址"
        return v, None

    if key == "cookie":
        # 明文原样返回，由 db.set_setting 负责加密后再落库
        return str(raw).strip(), None

    return str(raw).strip(), None

# 服务
DEFAULT_HOST = "0.0.0.0"      # 局域网可访问
DEFAULT_PORT = 8848
SESSION_HOURS = 12
MAX_LOGIN_FAILS = 5           # 同 IP 连续失败次数（硬锁）
LOGIN_LOCK_MINUTES = 10       # 硬锁时长（仅 IP 维度）
# 用户名维度的**软退避**：对外提供服务后，「知道一个用户名就能把它锁死 10 分钟」
# 本身就是一种攻击（拿管理员用户名当武器，别人也用不了）。所以用户名维度
# 只产生秒级、递增、有封顶的等待，真正的重锁只按来源 IP 施加。
USER_SOFT_THRESHOLD = 5       # 同一用户名失败超过这个次数才开始退避
USER_SOFT_CAP_SECONDS = 60    # 退避封顶
# 同一来源的登录尝试总量（含成功）：挡住「慢慢撞、不触发失败阈值」的形态
LOGIN_ATTEMPTS_PER_IP = 40
LOGIN_ATTEMPTS_WINDOW = 300   # 秒

# ---------------- 反向代理与传输安全 ----------------
def _env_flag(name, default="0"):
    return os.environ.get(name, default).strip().lower() in ("1", "true", "yes", "on")


# 是否信任 X-Forwarded-For。**默认不信任**。
# 直接暴露给公网（或前面只有一个「追加型」代理）时，XFF 的第 0 段是攻击者
# 自己填的：一旦采信，验证码限速、注册配额、登录锁定三道 IP 防线会同时失效
# （实测：每次换一个伪造 XFF，45 次验证码请求全部放行）。
# 只有在**你自己的**反向代理会重写该头（nginx: `$remote_addr`）时才置 1。
TRUST_PROXY = _env_flag("WB_TRUST_PROXY")
# 强制跳转 HTTPS（配合反代时用；读到 X-Forwarded-Proto: https 就不跳）
FORCE_HTTPS = _env_flag("WB_FORCE_HTTPS")
# 会话 Cookie 是否只走 HTTPS。纯局域网 HTTP 部署必须留 0，否则浏览器不发送，
# 表现为「登录成功但立刻又跳回登录页」，极难排查。
COOKIE_SECURE = _env_flag("WB_COOKIE_SECURE")

# 访问日志：waitress 自己不记 access log，上线后没有访问日志等于出事无据可查。
# 只记非静态资源请求，写进 logs/app.log（滚动 2MB × 3）。
ACCESS_LOG = _env_flag("WB_ACCESS_LOG", "1")

# waitress 线程数。与容器 CPU 上限配套：线程越多，单实例能同时吃进的
# 慢请求（如采集、导出）就越多 —— 对外提供服务时这是资源上限的一部分。
THREADS = int(os.environ.get("WB_THREADS") or 8)

# 首个管理员的初始口令。**不给默认值**：留空时 db.init_db 会生成一个随机口令
# 并只在启动日志里打印一次 —— 硬编码一个 admin123 等于把公网实例的钥匙挂在门上。
ADMIN_USER = os.environ.get("WB_ADMIN_USER") or "admin"
ADMIN_PASSWORD = os.environ.get("WB_ADMIN_PASSWORD") or ""

# 口令黑名单：这些是自动撞库字典的头几页，命中即拒。
# 只在「设置/修改口令」时校验（登录不校验），所以不会把用老口令的人挡在门外。
WEAK_PASSWORDS = {
    "12345678", "123456789", "1234567890", "password", "password1", "password123",
    "passw0rd", "qwertyui", "qwerty123", "abc12345", "abcd1234", "admin123",
    "admin888", "admin1234", "administrator", "root1234", "letmein1", "welcome1",
    "iloveyou", "monkey123", "dragon123", "sunshine", "princess", "football",
    "baseball", "11111111", "00000000", "88888888", "66666666", "asdasd123",
    "1qaz2wsx", "zxcvbnm1", "a1234567", "workbuddy", "codebuddy",
}

# ---------------- 账号与口令策略 ----------------
USERNAME_RE = r"^[A-Za-z0-9][A-Za-z0-9_.\-]{2,31}$"   # 3~32 位，字母开头
PASSWORD_MIN = 8
PASSWORD_MAX = 128
# 开启注册后，未配置 Cookie 的新账号在概览页会被提示「去配置」——
# 采集只使用**本人**的 Cookie，绝不复用别人的（否则会串号）。
PROFILE_EMAIL_MAX = 128

# 会话 Cookie 的 Secure 开关在文件上方的「反向代理与传输安全」段，
# 与 TRUST_PROXY / FORCE_HTTPS 放在一起 —— 这三个必须一起决定，
# 拆开写很容易出现「开了强制 HTTPS 却忘了 Secure」这类半截配置。


def ensure_dirs():
    for d in (DATA_DIR, LOG_DIR, EXPORT_DIR, BACKUP_DIR):
        os.makedirs(d, exist_ok=True)


def _instance_read():
    """读 data/instance.json（不存在或损坏都当空字典，不让启动因此失败）。"""
    ensure_dirs()
    if not os.path.exists(INSTANCE_FILE):
        return {}
    try:
        with open(INSTANCE_FILE, "r", encoding="utf-8") as f:
            return json.load(f) or {}
    except (OSError, ValueError):
        return {}


def _instance_init(key, maker):
    """取 instance.json 里的 key，没有就生成并持久化。"""
    data = _instance_read()
    val = data.get(key)
    if not val:
        val = maker()
        data[key] = val
        try:
            with open(INSTANCE_FILE, "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            # 这个文件里躺着 SECRET_KEY 与 cookie_key，权限等同管理员口令。
            # 默认 umask 022 会留下 0644（同机其他用户可读），所以在 POSIX 上
            # 显式收紧到 0600。Windows 没有这个概念，忽略即可。
            if os.name == "posix":
                os.chmod(INSTANCE_FILE, 0o600)
        except OSError:
            pass                      # 只读文件系统时退化为「本次进程内有效」
    return val


def secret_key():
    """SECRET_KEY 持久化在 data/instance.json，避免每次重启把登录态全踢掉。"""
    return _instance_init("secret_key", lambda: secrets.token_hex(32))


def encryption_key():
    """Cookie 静态加密的主密钥（32 字节）。

    与 SECRET_KEY **分开**存放：两者轮换的代价完全不同 —— 换 SECRET_KEY
    只是让所有人重新登录，换这把会让已存的 Cookie 全部解不开。
    所以混用同一个值会让「想轮换其中一个」变成一件危险的事。
    """
    return bytes.fromhex(_instance_init("cookie_key", lambda: secrets.token_hex(32)))