chore: merge 3.3-dev mypy fixes

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

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

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
 
 

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 ""