fix: handle missing anthropic usage on filtered responses

astrbot/core/platform/sources/qqofficial_webhook/qo_webhook_server.py

@@ -1,13 +1,10 @@
 import asyncio
-import json
 import logging
 import time
-from binascii import Error as BinasciiError
 from typing import cast
 
 import quart
 from botpy import BotAPI, BotHttp, BotWebSocket, Client, ConnectionSession, Token
-from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives.asymmetric import ed25519
 
 from astrbot.api import logger
@@ -16,57 +13,6 @@ from astrbot.api import logger
 for handler in logging.root.handlers[:]:
     logging.root.removeHandler(handler)
 
-_SIGNATURE_HEADER = "X-Signature-Ed25519"
-_SIGNATURE_TIMESTAMP_HEADER = "X-Signature-Timestamp"
-_ED25519_SEED_SIZE = 32
-_ED25519_SIGNATURE_SIZE = 64
-
-
-def _build_ed25519_seed(secret: str) -> bytes:
-    if not secret:
-        raise ValueError("QQ official bot secret is empty.")
-
-    seed = secret.encode("utf-8")
-    while len(seed) < _ED25519_SEED_SIZE:
-        seed *= 2
-    return seed[:_ED25519_SEED_SIZE]
-
-
-def _sign_qq_webhook_payload(secret: str, timestamp: str, payload: bytes) -> str:
-    seed = _build_ed25519_seed(secret)
-    private_key = ed25519.Ed25519PrivateKey.from_private_bytes(seed)
-    return private_key.sign(timestamp.encode("utf-8") + payload).hex()
-
-
-def _verify_qq_webhook_signature(
-    secret: str,
-    timestamp: str | None,
-    signature: str | None,
-    body: bytes,
-) -> bool:
-    if not timestamp or not signature:
-        return False
-
-    try:
-        signature_buffer = bytes.fromhex(signature)
-    except (BinasciiError, ValueError):
-        return False
-
-    if (
-        len(signature_buffer) != _ED25519_SIGNATURE_SIZE
-        or signature_buffer[63] & 224 != 0
-    ):
-        return False
-
-    try:
-        seed = _build_ed25519_seed(secret)
-        private_key = ed25519.Ed25519PrivateKey.from_private_bytes(seed)
-        public_key = private_key.public_key()
-        public_key.verify(signature_buffer, timestamp.encode("utf-8") + body)
-    except (InvalidSignature, ValueError):
-        return False
-    return True
-
 
 class QQOfficialWebhook:
     def __init__(
@@ -81,12 +27,7 @@ class QQOfficialWebhook:
         if isinstance(self.port, str):
             self.port = int(self.port)
 
-        self.http: BotHttp = BotHttp(
-            timeout=300,
-            is_sandbox=self.is_sandbox,
-            app_id=self.appid,
-            secret=self.secret,
-        )
+        self.http: BotHttp = BotHttp(timeout=300, is_sandbox=self.is_sandbox)
         self.api: BotAPI = BotAPI(http=self.http)
         self.token = Token(self.appid, self.secret)
 
@@ -99,7 +40,6 @@ class QQOfficialWebhook:
         self.client = botpy_client
         self.event_queue = event_queue
         self.shutdown_event = asyncio.Event()
-        self._connection: ConnectionSession | None = None
 
         # Cache for extra fields extracted from raw webhook payloads, keyed by message id
         self._extra_data_cache: dict[str, dict] = {}
@@ -115,13 +55,6 @@ class QQOfficialWebhook:
         # 直接注入到 botpy 的 Client，移花接木！
         self.client.api = self.api
         self.client.http = self.http
-        self._setup_connection()
-
-    def _setup_connection(self) -> None:
-        if self._connection is not None:
-            return
-        self.client.api = self.api
-        self.client.http = self.http
 
         async def bot_connect() -> None:
             pass
@@ -172,24 +105,7 @@ class QQOfficialWebhook:
         Returns:
             响应数据
         """
-        body = await request.get_data()
-        if not _verify_qq_webhook_signature(
-            self.secret,
-            request.headers.get(_SIGNATURE_TIMESTAMP_HEADER),
-            request.headers.get(_SIGNATURE_HEADER),
-            body,
-        ):
-            logger.warning("qq_official_webhook signature verification failed.")
-            return {"error": "Invalid signature"}, 401
-
-        try:
-            msg = json.loads(body.decode("utf-8"))
-        except json.JSONDecodeError:
-            logger.warning("qq_official_webhook callback body is not valid JSON.")
-            return {"error": "Invalid JSON"}, 400
-        if not isinstance(msg, dict):
-            return {"error": "Invalid JSON"}, 400
-
+        msg: dict = await request.json
         logger.debug(f"收到 qq_official_webhook 回调: {msg}")
 
         event = msg.get("t")
@@ -220,13 +136,6 @@ class QQOfficialWebhook:
 
         if event and opcode == BotWebSocket.WS_DISPATCH_EVENT:
             event = msg["t"].lower()
-            if self._connection is None:
-                logger.warning(
-                    "qq_official_webhook botpy connection is not initialized; "
-                    "creating parser connection lazily.",
-                )
-                self._setup_connection()
-
             # Extract extra fields from raw payload before botpy parses and discards them
             if data:
                 msg_id = data.get("id")

astrbot/core/star/base.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Any
+from typing import Any, Protocol
 
 from astrbot.core import html_renderer
 from astrbot.core.utils.command_parser import CommandParserMixin
@@ -9,9 +9,6 @@ from astrbot.core.utils.plugin_kv_store import PluginKVStoreMixin
 
 from .star import StarMetadata, star_map, star_registry
 
-if TYPE_CHECKING:
-    from .context import Context
-
 logger = logging.getLogger("astrbot")
 
 
@@ -20,9 +17,11 @@ class Star(CommandParserMixin, PluginKVStoreMixin):
 
     author: str
     name: str
-    context: Context
 
-    def __init__(self, context: Context, config: dict | None = None) -> None:
+    class _ContextLike(Protocol):
+        def get_config(self, umo: str | None = None) -> Any: ...
+
+    def __init__(self, context: _ContextLike, config: dict | None = None) -> None:
         self.context = context
 
     def _get_context_config(self) -> Any:

tests/test_qqofficial_webhook_signature.py

@@ -1,124 +0,0 @@
-import asyncio
-import json
-
-import pytest
-
-from astrbot.core.platform.sources.qqofficial_webhook.qo_webhook_server import (
-    _SIGNATURE_HEADER,
-    _SIGNATURE_TIMESTAMP_HEADER,
-    QQOfficialWebhook,
-    _sign_qq_webhook_payload,
-    _verify_qq_webhook_signature,
-)
-
-
-class FakeRequest:
-    def __init__(self, body: bytes, headers: dict[str, str] | None = None) -> None:
-        self._body = body
-        self.headers = headers or {}
-
-    async def get_data(self) -> bytes:
-        return self._body
-
-
-class FakeBotpyClient:
-    api = None
-    http = None
-
-    def ws_dispatch(self, *_args, **_kwargs) -> None:
-        return None
-
-
-def test_qq_webhook_signature_verification_accepts_valid_signature():
-    secret = "test-secret"
-    timestamp = "1710000000"
-    body = b'{"op":12,"d":0}'
-    signature = _sign_qq_webhook_payload(secret, timestamp, body)
-
-    assert _verify_qq_webhook_signature(secret, timestamp, signature, body)
-
-
-def test_qq_webhook_signature_verification_rejects_tampered_body():
-    secret = "test-secret"
-    timestamp = "1710000000"
-    body = b'{"op":12,"d":0}'
-    signature = _sign_qq_webhook_payload(secret, timestamp, body)
-
-    assert not _verify_qq_webhook_signature(
-        secret,
-        timestamp,
-        signature,
-        b'{"op":12,"d":1}',
-    )
-
-
-@pytest.mark.asyncio
-async def test_qq_webhook_callback_rejects_missing_signature():
-    webhook = object.__new__(QQOfficialWebhook)
-    webhook.secret = "test-secret"
-
-    result = await webhook.handle_callback(FakeRequest(b'{"op":12,"d":0}'))
-
-    assert result == ({"error": "Invalid signature"}, 401)
-
-
-@pytest.mark.asyncio
-async def test_qq_webhook_callback_accepts_signed_validation():
-    secret = "test-secret"
-    event_ts = "1710000000"
-    plain_token = "plain-token"
-    body = json.dumps(
-        {"op": 13, "d": {"event_ts": event_ts, "plain_token": plain_token}},
-        separators=(",", ":"),
-    ).encode("utf-8")
-    signature = _sign_qq_webhook_payload(secret, event_ts, body)
-    webhook = object.__new__(QQOfficialWebhook)
-    webhook.secret = secret
-
-    result = await webhook.handle_callback(
-        FakeRequest(
-            body,
-            {
-                _SIGNATURE_TIMESTAMP_HEADER: event_ts,
-                _SIGNATURE_HEADER: signature,
-            },
-        )
-    )
-
-    assert result == {
-        "plain_token": plain_token,
-        "signature": _sign_qq_webhook_payload(secret, event_ts, plain_token.encode()),
-    }
-
-
-@pytest.mark.asyncio
-async def test_qq_webhook_callback_lazily_creates_botpy_connection():
-    secret = "test-secret"
-    timestamp = "1710000000"
-    body = json.dumps(
-        {"op": 0, "t": "UNKNOWN_EVENT", "id": "event-id", "d": {"id": "message-id"}},
-        separators=(",", ":"),
-    ).encode("utf-8")
-    signature = _sign_qq_webhook_payload(secret, timestamp, body)
-    webhook = QQOfficialWebhook(
-        {"appid": "123", "secret": secret},
-        asyncio.Queue(),
-        FakeBotpyClient(),
-    )
-
-    result = await webhook.handle_callback(
-        FakeRequest(
-            body,
-            {
-                _SIGNATURE_TIMESTAMP_HEADER: timestamp,
-                _SIGNATURE_HEADER: signature,
-            },
-        )
-    )
-
-    assert result == {"opcode": 12}
-    assert webhook._connection is not None
-    assert webhook.http._token is not None
-    assert webhook.http._token.app_id == "123"
-    assert webhook.client.api is webhook.api
-    assert webhook.client.http is webhook.http