mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-02 13:28:32 +08:00
feat(config): support deployment secrets without persisting values
This commit is contained in:
jxxghp
@@ -12,6 +12,7 @@ import shutil
|
||||
import sys
|
||||
import threading
|
||||
from asyncio import AbstractEventLoop
|
||||
from contextvars import ContextVar
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import (
|
||||
@@ -29,8 +30,13 @@ from typing import (
|
||||
from urllib.parse import quote, urlencode, urlparse
|
||||
|
||||
from dotenv import set_key, unset_key
|
||||
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, TypeAdapter, model_validator
|
||||
from pydantic_settings import (
|
||||
BaseSettings,
|
||||
PydanticBaseSettingsSource,
|
||||
SecretsSettingsSource,
|
||||
SettingsConfigDict,
|
||||
)
|
||||
|
||||
from app.foundation.environment import (
|
||||
cpu_arch,
|
||||
@@ -770,26 +776,142 @@ class ConfigModel(BaseModel):
|
||||
AUDIO_OUTPUT_INCLUDE_TEXT: bool = Field(default=False, description='回复语音时是否同时附带文字说明')
|
||||
|
||||
|
||||
# 每次构造独立记录外部文件字段，避免并发构造配置时互相污染写回保护。
|
||||
_secret_fields_context: ContextVar[Optional[set[str]]] = ContextVar(
|
||||
"settings_secret_fields", default=None
|
||||
)
|
||||
|
||||
|
||||
def _get_secrets_dir() -> Optional[str]:
|
||||
"""
|
||||
从进程环境选择 secrets 目录；显式空值或 false 关闭文件来源。
|
||||
|
||||
:return: secrets 目录路径；关闭时返回 None
|
||||
"""
|
||||
value = os.environ.get(
|
||||
"MOVIEPILOT_SECRETS_DIR", os.environ.get("SECRETS_DIR", "/run/secrets")
|
||||
).strip()
|
||||
return None if value.lower() in {"", "false", "off", "0"} else value
|
||||
|
||||
|
||||
# 严格类型检查跳过第三方源码导入，因此无法静态解析 Pydantic 来源基类。
|
||||
class _DeploymentSecretsSource(SecretsSettingsSource): # type: ignore[misc]
|
||||
"""
|
||||
按配置名读取外部文件，隔离坏文件并保护编排管理的字段不被写回。
|
||||
|
||||
复用 Pydantic 的来源顺序和复杂值解析；只记录字段名，不缓存或记录原文。
|
||||
"""
|
||||
|
||||
def __call__(self) -> dict[str, Any]:
|
||||
"""
|
||||
逐项读取并校验 secret，缺失目录静默跳过，其余错误只告警字段名。
|
||||
|
||||
:return: 通过字段类型与约束校验的配置值
|
||||
"""
|
||||
values: dict[str, Any] = {}
|
||||
if self.secrets_dir is None:
|
||||
return values
|
||||
directories = (
|
||||
[self.secrets_dir]
|
||||
if isinstance(self.secrets_dir, (str, os.PathLike))
|
||||
else self.secrets_dir
|
||||
)
|
||||
for directory in directories:
|
||||
path = Path(directory).expanduser()
|
||||
try:
|
||||
if not path.exists():
|
||||
continue
|
||||
if not path.is_dir():
|
||||
raise NotADirectoryError
|
||||
self._read_directory(path, values)
|
||||
except OSError:
|
||||
logger.warning("secrets 目录不可用，请检查目录类型与读取权限")
|
||||
return values
|
||||
|
||||
def _read_directory(self, directory: Path, values: dict[str, Any]) -> None:
|
||||
"""
|
||||
一次索引目录后按精确字段名定位文件，支持 K8s 投影链接。
|
||||
|
||||
:param directory: 已确认存在的 secrets 目录
|
||||
:param values: 按目录顺序合并的配置值，后面的同名文件优先
|
||||
"""
|
||||
protected = _secret_fields_context.get()
|
||||
entries = {entry.name: entry for entry in directory.iterdir()}
|
||||
for name, field in self.settings_cls.model_fields.items():
|
||||
path = entries.get(name)
|
||||
if path is None:
|
||||
continue
|
||||
try:
|
||||
if protected is not None:
|
||||
protected.add(name)
|
||||
# 即使该文件无效，外部字段也不得被默认值或自动修正写回遮蔽。
|
||||
values.pop(name, None)
|
||||
if name in self.current_state:
|
||||
continue
|
||||
raw = path.read_text(encoding="utf-8-sig").strip()
|
||||
if not raw or (name == "API_TOKEN" and len(raw) < 16):
|
||||
raise ValueError
|
||||
value = self.prepare_field_value(name, field, raw, False)
|
||||
values[name] = TypeAdapter(field.rebuild_annotation()).validate_python(value)
|
||||
except (OSError, ValueError):
|
||||
if protected is not None:
|
||||
protected.add(name)
|
||||
logger.warning(f"配置项 '{name}' 的 secret 无法读取或格式无效，已跳过，请检查文件与字段格式")
|
||||
|
||||
|
||||
class Settings(BaseSettings, ConfigModel, LogConfigModel):
|
||||
"""
|
||||
加载部署环境配置并提供运行目录、外部地址和配置更新入口。
|
||||
|
||||
字段语义继承配置模型；构造实例会创建配置目录，二进制部署还会补齐内置环境文件。
|
||||
|
||||
:cvar model_config: 环境加载策略，配置项区分大小写，并从部署环境文件按 UTF-8 读取
|
||||
:cvar model_config: 来源优先级为构造参数、环境变量、环境文件、secrets、默认值
|
||||
:ivar _secret_fields: 存在外部 secret 的字段名，包含被高优先级来源遮蔽或无效的文件
|
||||
"""
|
||||
|
||||
model_config = SettingsConfigDict(
|
||||
case_sensitive=True,
|
||||
env_file=get_env_path(),
|
||||
env_file_encoding="utf-8",
|
||||
secrets_dir=_get_secrets_dir(),
|
||||
)
|
||||
_secret_fields: set[str] = PrivateAttr(default_factory=set)
|
||||
|
||||
@classmethod
|
||||
def settings_customise_sources(
|
||||
cls,
|
||||
settings_cls: type[BaseSettings],
|
||||
init_settings: PydanticBaseSettingsSource,
|
||||
env_settings: PydanticBaseSettingsSource,
|
||||
dotenv_settings: PydanticBaseSettingsSource,
|
||||
file_secret_settings: PydanticBaseSettingsSource,
|
||||
) -> tuple[PydanticBaseSettingsSource, ...]:
|
||||
"""
|
||||
保持既有来源优先级，仅替换 secrets 读取以隔离文件错误。
|
||||
|
||||
:param settings_cls: 本次构造的设置类型
|
||||
:param init_settings: 显式构造参数来源
|
||||
:param env_settings: 进程环境来源
|
||||
:param dotenv_settings: 部署环境文件来源
|
||||
:param file_secret_settings: 框架按 model_config 或 _secrets_dir 创建的来源
|
||||
:return: 从高到低排列的来源
|
||||
"""
|
||||
secret_source = _DeploymentSecretsSource(
|
||||
settings_cls, secrets_dir=file_secret_settings.secrets_dir
|
||||
)
|
||||
return init_settings, env_settings, dotenv_settings, secret_source
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
"""
|
||||
加载环境配置并确保运行时目录和内置配置可用。
|
||||
"""
|
||||
super().__init__(**kwargs)
|
||||
protected: set[str] = set()
|
||||
token = _secret_fields_context.set(protected)
|
||||
try:
|
||||
super().__init__(**kwargs)
|
||||
self._secret_fields = protected
|
||||
finally:
|
||||
_secret_fields_context.reset(token)
|
||||
for path in [self.CONFIG_PATH, self.TEMP_PATH, self.LOG_PATH, self.COOKIE_PATH]:
|
||||
if not path.exists():
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
@@ -957,6 +1079,10 @@ class Settings(BaseSettings, ConfigModel, LogConfigModel):
|
||||
"""
|
||||
按字段策略序列化并更新 env 配置。
|
||||
"""
|
||||
if field_name in (_secret_fields_context.get() or set()):
|
||||
message = f"配置项 '{field_name}' 由 secret 文件管理，请修改文件并重启"
|
||||
logger.warning(message)
|
||||
return False, message
|
||||
policy = _get_setting_policy(Settings.model_fields.get(field_name))
|
||||
# 成功且无提示时使用空字符串，保证与 Tuple[bool, str] 返回类型一致
|
||||
message = ""
|
||||
@@ -1018,6 +1144,10 @@ class Settings(BaseSettings, ConfigModel, LogConfigModel):
|
||||
if not field:
|
||||
return False, f"配置项 '{key}' 不存在"
|
||||
original_value = getattr(self, key)
|
||||
if key in self._secret_fields:
|
||||
if value == original_value:
|
||||
return None, ""
|
||||
return False, f"配置项 '{key}' 由 secret 文件管理，请修改文件并重启"
|
||||
converted_value, needs_update = self.generic_type_converter(
|
||||
value, original_value, field.annotation, field.default, key
|
||||
)
|
||||
|
||||
@@ -420,7 +420,7 @@ JSON values are copied so callers cannot mutate shared cache state.
|
||||
|
||||
**Location:** `ConfigModel` and `Settings` in `app/runtime/config.py`
|
||||
|
||||
These values are read from environment variables (or `.moviepilot.env`) at startup and are immutable at runtime. They are not stored in the database.
|
||||
These values are loaded at startup from process environment variables, `<CONFIG_DIR>/app.env`, secret files, and field defaults, in that order. Explicit constructor arguments take precedence. They are not stored in the database. Web/CLI updates write `app.env` only for locally managed fields; environment and secret-managed fields reject persistence. See [Secrets deployment](../secrets.md) for directory selection, formats, rotation, and migration.
|
||||
|
||||
**Access:**
|
||||
|
||||
|
||||
60
docs/secrets.md
Normal file
60
docs/secrets.md
Normal file
@@ -0,0 +1,60 @@
|
||||
# 使用 Docker / Kubernetes Secrets
|
||||
|
||||
MoviePilot V3 支持把部署配置作为 UTF-8 文件挂载到 `/run/secrets/`，例如
|
||||
`/run/secrets/API_TOKEN`、`/run/secrets/GITHUB_TOKEN` 或
|
||||
`/run/secrets/DB_POSTGRESQL_PASSWORD`。文件内容直接进入配置解析，不导出为进程环境变量，
|
||||
不复制到 `app.env`。此能力针对主程序的部署配置，不读取插件数据库配置。
|
||||
|
||||
配置优先级为：**显式构造参数 > 进程环境变量 > `<CONFIG_DIR>/app.env` > secret 文件 > 字段默认值**。
|
||||
Web 设置通常写入 `app.env`；环境变量管理的项仍拒绝界面写回。
|
||||
只要目录中存在同名 secret 项（包括无效文件、目录或被高优先级来源遮蔽的项），
|
||||
界面与自动修正都不会为它写入 `app.env`，并提示修改文件后重启。原值不变的提交无需更新。
|
||||
|
||||
## 目录与格式
|
||||
|
||||
- 默认读取 `/run/secrets`，目录不存在时静默使用其他来源与默认值。
|
||||
- 在**进程环境变量**中设置 `MOVIEPILOT_SECRETS_DIR` 可更换目录，兼容 `SECRETS_DIR`；前者优先。
|
||||
显式设为空字符串、`false`、`off` 或 `0` 可关闭 secrets。目录开关在模块导入时确定，
|
||||
不从 `app.env` 或 secret 文件读取。`CONFIG_DIR` 仍应在进程环境中设置。
|
||||
- 文件名必须与配置项大小写完全一致，例如 `API_TOKEN`，不能写成 `api_token`。
|
||||
- 文件写**裸值**，不要照搬 `app.env` 的外层引号；引号属于文件内容。
|
||||
- 内容使用 UTF-8，允许 UTF-8 BOM；首尾空白和末尾换行会去除，内部换行保留。
|
||||
- list / dict 字段使用 JSON，例如 `ALLOWED_HOSTS` 的内容为 `["https://mp.example.com"]`，
|
||||
`DB_CONNECT_ARGS` 的内容为 `{"connect_timeout": 10}`。
|
||||
- 空文件、非法 UTF-8、非法 JSON、类型或字段约束错误、过短的 `API_TOKEN` 会按项跳过，
|
||||
日志只提示字段名。`API_TOKEN` 需至少 16 个字符；无其他有效来源时保持未配置，
|
||||
不为无效的外部令牌生成另一份密钥。错误目录或读取权限也会告警，请修正挂载和运行用户权限。
|
||||
- 支持 K8s 投影文件的符号链接。轮换文件内容后必须重启 MoviePilot；现有进程不会热加载。
|
||||
- 当前不支持 `XXX_FILE` 约定。单项任意路径可通过挂载到 secrets 目录中的同名文件实现。
|
||||
|
||||
## Compose 示例
|
||||
|
||||
```yaml
|
||||
services:
|
||||
moviepilot:
|
||||
image: jxxghp/moviepilot-v3:latest
|
||||
volumes:
|
||||
- ./config:/config
|
||||
secrets:
|
||||
- source: moviepilot_api_token
|
||||
target: API_TOKEN
|
||||
- source: moviepilot_github_token
|
||||
target: GITHUB_TOKEN
|
||||
|
||||
secrets:
|
||||
moviepilot_api_token:
|
||||
file: ./private/api_token
|
||||
moviepilot_github_token:
|
||||
file: ./private/github_token
|
||||
```
|
||||
|
||||
把实际密钥放到 `./private/` 对应文件并从版本管理中排除；Compose 文件只引用路径。
|
||||
Docker Swarm 可将顶层声明改为已有的外部 secret；Kubernetes 可把 Secret 的 key 按配置名投影到
|
||||
`/run/secrets`。按部署平台提供合适的只读权限，确保 MoviePilot 运行用户能够读取。
|
||||
|
||||
## 从旧配置迁移
|
||||
|
||||
先准备 secret 文件，再从编排环境变量和 `app.env` 中移除同名值，最后重启。
|
||||
高优先级值仍存在时会继续生效，轮换 secret 无法覆盖它。已经写在 `app.env` 里的历史值
|
||||
需要手工删除整行；界面“清空”可能写入字段默认值，不代表删除覆盖项。
|
||||
外部管理的配置不能通过 Web 或 CLI 改为本地配置；需要先移除 secret 挂载并重启。
|
||||
304
tests/test_config_secrets.py
Normal file
304
tests/test_config_secrets.py
Normal file
@@ -0,0 +1,304 @@
|
||||
"""
|
||||
验证部署 secret 的读取、来源顺序、坏文件隔离与配置写回保护。
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from app.runtime.config import Settings, _get_secrets_dir
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def secret_config(tmp_path, monkeypatch):
|
||||
"""
|
||||
为每个用例隔离环境、环境文件和 secret 目录。
|
||||
|
||||
:yields: 配置构造函数、secret 目录、环境文件路径和收集的安全告警
|
||||
"""
|
||||
for name in Settings.model_fields:
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
directory = tmp_path / "secrets"
|
||||
directory.mkdir()
|
||||
env_file = tmp_path / "app.env"
|
||||
env_file.write_text("# isolated\n", encoding="utf-8")
|
||||
messages = []
|
||||
monkeypatch.setattr("app.runtime.config.get_env_path", lambda: env_file)
|
||||
monkeypatch.setattr("app.runtime.config.logger.warning", messages.append)
|
||||
|
||||
def create(**kwargs: Any) -> Settings:
|
||||
"""
|
||||
在临时目录内构造真实配置实例。
|
||||
"""
|
||||
options = dict(_secrets_dir=directory, _env_file=env_file, CONFIG_DIR=str(tmp_path))
|
||||
options.update(kwargs)
|
||||
return Settings(**options)
|
||||
|
||||
yield create, directory, env_file, messages
|
||||
|
||||
|
||||
def test_secrets_priority_and_no_environment_export(secret_config, monkeypatch):
|
||||
"""
|
||||
构造参数、进程环境、dotenv、secret 与默认值按原有顺序生效。
|
||||
"""
|
||||
create, directory, env_file, _messages = secret_config
|
||||
(directory / "LOG_MAX_FILE_SIZE").write_text("7\n", encoding="utf-8")
|
||||
assert create().LOG_MAX_FILE_SIZE == 7
|
||||
env_file.write_text("LOG_MAX_FILE_SIZE='8'\n", encoding="utf-8")
|
||||
assert create().LOG_MAX_FILE_SIZE == 8
|
||||
monkeypatch.setenv("LOG_MAX_FILE_SIZE", "9")
|
||||
assert create().LOG_MAX_FILE_SIZE == 9
|
||||
assert create(LOG_MAX_FILE_SIZE=10).LOG_MAX_FILE_SIZE == 10
|
||||
monkeypatch.delenv("LOG_MAX_FILE_SIZE")
|
||||
env_file.write_text("# isolated\n", encoding="utf-8")
|
||||
(directory / "LOG_MAX_FILE_SIZE").unlink()
|
||||
assert create().LOG_MAX_FILE_SIZE == 5
|
||||
token = "file-token-0123456789"
|
||||
(directory / "API_TOKEN").write_text(token + "\n", encoding="utf-8")
|
||||
before = env_file.read_bytes()
|
||||
assert create().API_TOKEN == token
|
||||
assert "API_TOKEN" not in os.environ
|
||||
assert env_file.read_bytes() == before
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name, raw, expected", [
|
||||
("DEBUG", "true", True),
|
||||
("LOG_MAX_FILE_SIZE", "9", 9),
|
||||
("ALLOWED_HOSTS", '["https://example.test"]', ["https://example.test"]),
|
||||
("DB_CONNECT_ARGS", '{"connect_timeout":10}', {"connect_timeout": 10}),
|
||||
("GITHUB_TOKEN", "\ufefffile-token-0123456789\r\n", "file-token-0123456789"),
|
||||
("GITHUB_TOKEN", "first\nsecond\n", "first\nsecond"),
|
||||
("GITHUB_TOKEN", "'quoted-value'", "'quoted-value'"),
|
||||
])
|
||||
def test_secrets_formats(secret_config, name, raw, expected):
|
||||
"""
|
||||
标量与 JSON 复合值按字段解析，BOM 和末尾换行不进入生效值。
|
||||
"""
|
||||
create, directory, env_file, _messages = secret_config
|
||||
(directory / name).write_text(raw, encoding="utf-8")
|
||||
before = env_file.read_bytes()
|
||||
assert getattr(create(), name) == expected
|
||||
assert env_file.read_bytes() == before
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name, raw", [
|
||||
("API_TOKEN", b"short-token"),
|
||||
("API_TOKEN", b"\xff\xfe"),
|
||||
("DEBUG", b"notabool-sensitive"),
|
||||
("LOG_MAX_FILE_SIZE", b"notanumber-sensitive"),
|
||||
("LOG_LEVEL", b""),
|
||||
("ALLOWED_HOSTS", b"not-json-sensitive"),
|
||||
("ALLOWED_HOSTS", b"{}"),
|
||||
("DB_CONNECT_ARGS", b"[]"),
|
||||
("DB_CONNECT_ARGS", b"not-json-sensitive"),
|
||||
("API_WORKERS", b"0"),
|
||||
])
|
||||
def test_invalid_secret_is_skipped_without_write_or_leak(secret_config, name, raw):
|
||||
"""
|
||||
不可读编码、空值、类型和约束错误只影响单项，不写环境文件或泄露原文。
|
||||
"""
|
||||
create, directory, env_file, messages = secret_config
|
||||
baseline = create()
|
||||
(directory / name).write_bytes(raw)
|
||||
(directory / "GITHUB_TOKEN").write_text("valid-other-file", encoding="utf-8")
|
||||
before = env_file.read_bytes()
|
||||
config = create()
|
||||
assert getattr(config, name) == getattr(baseline, name)
|
||||
assert config.GITHUB_TOKEN == "valid-other-file"
|
||||
assert config.update_setting(name, "replacement")[0] is False
|
||||
assert env_file.read_bytes() == before
|
||||
assert messages and name in messages[-1]
|
||||
assert "sensitive" not in " ".join(messages)
|
||||
assert "short-token" not in " ".join(messages)
|
||||
|
||||
|
||||
def test_secret_is_readonly_even_when_shadowed(secret_config):
|
||||
"""
|
||||
已存在 secret 的字段不能被界面改写，历史 dotenv 遮蔽需手工移除。
|
||||
"""
|
||||
create, directory, env_file, _messages = secret_config
|
||||
(directory / "DEBUG").write_text("true", encoding="utf-8")
|
||||
config = create()
|
||||
assert config.update_setting("DEBUG", True) == (None, "")
|
||||
success, message = config.update_setting("DEBUG", False)
|
||||
assert success is False and "secret" in message
|
||||
assert config.DEBUG is True
|
||||
env_file.write_text("DEBUG='False'\n", encoding="utf-8")
|
||||
config = create()
|
||||
assert config.DEBUG is False
|
||||
assert config.update_setting("DEBUG", True)[0] is False
|
||||
env_file.write_text("# removed override\n", encoding="utf-8")
|
||||
assert create().DEBUG is True
|
||||
|
||||
|
||||
def test_shadowed_secret_prevents_automatic_correction_write(secret_config):
|
||||
"""
|
||||
即使高优先级文件需要自动修正，也不能覆盖外部管理的 secret 项。
|
||||
"""
|
||||
create, directory, env_file, _messages = secret_config
|
||||
(directory / "DEBUG").write_text("true", encoding="utf-8")
|
||||
env_file.write_text("DEBUG='bad'\n", encoding="utf-8")
|
||||
before = env_file.read_bytes()
|
||||
assert create().DEBUG is False
|
||||
assert env_file.read_bytes() == before
|
||||
|
||||
|
||||
def test_secrets_rotation_and_instance_isolation(secret_config):
|
||||
"""
|
||||
文件轮换仅新实例读取；禁用 secrets 的实例不会继承其他实例的写回锁。
|
||||
"""
|
||||
create, directory, _env_file, _messages = secret_config
|
||||
path = directory / "GITHUB_TOKEN"
|
||||
path.write_text("before", encoding="utf-8")
|
||||
first = create()
|
||||
path.write_text("after", encoding="utf-8")
|
||||
assert first.GITHUB_TOKEN == "before"
|
||||
assert create().GITHUB_TOKEN == "after"
|
||||
plain = create(_secrets_dir=())
|
||||
assert plain.update_setting("GITHUB_TOKEN", "local")[0] is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("kind", ["missing", "file", "unreadable", "entry_directory"])
|
||||
def test_unavailable_secret_paths(secret_config, monkeypatch, kind):
|
||||
"""
|
||||
缺失目录不告警，错误目录或无权限文件按项告警并保留默认值。
|
||||
"""
|
||||
create, directory, _env_file, messages = secret_config
|
||||
if kind == "missing":
|
||||
directory.rmdir()
|
||||
elif kind == "file":
|
||||
directory.rmdir()
|
||||
directory.write_text("not-a-directory", encoding="utf-8")
|
||||
elif kind == "entry_directory":
|
||||
(directory / "API_TOKEN").mkdir()
|
||||
else:
|
||||
(directory / "API_TOKEN").write_text("unreadable-token", encoding="utf-8")
|
||||
original = Path.read_text
|
||||
|
||||
def read(path, *args, **kwargs):
|
||||
"""
|
||||
仅对目标 secret 模拟运行用户无读取权限。
|
||||
"""
|
||||
if path == directory / "API_TOKEN":
|
||||
raise PermissionError
|
||||
return original(path, *args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(Path, "read_text", read)
|
||||
assert create().API_TOKEN is None
|
||||
assert bool(messages) is (kind != "missing")
|
||||
|
||||
|
||||
def test_kubernetes_symlink_and_exact_case(secret_config):
|
||||
"""
|
||||
K8s 的投影符号链接可读，小写文件名不能冒充大写配置名。
|
||||
"""
|
||||
create, directory, env_file, _messages = secret_config
|
||||
(directory / "api_token").write_text("lowercase-token-012345", encoding="utf-8")
|
||||
assert create().API_TOKEN is None
|
||||
(directory / "api_token").unlink()
|
||||
data = directory / "..data"
|
||||
data.mkdir()
|
||||
(data / "API_TOKEN").write_text("projected-token-012345", encoding="utf-8")
|
||||
(directory / "API_TOKEN").symlink_to("..data/API_TOKEN")
|
||||
before = env_file.read_bytes()
|
||||
assert create().API_TOKEN == "projected-token-012345"
|
||||
assert env_file.read_bytes() == before
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["", "false", "off", "0", "FALSE"])
|
||||
def test_secrets_directory_can_be_disabled(monkeypatch, value):
|
||||
"""
|
||||
环境变量可显式关闭 secrets，专用变量优先于通用变量。
|
||||
"""
|
||||
monkeypatch.setenv("SECRETS_DIR", "/other/secrets")
|
||||
monkeypatch.setenv("MOVIEPILOT_SECRETS_DIR", value)
|
||||
assert _get_secrets_dir() is None
|
||||
|
||||
|
||||
def test_secrets_directory_default_and_override(monkeypatch):
|
||||
"""
|
||||
默认路径与两个进程环境变量均有明确优先级。
|
||||
"""
|
||||
monkeypatch.delenv("MOVIEPILOT_SECRETS_DIR", raising=False)
|
||||
monkeypatch.delenv("SECRETS_DIR", raising=False)
|
||||
assert _get_secrets_dir() == "/run/secrets"
|
||||
monkeypatch.setenv("SECRETS_DIR", "/custom/secrets")
|
||||
assert _get_secrets_dir() == "/custom/secrets"
|
||||
monkeypatch.setenv("MOVIEPILOT_SECRETS_DIR", "/preferred/secrets")
|
||||
assert _get_secrets_dir() == "/preferred/secrets"
|
||||
|
||||
|
||||
def test_secrets_directory_is_indexed_once(secret_config, monkeypatch):
|
||||
"""
|
||||
无关文件增多时仍只扫描一次目录，而非逐字段重复扫描。
|
||||
"""
|
||||
create, directory, _env_file, _messages = secret_config
|
||||
for index in range(100):
|
||||
(directory / f"unrelated-{index}").touch()
|
||||
(directory / "DEBUG").write_text("true", encoding="utf-8")
|
||||
original = Path.iterdir
|
||||
scanned = []
|
||||
|
||||
def scan(path):
|
||||
"""
|
||||
记录目标目录扫描次数。
|
||||
"""
|
||||
if path == directory:
|
||||
scanned.append(path)
|
||||
return original(path)
|
||||
|
||||
monkeypatch.setattr(Path, "iterdir", scan)
|
||||
assert create().DEBUG is True
|
||||
assert scanned == [directory]
|
||||
|
||||
|
||||
def test_unreadable_secret_directory_is_skipped(secret_config, monkeypatch):
|
||||
"""
|
||||
整个目录无权限时保留配置默认值且明确告警。
|
||||
"""
|
||||
create, directory, _env_file, messages = secret_config
|
||||
original = Path.iterdir
|
||||
|
||||
def scan(path):
|
||||
"""
|
||||
模拟运行用户无法列出 secret 目录。
|
||||
"""
|
||||
if path == directory:
|
||||
raise PermissionError
|
||||
return original(path)
|
||||
|
||||
monkeypatch.setattr(Path, "iterdir", scan)
|
||||
assert create().API_TOKEN is None
|
||||
assert messages == ["secrets 目录不可用，请检查目录类型与读取权限"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("selector", ["MOVIEPILOT_SECRETS_DIR", "SECRETS_DIR", "disabled"])
|
||||
def test_module_singleton_loads_secret_without_constructor_override(secret_config, selector):
|
||||
"""
|
||||
真实导入单例会自动读取进程指定目录，关闭开关则不读取文件。
|
||||
"""
|
||||
_create, directory, env_file, _messages = secret_config
|
||||
(directory / "API_TOKEN").write_text("module-token-0123456789", encoding="utf-8")
|
||||
environment = dict(os.environ)
|
||||
environment.pop("MOVIEPILOT_SECRETS_DIR", None)
|
||||
environment.pop("SECRETS_DIR", None)
|
||||
environment["CONFIG_DIR"] = str(env_file.parent)
|
||||
environment["SECRETS_DIR" if selector == "disabled" else selector] = str(directory)
|
||||
if selector == "disabled":
|
||||
environment["MOVIEPILOT_SECRETS_DIR"] = "false"
|
||||
before = env_file.read_bytes()
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", (
|
||||
"import json, os; from app.runtime.config import settings; "
|
||||
"print(json.dumps([settings.API_TOKEN, 'API_TOKEN' in os.environ]))"
|
||||
)],
|
||||
env=environment, check=True, capture_output=True, text=True, timeout=30,
|
||||
)
|
||||
expected = None if selector == "disabled" else "module-token-0123456789"
|
||||
assert json.loads(result.stdout.strip().splitlines()[-1]) == [expected, False]
|
||||
assert env_file.read_bytes() == before
|
||||
Reference in New Issue
Block a user