MeshSate/app/config.py

"""应用配置加载。

所有配置项均以 ``MESHSAFE_`` 前缀从环境变量读取，敏感项不提供默认值，
缺失时启动即报错，避免出现"用默认弱口令上线"的情况。
"""

from __future__ import annotations

import base64
import functools
from typing import Dict, List, Optional

from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="MESHSAFE_",
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    # 运行环境
    env: str = "production"
    log_level: str = "INFO"
    timezone: str = "Asia/Shanghai"

    # 服务监听
    host: str = "0.0.0.0"
    port: int = 8443
    root_path: str = "/"

    # 基础设施
    db_url: str
    db_echo: bool = False
    redis_url: str = "redis://redis:6379/0"

    # 主密钥
    master_key: str

    # API 调用方
    api_clients: str = ""
    signature_window: int = 300
    require_signature: bool = True
    sign_response: bool = True

    # 控制台
    admin_username: str = "admin"
    admin_password: str
    login_max_attempts: int = 5
    login_lock_minutes: int = 15
    session_timeout_minutes: int = 30

    # ClamAV
    clamav_host: str = "clamav"
    clamav_port: int = 3310
    clamav_timeout: int = 120
    clamav_use_clamd: bool = True

    # 策略
    policy_file: str = "/app/config/policy.yaml"
    rules_dir: str = "/app/rules"
    sandbox_enabled: bool = False

    # 文件处理
    max_upload_mb: int = 2048
    max_extract_depth: int = 5
    max_extract_total_mb: int = 4096
    max_extract_ratio: int = 100
    temp_ttl_minutes: int = 60

    # 合规
    audit_retention_days: int = 180
    audit_hash_chain: bool = True
    report_retention_days: int = 180

    # TLS
    enable_tls: bool = False
    tls_cert_file: str = "/app/certs/server.crt"
    tls_key_file: str = "/app/certs/server.key"
    trusted_proxy_count: int = 1

    @field_validator("master_key")
    @classmethod
    def _validate_master_key(cls, v: str) -> str:
        try:
            raw = base64.b64decode(v, validate=True)
        except Exception as exc:  # noqa: BLE001
            raise ValueError("MESHSAFE_MASTER_KEY 必须是合法的 base64 字符串") from exc
        if len(raw) < 32:
            raise ValueError("MESHSAFE_MASTER_KEY 解码后长度不得少于 32 字节")
        return v

    @field_validator("policy_file")
    @classmethod
    def _normalize_policy_file(cls, v: str) -> str:
        return v.strip()

    @property
    def master_key_bytes(self) -> bytes:
        return base64.b64decode(self.master_key, validate=True)

    @property
    def max_upload_bytes(self) -> int:
        return self.max_upload_mb * 1024 * 1024

    @property
    def max_extract_total_bytes(self) -> int:
        return self.max_extract_total_mb * 1024 * 1024

    @property
    def parsed_api_clients(self) -> List[Dict[str, str]]:
        """解析 ``client_id:secret,client_id2:secret2`` 形式的调用方清单。"""
        result: List[Dict[str, str]] = []
        for item in self.api_clients.split(","):
            item = item.strip()
            if not item:
                continue
            if ":" not in item:
                raise ValueError(f"非法的 API 调用方配置项: {item!r}")
            client_id, secret = item.split(":", 1)
            client_id = client_id.strip()
            secret = secret.strip()
            if not client_id or not secret:
                raise ValueError(f"非法的 API 调用方配置项: {item!r}")
            result.append({"client_id": client_id, "secret": secret})
        return result

    def redacted(self) -> Dict[str, object]:
        """用于 /health 等接口展示的脱敏配置快照。"""
        return {
            "env": self.env,
            "require_signature": self.require_signature,
            "sign_response": self.sign_response,
            "signature_window": self.signature_window,
            "max_upload_mb": self.max_upload_mb,
            "max_extract_depth": self.max_extract_depth,
            "max_extract_ratio": self.max_extract_ratio,
            "sandbox_enabled": self.sandbox_enabled,
            "audit_retention_days": self.audit_retention_days,
            "audit_hash_chain": self.audit_hash_chain,
            "api_clients_count": len(self.parsed_api_clients),
        }


@functools.lru_cache(maxsize=1)
def get_settings() -> Settings:
    return Settings()  # type: ignore[call-arg]