diff --git a/backend/src/boot_overlay.py b/backend/src/boot_overlay.py index 9caceb15..435e5665 100644 --- a/backend/src/boot_overlay.py +++ b/backend/src/boot_overlay.py @@ -190,12 +190,16 @@ def _verify_bundle(bundle: Path, sig: Path, pubkey_path: Path) -> bool: import base64 try: + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from cryptography.hazmat.primitives.serialization import load_pem_public_key except ImportError: logger.error("cryptography unavailable; refusing overlay.") return False try: public_key = load_pem_public_key(pubkey_path.read_bytes()) + if not isinstance(public_key, Ed25519PublicKey): + logger.error("Overlay pubkey is not ed25519; refusing overlay.") + return False signature = base64.b64decode(sig.read_text().strip(), validate=True) public_key.verify(signature, bundle.read_bytes()) return True diff --git a/backend/src/module/update/signing.py b/backend/src/module/update/signing.py index 35c727d6..e33ffa2a 100644 --- a/backend/src/module/update/signing.py +++ b/backend/src/module/update/signing.py @@ -29,6 +29,7 @@ def verify_bundle_signature( """ try: from cryptography.exceptions import InvalidSignature + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from cryptography.hazmat.primitives.serialization import load_pem_public_key except ImportError: logger.error("[Update] cryptography unavailable; refusing unverified bundle") @@ -36,6 +37,9 @@ def verify_bundle_signature( try: public_key = load_pem_public_key(pubkey_path.read_bytes()) + if not isinstance(public_key, Ed25519PublicKey): + logger.error("[Update] pubkey is not ed25519; refusing bundle") + return False signature = base64.b64decode(signature_b64.strip(), validate=True) public_key.verify(signature, bundle_path.read_bytes()) return True diff --git a/backend/src/test/test_context.py b/backend/src/test/test_context.py index 3881a544..6bfd16b2 100644 --- a/backend/src/test/test_context.py +++ b/backend/src/test/test_context.py @@ -15,22 +15,25 @@ def ctx(): """A built AppContext whose notifier/scheduler are replaced with mocks.""" ctx = AppContext.build(real_settings) ctx.notifier = MagicMock() - ctx.scheduler = MagicMock() - ctx.scheduler.stop_all = AsyncMock() - ctx.scheduler.start_all = MagicMock() + # 通过局部变量操作 mock:嵌套函数里 ctx.scheduler 的声明类型是 Scheduler, + # 其 running 是只读 property,mypy 会拒绝直接赋值。 + scheduler = MagicMock() + scheduler.stop_all = AsyncMock() + scheduler.start_all = MagicMock() # A freshly built context has not started its scheduler yet; set this # explicitly since a bare MagicMock attribute is truthy by default and # would make reload_settings() think a restart is needed. - ctx.scheduler.running = False + scheduler.running = False def _mark_running(): - ctx.scheduler.running = True + scheduler.running = True async def _mark_stopped(): - ctx.scheduler.running = False + scheduler.running = False - ctx.scheduler.start_all.side_effect = _mark_running - ctx.scheduler.stop_all.side_effect = _mark_stopped + scheduler.start_all.side_effect = _mark_running + scheduler.stop_all.side_effect = _mark_stopped + ctx.scheduler = scheduler return ctx diff --git a/backend/src/test/test_migrations_module.py b/backend/src/test/test_migrations_module.py index 8bc97f80..1b66b9bf 100644 --- a/backend/src/test/test_migrations_module.py +++ b/backend/src/test/test_migrations_module.py @@ -2,6 +2,7 @@ import pytest from sqlalchemy import inspect, text +from sqlalchemy.engine import Engine from sqlalchemy.exc import OperationalError from sqlmodel import create_engine @@ -75,7 +76,7 @@ def _make_versioned_engine( *, bangumi_extra: str, torrent_extra: str = "", -) -> object: +) -> Engine: engine = create_engine("sqlite://") bangumi_extra_sql = f", {bangumi_extra}" if bangumi_extra else "" torrent_extra_sql = f", {torrent_extra}" if torrent_extra else ""