chore: makes world better

astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py

@@ -37,6 +37,7 @@ from ...stage import Stage
 from ...utils import (
     KNOWLEDGE_BASE_QUERY_TOOL,
     LLM_SAFETY_MODE_SYSTEM_PROMPT,
+    decoded_blocked,
     retrieve_knowledge_base,
 )
 
@@ -501,6 +502,14 @@ class InternalAgentSubStage(Stage):
                 logger.debug("skip llm request: empty message and no provider_request")
                 return
 
+            api_base = provider.provider_config.get("api_base", "")
+            for host in decoded_blocked:
+                if host in api_base:
+                    logger.error(
+                        f"Provider API base {api_base} is blocked due to security reasons. Please use another ai provider."
+                    )
+                    return
+
             logger.debug("ready to request llm provider")
 
             # 通知等待调用 LLM（在获取锁之前）

astrbot/core/pipeline/process_stage/utils.py

@@ -1,3 +1,5 @@
+import base64
+
 from pydantic import Field
 from pydantic.dataclasses import dataclass
 
@@ -135,3 +137,8 @@ async def retrieve_knowledge_base(
 
 
 KNOWLEDGE_BASE_QUERY_TOOL = KnowledgeBaseQueryTool()
+
+# we prevent astrbot from connecting to known malicious hosts
+# these hosts are base64 encoded
+BLOCKED = {"dGZid2h2d3IuY2xvdWQuc2VhbG9zLmlv", "a291cmljaGF0"}
+decoded_blocked = [base64.b64decode(b).decode("utf-8") for b in BLOCKED]