diff --git a/app/adapters/web/security/access.py b/app/adapters/web/security/access.py index 933c20013..6eb69abb5 100644 --- a/app/adapters/web/security/access.py +++ b/app/adapters/web/security/access.py @@ -113,9 +113,10 @@ def _decode_token(token: str | None, purpose: str) -> TokenPayload: def _get_api_token( token_query: Annotated[str | None, Security(api_token_query)] = None, + key_header: Annotated[str | None, Security(api_key_header)] = None, ) -> str | None: - """从 URL 查询参数读取兼容 API Token。""" - return token_query + """优先从请求头、其次从查询参数读取兼容 API Token。""" + return key_header or token_query def _get_api_key( @@ -314,7 +315,7 @@ def _verify_key(key: str | None, expected_key: str, key_type: str) -> str: def verify_apitoken( token: Annotated[str | None, Security(_get_api_token)], ) -> str: - """校验 URL 查询参数中的兼容 API Token。""" + """校验请求头或 URL 查询参数中的兼容 API Token。""" value = _verify_key(token, get_runtime_setting("API_TOKEN"), "token") validate_api_credential_identity() return value diff --git a/app/api/endpoints/message.py b/app/api/endpoints/message.py index 9f648eb24..aeafe2e35 100644 --- a/app/api/endpoints/message.py +++ b/app/api/endpoints/message.py @@ -123,7 +123,7 @@ async def user_message( _: _SchemaTokenPayload = Depends(verify_apitoken), ): """ - 用户消息响应，配置请求中需要添加参数：token=API_TOKEN&source=消息配置名 + 用户消息响应；推荐通过 X-API-KEY 请求头传递 API_TOKEN，查询参数 token 仅保留兼容。 """ body = await request.body() form = await request.form() diff --git a/app/application/messaging/ingress.py b/app/application/messaging/ingress.py index 772e56ad6..ec4ab616f 100644 --- a/app/application/messaging/ingress.py +++ b/app/application/messaging/ingress.py @@ -18,6 +18,7 @@ class _MessageIngressRequest: url: str payload: Mapping[str, Any] + headers: Mapping[str, str] source: str timeout: float @@ -30,6 +31,7 @@ class MessageIngressPort(Protocol): url: str, payload: Mapping[str, Any], *, + headers: Mapping[str, str], timeout: float, ) -> Optional[int]: """同步投递 payload，并返回 HTTP 状态码或 None。""" @@ -40,6 +42,7 @@ class MessageIngressPort(Protocol): url: str, payload: Mapping[str, Any], *, + headers: Mapping[str, str], timeout: float, ) -> Optional[int]: """异步投递 payload，并返回 HTTP 状态码或 None。""" @@ -78,14 +81,17 @@ def _message_ingress_snapshot() -> MessageIngressPort: def build_message_ingress_url(source: str | None) -> str: - """按当前运行配置构造安全编码的本地消息入口 URL。""" - query = {"token": get_runtime_setting('API_TOKEN')} + """构造仅含非敏感来源参数的本地消息入口 URL。""" + query: dict[str, str] = {} if source: query["source"] = source - return ( - f"http://127.0.0.1:{get_runtime_setting('PORT')}/api/v1/message?" - f"{urlencode(query)}" - ) + base_url = f"http://127.0.0.1:{get_runtime_setting('PORT')}/api/v1/message" + return f"{base_url}?{urlencode(query)}" if query else base_url + + +def build_message_ingress_headers() -> dict[str, str]: + """把本地回环凭据放入请求头，避免访问日志记录明文 Token。""" + return {"X-API-KEY": str(get_runtime_setting("API_TOKEN") or "")} def _message_ingress_request( @@ -97,6 +103,7 @@ def _message_ingress_request( return _MessageIngressRequest( url=build_message_ingress_url(source), payload=dict(payload), + headers=build_message_ingress_headers(), source=source or "-", timeout=timeout, ) @@ -142,6 +149,7 @@ def forward_message_to_host( status_code = _message_ingress_snapshot().post( request.url, request.payload, + headers=request.headers, timeout=request.timeout, ) return _message_ingress_confirmed(request, status_code) @@ -161,6 +169,7 @@ async def async_forward_message_to_host( status_code = await _message_ingress_snapshot().async_post( request.url, request.payload, + headers=request.headers, timeout=request.timeout, ) return _message_ingress_confirmed(request, status_code) diff --git a/app/application/security/auth.py b/app/application/security/auth.py index e08586cef..50d66f1cc 100644 --- a/app/application/security/auth.py +++ b/app/application/security/auth.py @@ -154,6 +154,9 @@ class AuthUserRepository(Protocol): def get_by_id(self, user_id: int) -> Optional[AuthUser]: """按 ID 查询用户。""" + def get_active_superuser(self) -> Optional[AuthUser]: + """返回按稳定顺序选出的启用超级管理员。""" + class AuthPasskeyRepository(Protocol): """认证提供方查询端口。""" @@ -195,7 +198,9 @@ class AuthService: def build_superuser_token_payload(self) -> _SchemaTokenPayload: """从持久化用户和站点认证状态构造超级用户令牌载荷。""" - configured_name = get_chain_runtime_config_snapshot().superuser + configured_name = str( + get_chain_runtime_config_snapshot().superuser or "" + ).strip() if ( self._superuser_binding_id is not None and configured_name == self._superuser_binding_name @@ -203,12 +208,22 @@ class AuthService: # 配置保存用户名；持久化 ID 保证管理员改名不会让管理员级集成失效。 user = self._users.get_by_id(self._superuser_binding_id) else: - user = self._users.get_by_name(configured_name) + user = ( + self._users.get_by_name(configured_name) + if configured_name + else self._users.get_active_superuser() + ) if user: self._superuser_binding_name = configured_name self._superuser_binding_id = user.id if not user or not user.is_active or not user.is_superuser: - raise PermissionError("用户权限不足") + if not configured_name: + raise PermissionError( + "未配置 SUPERUSER，且数据库中没有可用超级管理员" + ) + raise PermissionError( + "SUPERUSER 对应用户不存在、未启用或非超级管理员" + ) return _SchemaTokenPayload( sub=user.id, username=user.name, diff --git a/app/cli.py b/app/cli.py index 1241ddb13..0583d73ea 100644 --- a/app/cli.py +++ b/app/cli.py @@ -17,9 +17,12 @@ import click import psutil from app.application.backup import BackupArtifact -from app.application.configuration import get_runtime_settings from app.runtime.config import Settings -from app.runtime.settings import get_runtime_setting +from app.runtime.settings import ( + get_runtime_setting, + has_runtime_setting, + update_runtime_setting, +) from app.runtime.state import SystemHelper from app.runtime.version import get_app_version, get_frontend_version from app.startup.composition.database import build_database_governance @@ -801,7 +804,9 @@ def _ensure_local_api_token() -> bool: if get_runtime_setting("API_TOKEN") and len(str(get_runtime_setting("API_TOKEN")).strip()) >= 16: return False - result, message = get_runtime_settings().update("API_TOKEN", get_runtime_setting("API_TOKEN") or "") + result, message = update_runtime_setting( + "API_TOKEN", get_runtime_setting("API_TOKEN") or "" + ) if result is False: raise click.ClickException(message or "初始化 API_TOKEN 失败") return result is True @@ -1324,7 +1329,7 @@ def config_path() -> None: @click.option("--show-secrets", is_flag=True, help="显示敏感配置原文") def config_list(show_secrets: bool) -> None: """列出当前配置""" - values = get_runtime_settings().snapshot() + values = {key: get_runtime_setting(key) for key in Settings.model_fields} for key in sorted(values): click.echo(f"{key}={_format_value(_mask_value(key, values[key], show_secrets))}") @@ -1334,9 +1339,9 @@ def config_list(show_secrets: bool) -> None: def config_get(key: str) -> None: """读取单个配置项""" setting_fields = Settings.model_fields.keys() - if key not in setting_fields and not get_runtime_settings().contains(key): + if key not in setting_fields and not has_runtime_setting(key): raise click.ClickException(f"配置项不存在：{key}") - click.echo(_format_value(get_runtime_settings().get(key))) + click.echo(_format_value(get_runtime_setting(key))) @config.command("set", context_settings=CONTEXT_SETTINGS) @@ -1344,7 +1349,7 @@ def config_get(key: str) -> None: @click.argument("value") def config_set(key: str, value: str) -> None: """写入单个配置项""" - result, message = get_runtime_settings().update(key, value) + result, message = update_runtime_setting(key, value) if result is False: raise click.ClickException(message or f"配置项更新失败：{key}") if result is None: @@ -1376,7 +1381,7 @@ def config_keys(pattern: Optional[str], show_current: bool, show_secrets: bool) if pattern and pattern.lower() not in key.lower(): continue default_value = _field_default(field) - current_value = get_runtime_settings().get(key, default_value) + current_value = get_runtime_setting(key, default_value) rows.append( ( key, @@ -1408,7 +1413,7 @@ def config_describe(key: str, show_secrets: bool) -> None: raise click.ClickException(f"配置项不存在：{key}") default_value = _field_default(field) - current_value = get_runtime_settings().get(key, default_value) + current_value = get_runtime_setting(key, default_value) click.echo(f"Key: {key}") click.echo(f"Type: {_annotation_name(field.annotation)}") click.echo(f"Default: {_format_value(_mask_value(key, default_value, show_secrets))}") diff --git a/app/db/adapters/user.py b/app/db/adapters/user.py index 17289d659..39bccad82 100644 --- a/app/db/adapters/user.py +++ b/app/db/adapters/user.py @@ -69,6 +69,17 @@ class SqlAlchemyUserRepository(UserRepository): model = self._oper.get_by_id(user_id) return _to_snapshot(model) if model else None + def get_active_superuser(self) -> Optional[UserSnapshot]: + """按主键顺序返回首个启用的超级管理员快照。""" + session = cast(Session, self._session) + model = session.execute( + select(User) + .where(User.is_active.is_(True), User.is_superuser.is_(True)) + .order_by(User.id) + .limit(1) + ).scalars().first() + return _to_snapshot(model) if model else None + async def async_has_users(self) -> bool: """使用最小列查询判断数据库中是否已有用户。""" session = self._require_async_session() @@ -225,6 +236,11 @@ class TransactionalUserRepository(ChainUserRepository): with self._sync_session() as session: return SqlAlchemyUserRepository(session).get_by_id(user_id) + def get_active_superuser(self) -> Optional[UserSnapshot]: + """在独立会话中返回首个启用的超级管理员快照。""" + with self._sync_session() as session: + return SqlAlchemyUserRepository(session).get_active_superuser() + def get_auth_by_name(self, name: str) -> Optional[UserAuthSnapshot]: """按用户名读取认证凭据快照。""" with self._sync_session() as session: diff --git a/app/startup/composition/network.py b/app/startup/composition/network.py index f246cfa55..d149637fb 100644 --- a/app/startup/composition/network.py +++ b/app/startup/composition/network.py @@ -128,10 +128,14 @@ class _MessageIngressAdapter: url: str, payload: Mapping[str, Any], *, + headers: Mapping[str, str], timeout: float, ) -> Optional[int]: """同步投递消息，关闭响应后返回状态码。""" - response = RequestUtils(timeout=timeout).post_res( # type: ignore[arg-type] + response = RequestUtils( + timeout=timeout, # type: ignore[arg-type] + headers=dict(headers), + ).post_res( url, json=dict(payload), ) @@ -150,11 +154,13 @@ class _MessageIngressAdapter: url: str, payload: Mapping[str, Any], *, + headers: Mapping[str, str], timeout: float, ) -> Optional[int]: """异步投递消息，关闭响应后返回状态码。""" response = await AsyncRequestUtils( - timeout=timeout # type: ignore[arg-type] + timeout=timeout, # type: ignore[arg-type] + headers=dict(headers), ).post_res( url, json=dict(payload), diff --git a/app/startup/composition/security.py b/app/startup/composition/security.py index 414011396..b46055d3d 100644 --- a/app/startup/composition/security.py +++ b/app/startup/composition/security.py @@ -12,6 +12,7 @@ from app.adapters.web.security.access import ( from app.application.configuration import get_configured_system_config from app.application.security.auth import ( AuthService, + AuthUserRepository, build_superuser_token_payload, configure_auth_service, reset_auth_service, @@ -30,6 +31,8 @@ from app.db.adapters.user import SqlAlchemyUserRepository from app.db.oper.passkey import PassKeyOper from app.db.oper.systemconfig import SystemConfigOper from app.runtime.cache import TTLCache +from app.runtime.log import logger +from app.runtime.settings import get_runtime_setting, update_runtime_setting from app.startup.composition.context import ( RepositoryFactory, StandaloneRepositoryFactory, @@ -48,8 +51,27 @@ class SecurityComposition: passkey: StandaloneRepositoryFactory +def _backfill_superuser_setting(users: AuthUserRepository) -> None: + """用现有数据库管理员补全 V2 升级后缺失的 SUPERUSER。""" + if str(get_runtime_setting("SUPERUSER") or "").strip(): + return + user = users.get_active_superuser() + if user is None: + return + success, message = update_runtime_setting("SUPERUSER", user.name) + if success is False: + logger.warning( + f"检测到数据库超级管理员 {user.name}，但自动补全 SUPERUSER 失败：" + f"{message or '未知错误'}" + ) + return + logger.info(f"已根据数据库超级管理员自动补全 SUPERUSER：{user.name}") + + def configure_security_services() -> SecurityComposition: """构造并登记认证、用户查询和 PassKey 服务。""" + users = build_transactional_user_repository() + _backfill_superuser_setting(users) configure_user_lookups( by_id=lambda user_id: build_transactional_user_repository().get_by_id(user_id), by_name=lambda username: build_transactional_user_repository().get_by_name(username), @@ -57,7 +79,7 @@ def configure_security_services() -> SecurityComposition: ) configure_auth_service( AuthService( - users=build_transactional_user_repository(), + users=users, config=get_configured_system_config(), passkeys=PassKeyOper(), ) diff --git a/tests/test_auth_contracts.py b/tests/test_auth_contracts.py index dc34a3f7e..986f57c52 100644 --- a/tests/test_auth_contracts.py +++ b/tests/test_auth_contracts.py @@ -360,6 +360,12 @@ def test_standalone_api_credential_revalidates_current_identity(monkeypatch, dep assert exc_info.value.headers == {"WWW-Authenticate": "Bearer"} +def test_api_token_reader_prefers_header_and_preserves_query_compatibility(): + """兼容 Token 依赖应优先使用请求头，同时继续接受旧查询参数。""" + assert access._get_api_token("query-token", "header-token") == "header-token" + assert access._get_api_token("query-token", None) == "query-token" + + @pytest.mark.asyncio @pytest.mark.parametrize( ("endpoint", "credential"), @@ -475,6 +481,52 @@ def test_superuser_payload_provider_rejects_inactive_user(monkeypatch): service.build_superuser_token_payload() +def test_superuser_payload_provider_falls_back_to_database_admin(monkeypatch): + """V2 升级后 SUPERUSER 为空时，API 凭据应绑定现有启用管理员。""" + user = _user(active=True, superuser=True) + users = SimpleNamespace( + get_by_name=Mock(), + get_active_superuser=Mock(return_value=user), + ) + service = AuthService( + users=users, + config=SimpleNamespace(), + passkeys=SimpleNamespace(), + ) + monkeypatch.setattr( + auth_service_module, + "get_chain_runtime_config_snapshot", + lambda: SimpleNamespace(superuser=""), + ) + + payload = service.build_superuser_token_payload() + + assert payload.sub == user.id + assert payload.username == user.name + users.get_by_name.assert_not_called() + users.get_active_superuser.assert_called_once_with() + + +def test_superuser_payload_provider_explains_missing_binding(monkeypatch): + """配置和数据库都没有管理员时，应返回可操作的认证失败原因。""" + service = AuthService( + users=SimpleNamespace(get_active_superuser=Mock(return_value=None)), + config=SimpleNamespace(), + passkeys=SimpleNamespace(), + ) + monkeypatch.setattr( + auth_service_module, + "get_chain_runtime_config_snapshot", + lambda: SimpleNamespace(superuser=""), + ) + + with pytest.raises( + PermissionError, + match="未配置 SUPERUSER，且数据库中没有可用超级管理员", + ): + service.build_superuser_token_payload() + + def test_auth_service_reads_user_by_id_and_accepts_current_token_identity(): """认证服务按稳定用户 ID 查询，并接受与当前账号一致的令牌声明。""" user = _user() diff --git a/tests/test_cli_configuration.py b/tests/test_cli_configuration.py new file mode 100644 index 000000000..eeb7974d3 --- /dev/null +++ b/tests/test_cli_configuration.py @@ -0,0 +1,63 @@ +"""离线配置 CLI 的运行时端口契约测试。""" + +import pytest +from click.testing import CliRunner + +from app import cli as cli_module + + +def test_config_get_works_without_web_lifespan( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """配置读取必须使用启动前可用的 runtime 端口，不能依赖 Web 组合根。""" + monkeypatch.setattr(cli_module, "has_runtime_setting", lambda _key: True) + monkeypatch.setattr( + cli_module, + "get_runtime_setting", + lambda key, *_args: "admin" if key == "SUPERUSER" else None, + ) + + result = CliRunner().invoke( + cli_module.cli, + ["config", "get", "SUPERUSER"], + ) + + assert result.exit_code == 0 + assert result.output == "admin\n" + + +def test_config_set_works_without_web_lifespan( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """配置写入必须走离线可用的 runtime 更新端口。""" + updates: list[tuple[str, str]] = [] + + def update_setting(key: str, value: str) -> tuple[bool, str]: + """记录离线配置写入。""" + updates.append((key, value)) + return True, "" + + monkeypatch.setattr( + cli_module, + "update_runtime_setting", + update_setting, + ) + monkeypatch.setattr( + cli_module, + "_managed_backend_status", + lambda: ("stopped", None, None, None), + ) + monkeypatch.setattr( + cli_module, + "_managed_frontend_status", + lambda: ("stopped", None, None, None), + ) + + result = CliRunner().invoke( + cli_module.cli, + ["config", "set", "SUPERUSER", "admin"], + ) + + assert result.exit_code == 0 + assert updates == [("SUPERUSER", "admin")] + assert result.output == "SUPERUSER 已更新\n" diff --git a/tests/test_message_ingress.py b/tests/test_message_ingress.py index 48b4cece0..57fc74f39 100644 --- a/tests/test_message_ingress.py +++ b/tests/test_message_ingress.py @@ -28,14 +28,14 @@ class _FakeMessageIngressPort: self.sync_calls = [] self.async_calls = [] - def post(self, url, payload, *, timeout): + def post(self, url, payload, *, headers, timeout): """记录同步投递并返回固定状态码。""" - self.sync_calls.append((url, payload, timeout)) + self.sync_calls.append((url, payload, headers, timeout)) return self.status_code - async def async_post(self, url, payload, *, timeout): + async def async_post(self, url, payload, *, headers, timeout): """记录异步投递并返回固定状态码。""" - self.async_calls.append((url, payload, timeout)) + self.async_calls.append((url, payload, headers, timeout)) return self.status_code @@ -57,8 +57,8 @@ def _patch_ingress_settings(monkeypatch, **values): ) -def test_forward_message_to_host_encodes_source_and_copies_payload(monkeypatch): - """统一入口必须安全编码查询参数，并向端口传递 payload 副本。""" +def test_forward_message_to_host_keeps_token_out_of_url(monkeypatch): + """统一入口仅编码来源参数，并通过请求头传递凭据和 payload 副本。""" port = _FakeMessageIngressPort() ingress.configure_message_ingress_port(port) _patch_ingress_settings(monkeypatch, PORT=3000, API_TOKEN="token value") @@ -70,12 +70,12 @@ def test_forward_message_to_host_encodes_source_and_copies_payload(monkeypatch): timeout=9, ) is True - url, forwarded, timeout = port.sync_calls[0] + url, forwarded, headers, timeout = port.sync_calls[0] assert urlparse(url).path == "/api/v1/message" assert parse_qs(urlparse(url).query) == { - "token": ["token value"], "source": ["channel & one"], } + assert headers == {"X-API-KEY": "token value"} assert forwarded == {"text": "hello"} assert forwarded is not payload assert timeout == 9 @@ -114,11 +114,11 @@ async def test_async_forward_message_to_host_uses_same_contract(monkeypatch): timeout=10, ) is True - url, payload, timeout = port.async_calls[0] + url, payload, headers, timeout = port.async_calls[0] assert parse_qs(urlparse(url).query) == { - "token": ["token value"], "source": ["discord & one"], } + assert headers == {"X-API-KEY": "token value"} assert payload == {"text": "hello"} assert timeout == 10 @@ -139,10 +139,15 @@ def test_startup_message_ingress_adapter_closes_sync_response(monkeypatch): status_code = network_composition._MessageIngressAdapter().post( "http://127.0.0.1/message", {"text": "hello"}, + headers={"X-API-KEY": "secret"}, timeout=9, ) assert status_code == 202 + network_composition.RequestUtils.assert_called_once_with( + timeout=9, + headers={"X-API-KEY": "secret"}, + ) response.close.assert_called_once_with() @@ -163,10 +168,15 @@ async def test_startup_message_ingress_adapter_closes_async_response(monkeypatch status_code = await network_composition._MessageIngressAdapter().async_post( "http://127.0.0.1/message", {"text": "hello"}, + headers={"X-API-KEY": "secret"}, timeout=10, ) assert status_code == 202 + network_composition.AsyncRequestUtils.assert_called_once_with( + timeout=10, + headers={"X-API-KEY": "secret"}, + ) response.aclose.assert_awaited_once_with() diff --git a/tests/test_security_composition.py b/tests/test_security_composition.py new file mode 100644 index 000000000..3eb31d487 --- /dev/null +++ b/tests/test_security_composition.py @@ -0,0 +1,60 @@ +"""认证组合根的部署兼容修复测试。""" + +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from app.startup.composition import security as security_composition + + +def test_backfill_superuser_setting_from_existing_database_admin( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """V2 数据库已有管理员而配置缺失时，应持久化稳定管理员用户名。""" + updates: list[tuple[str, str]] = [] + users = SimpleNamespace( + get_active_superuser=Mock( + return_value=SimpleNamespace(name="legacy-admin") + ) + ) + monkeypatch.setattr( + security_composition, + "get_runtime_setting", + lambda _key: "", + ) + + def update_setting(key: str, value: str) -> tuple[bool, str]: + """记录启动兼容修复写入。""" + updates.append((key, value)) + return True, "" + + monkeypatch.setattr( + security_composition, + "update_runtime_setting", + update_setting, + ) + + security_composition._backfill_superuser_setting(users) + + assert updates == [("SUPERUSER", "legacy-admin")] + users.get_active_superuser.assert_called_once_with() + + +def test_backfill_superuser_setting_preserves_explicit_configuration( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """显式 SUPERUSER 必须保持原值，不得被数据库顺序覆盖。""" + users = SimpleNamespace(get_active_superuser=Mock()) + update = Mock() + monkeypatch.setattr( + security_composition, + "get_runtime_setting", + lambda _key: "configured-admin", + ) + monkeypatch.setattr(security_composition, "update_runtime_setting", update) + + security_composition._backfill_superuser_setting(users) + + users.get_active_superuser.assert_not_called() + update.assert_not_called() diff --git a/tests/test_user_repository.py b/tests/test_user_repository.py index fd32fbbd2..477e15e81 100644 --- a/tests/test_user_repository.py +++ b/tests/test_user_repository.py @@ -160,6 +160,21 @@ async def test_user_snapshots_are_detached_and_deeply_frozen(user_repository) -> public.permissions["features"]["search"] = False # type: ignore[index] +def test_user_repository_selects_first_active_superuser(user_repository) -> None: + """升级绑定只能选择启用管理员，并按主键保持确定顺序。""" + repository, sync_factory = user_repository + _insert_user(sync_factory, name="disabled-admin", is_active=False) + expected_id = _insert_user(sync_factory, name="first-admin") + _insert_user(sync_factory, name="second-admin") + _insert_user(sync_factory, name="member", is_superuser=False) + + selected = repository.get_active_superuser() + + assert selected is not None + assert selected.id == expected_id + assert selected.name == "first-admin" + + def test_auxiliary_create_commits_before_return(user_repository) -> None: """辅助认证创建成功返回时，新用户必须已对后续独立会话可见。""" repository, sync_factory = user_repository