fix(common,agent): unify frame size limit, quote YAML special chars, lock config removal in disable_restrictions

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: misakano7545

.github/workflows/build-windows.yml

@@ -19,7 +19,7 @@ jobs:
           - name: win7
             python-version: "3.8"
           - name: win10
-            python-version: "3.11"
+            python-version: "3.13"
 
     steps:
       - name: Checkout

.gitignore

@@ -18,4 +18,7 @@ configs/config.yaml
 agent/configs
 
 # Trae
-.trae/
+.trae/
+
+# Pycharm
+.idea/

agent/firewall_manager.py

@@ -308,9 +308,20 @@ def disable_restrictions() -> Tuple[bool, str]:
     if not is_admin():
         return False, "需要管理员权限"
 
-    _stop_clash_process()
-    if os.path.exists(CLASH_CONFIG_FILE):
-        os.remove(CLASH_CONFIG_FILE)
+    with CLASH_PROCESS_LOCK:
+        global _clash_process
+        if _clash_process is not None:
+            try:
+                _clash_process.terminate()
+                _clash_process.wait(timeout=5)
+            except Exception:
+                try:
+                    _clash_process.kill()
+                except Exception:
+                    pass
+            _clash_process = None
+        if os.path.exists(CLASH_CONFIG_FILE):
+            os.remove(CLASH_CONFIG_FILE)
 
     return True, "已解除网络限制"
 

common/clash_config.py

@@ -121,10 +121,29 @@ def build_clash_config(mode: str, rules: List[Dict[str, str]]) -> Dict[str, Any]
     return config
 
 
+def _yaml_quote(s: str) -> str:
+    """Quote a string value if it contains YAML-special characters."""
+    if not s:
+        return "''"
+    needs_quote = (
+        s.startswith(("{", "[", "'", '"', "*", "&", "!", "|", ">", "%", "@", "`"))
+        or "#" in s
+        or ": " in s
+        or s.endswith(":")
+        or s.startswith("- ")
+        or s.strip() != s
+        or s.lower() in ("true", "false", "null", "yes", "no", "on", "off")
+    )
+    if not needs_quote:
+        return s
+    escaped = s.replace("'", "''")
+    return f"'{escaped}'"
+
+
 def config_to_yaml(config: Dict[str, Any]) -> str:
     """Convert config dict to YAML string."""
     lines = []
-    
+
     def add_line(key: str, value, indent: int = 0):
         prefix = "  " * indent
         if isinstance(value, dict):
@@ -139,15 +158,15 @@ def add_line(key: str, value, indent: int = 0):
                     for k, v in item.items():
                         add_line(k, v, indent + 2)
                 else:
-                    lines.append(f"{prefix}- {item}")
+                    lines.append(f"{prefix}- {_yaml_quote(str(item))}")
         elif isinstance(value, bool):
             lines.append(f"{prefix}{key}: {str(value).lower()}")
         elif isinstance(value, int):
             lines.append(f"{prefix}{key}: {value}")
         else:
-            lines.append(f"{prefix}{key}: {value}")
-    
+            lines.append(f"{prefix}{key}: {_yaml_quote(str(value))}")
+
     for key, value in config.items():
         add_line(key, value)
-    
+
     return "\n".join(lines)

common/protocol.py

@@ -23,11 +23,13 @@
 MSG_REQUEST_CONFIG = "request_config"
 MSG_CONFIG_RESPONSE = "config_response"
 
+MAX_FRAME_SIZE = 1 * 1024 * 1024  # 1 MiB — JSON 命令帧不应超过此大小
+
 
 def encode_frame(obj: Dict[str, Any]) -> bytes:
     body = json.dumps(obj, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
-    if len(body) > 0xFFFFFF:
-        raise ValueError("frame too large")
+    if len(body) > MAX_FRAME_SIZE:
+        raise ValueError("frame too large (%d bytes, max %d)" % (len(body), MAX_FRAME_SIZE))
     return struct.pack(">I", len(body)) + body
 
 
@@ -59,8 +61,8 @@ def recv_exact(conn: socket.socket, n: int) -> bytes:
 def read_frame_from_socket(conn: socket.socket) -> Dict[str, Any]:
     header = recv_exact(conn, 4)
     (n,) = struct.unpack(">I", header)
-    if n > 16 * 1024 * 1024:
-        raise ValueError("frame size unreasonable")
+    if n > MAX_FRAME_SIZE:
+        raise ValueError("frame size unreasonable (%d bytes, max %d)" % (n, MAX_FRAME_SIZE))
     body = recv_exact(conn, n)
     obj = json.loads(body.decode("utf-8"))
     if not isinstance(obj, dict):