Fix critical bugs: JSON crash, data loss, thread safety, security vulnerabilities

agent_loop.py

@@ -67,8 +67,14 @@ def agent_runner_loop(client, system_prompt, user_input, handler, tools_schema,
         _hook('llm_after', locals())
 
         if not response.tool_calls: tool_calls = [{'tool_name': 'no_tool', 'args': {}}]
-        else: tool_calls = [{'tool_name': tc.function.name, 'args': json.loads(tc.function.arguments), 'id': tc.id}
-                          for tc in response.tool_calls]
+        else:
+            tool_calls = []
+            for tc in response.tool_calls:
+                try: args = json.loads(tc.function.arguments)
+                except (json.JSONDecodeError, TypeError):
+                    args = {'_raw': tc.function.arguments}
+                    yield f"⚠️ Tool '{tc.function.name}' args parse failed, raw={repr(tc.function.arguments)[:200]}\n"
+                tool_calls.append({'tool_name': tc.function.name, 'args': args, 'id': tc.id})
 
         tool_results = []; next_prompts = set(); exit_reason = {}
         for ii, tc in enumerate(tool_calls):

agentmain.py

@@ -17,7 +17,7 @@
 BANNED_TOOLS = (['ask_user', 'start_long_term_update'] if '--no-user-tools' in sys.argv else [])
 def load_tool_schema(suffix=''):
     global TOOLS_SCHEMA
-    TS = open(os.path.join(script_dir, f'assets/tools_schema{suffix}.json'), 'r', encoding='utf-8').read()
+    with open(os.path.join(script_dir, f'assets/tools_schema{suffix}.json'), 'r', encoding='utf-8') as f: TS = f.read()
     TOOLS_SCHEMA = json.loads(TS if os.name == 'nt' else TS.replace('powershell', 'bash'))
     TOOLS_SCHEMA = [t for t in TOOLS_SCHEMA if t.get('function', {}).get('name') not in BANNED_TOOLS]
 load_tool_schema()
@@ -26,11 +26,12 @@ def load_tool_schema(suffix=''):
 mem_dir = os.path.join(script_dir, 'memory')
 if not os.path.exists(mem_dir): os.makedirs(mem_dir)
 mem_txt = os.path.join(mem_dir, 'global_mem.txt')
-if not os.path.exists(mem_txt): open(mem_txt, 'w', encoding='utf-8').write('# [Global Memory - L2]\n')
+    with open(mem_txt, 'w', encoding='utf-8') as f: f.write('# [Global Memory - L2]\n')
 mem_insight = os.path.join(mem_dir, 'global_mem_insight.txt')
 if not os.path.exists(mem_insight):
-    t = os.path.join(script_dir, f'assets/global_mem_insight_template{lang_suffix}.txt')
-    open(mem_insight, 'w', encoding='utf-8').write(open(t, encoding='utf-8').read() if os.path.exists(t) else '')
+    with open(mem_insight, 'w', encoding='utf-8') as fw:
+        if os.path.exists(t):
+            with open(t, encoding='utf-8') as fr: fw.write(fr.read())
 
 def get_system_prompt():
     with open(os.path.join(script_dir, f'assets/sys_prompt{lang_suffix}.txt'), 'r', encoding='utf-8') as f: prompt = f.read()
@@ -110,11 +111,14 @@ def get_llm_name(self, b=None, model=False):
     def get_ctx_multiplier(self): return getattr(self.llmclient.backend, 'maxlen_multiplier', 1.0)
 
     def abort(self):
-        if not self.is_running: return
-        print('Abort current task...')
-        self.stop_sig = True
-        if self.handler is not None: self.handler.code_stop_signal.append(1)
-        for sess in getattr(self.llmclient.backend, '_sessions', [self.llmclient.backend]):
+        with self.lock:
+            if not self.is_running: return
+            print('Abort current task...')
+            self.stop_sig = True
+            handler = self.handler
+            client = self.llmclient
+        if handler is not None: handler.code_stop_signal.append(1)
+        for sess in getattr(client.backend, '_sessions', [client.backend]):
             sess.should_stop = lambda: self.stop_sig  # live read; cleared by run()'s finally
             try: sess.active_response.close()
             except Exception: pass
@@ -130,7 +134,8 @@ def _handle_slash_cmd(self, raw_query, display_queue):
         if _sm := re.match(r'/session\.(\w+)=(.*)', raw_query.strip()):
             k, v = _sm.group(1), _sm.group(2)
             vfile = os.path.join(script_dir, 'temp', v)
-            if os.path.isfile(vfile): v = open(vfile, encoding='utf-8').read().strip()
+            if os.path.isfile(vfile):
+                with open(vfile, encoding='utf-8') as vf: v = vf.read().strip()
             try: v = json.loads(v)  # cover number parsing
             except (json.JSONDecodeError, ValueError): pass
             setattr(self.llmclient.backend, k, v)
@@ -148,7 +153,9 @@ def run(self):
             raw_query = self._handle_slash_cmd(raw_query, display_queue)
             if raw_query is None:
                 self.task_queue.task_done(); continue
-            self.is_running = True; self._current_queue = display_queue
+            with self.lock:
+                self.is_running = True
+                self._current_queue = display_queue
             if len(raw_query) > 2000:
                 task_file = os.path.join(script_dir, 'temp', f'user_prompt_{os.getpid()}_{time.time_ns()}.md')
                 with open(task_file, 'w', encoding='utf-8') as f: f.write(raw_query)
@@ -194,8 +201,9 @@ def run(self):
                 print(f"Backend Error: {format_error(e)}")
                 display_queue.put({'done': full_resp + f'\n```\n{format_error(e)}\n```', 'source': source, 'turn': curr_turn, 'outputs': turn_resps.copy()})
             finally:
-                if self.stop_sig: print('User aborted the task.')
-                self.is_running = self.stop_sig = False  # keep _current_queue: its final 'done' may still be unclaimed (refreshed UI salvages it); next task overwrites it
+                with self.lock:
+                    if self.stop_sig: print('User aborted the task.')
+                    self.is_running = self.stop_sig = False  # keep _current_queue: its final 'done' may still be unclaimed (refreshed UI salvages it); next task overwrites it
                 self.task_queue.task_done()
                 if self.handler is not None: self.handler.code_stop_signal.append(1)
 
@@ -249,7 +257,8 @@ def run(self):
     elif args.func:
         infile = args.func; outfile = os.path.splitext(args.func)[0] + '.out.txt'
 
-    if histfile and os.path.isfile(histfile): agent.llmclient.backend.history = json.loads(open(histfile, encoding='utf-8').read())
+    if histfile and os.path.isfile(histfile):
+        with open(histfile, encoding='utf-8') as f: agent.llmclient.backend.history = json.loads(f.read())
 
     if args.func or args.task:
         agent.peer_hint = False
@@ -306,7 +315,8 @@ def _hub_put(t):
                     print(f'[Reflect] drain error: {e}'); result = f'[ERROR] {e}'
                 log_dir = os.path.join(script_dir, 'temp/reflect_logs'); os.makedirs(log_dir, exist_ok=True)
                 script_name = os.path.splitext(os.path.basename(args.reflect))[0]
-                open(os.path.join(log_dir, f'{script_name}_{datetime.now():%Y-%m-%d}.log'), 'a', encoding='utf-8').write(f'[{datetime.now():%m-%d %H:%M}]\n{result}\n\n')
+                with open(os.path.join(log_dir, f'{script_name}_{datetime.now():%Y-%m-%d}.log'), 'a', encoding='utf-8') as lf:
+                    lf.write(f'[{datetime.now():%m-%d %H:%M}]\n{result}\n\n')
                 if (on_done := getattr(mod, 'on_done', None)):
                     try: on_done(result)
                     except Exception as e: print(f'[Reflect] on_done error: {e}')

assets/ga_ultraplan.py

@@ -1,3 +1,5 @@
+import secrets
+import hmac
 from contextlib import contextmanager, redirect_stdout, redirect_stderr
 from concurrent.futures import ThreadPoolExecutor
 from time import time, sleep
@@ -8,6 +10,11 @@
 
 _ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 _PORT = int(os.environ.get("GA_ULTRAPLAN_PORT", "47831"))
+_AUTH_TOKEN = os.environ.get("GA_ULTRAPLAN_TOKEN", secrets.token_urlsafe(32))
+# SECURITY: Log token on startup for first-time setup
+if "GA_ULTRAPLAN_TOKEN" not in os.environ:
+    print(f"[SECURITY] No GA_ULTRAPLAN_TOKEN set. Generated token: {_AUTH_TOKEN}")
+    print(f"[SECURITY] Set GA_ULTRAPLAN_TOKEN env var to authenticate requests.")
 _T0 = time(); _phases = []; _phase_stack = []; _tasks = []; _current = "idle"; _events = []; _srv = None; _last = time(); _lock = threading.Lock(); _exec_lock = threading.Lock()
 _TASK_SLUG = "task"; _FUNC_SEQ = 0; _PLANNED = False; _SESSION = None; _sessions = {}
 _RUN_DIR = os.path.abspath(os.environ.get("GA_ULTRAPLAN_RUNDIR", os.path.join(_ROOT, "temp", "ultraplan_default")))
@@ -71,6 +78,11 @@ def do_GET(self):
     def do_POST(self):
         global _TASK_SLUG, _PLANNED
         if self.path != "/exec": self.send_response(404); self.end_headers(); return
+        # SECURITY: Require authentication token
+        auth = self.headers.get("Authorization", "")
+        if not auth.startswith("Bearer ") or not hmac.compare_digest(auth[7:], _AUTH_TOKEN):
+            self.send_response(401); self.send_header("Content-Type", "application/json")
+            self.end_headers(); self.wfile.write(json.dumps({"error": "Unauthorized"}).encode()); return
         n = int(self.headers.get("Content-Length", "0")); req = json.loads(self.rfile.read(n).decode("utf-8"))
         out = io.StringIO(); err = io.StringIO(); rc = 0
         with _exec_lock, redirect_stdout(out), redirect_stderr(err):

ga.py

@@ -1,4 +1,5 @@
 import sys, os, re, json, time, threading, importlib, webbrowser
+import builtins
 from datetime import datetime
 from pathlib import Path
 import tempfile, traceback, subprocess, itertools, collections, difflib, shutil
@@ -23,8 +24,8 @@ def code_run(code, code_type="python", timeout=60, cwd=None, code_cwd=None, stop
     cwd = cwd or os.path.join(script_dir, 'temp'); tmp_path = None
     if code_type in ["python", "py"]:
         tmp_file = tempfile.NamedTemporaryFile(suffix=".ai.py", delete=False, mode='w', encoding='utf-8', dir=code_cwd)
-        cr_header = os.path.join(script_dir, 'assets', 'code_run_header.py')
-        if os.path.exists(cr_header): tmp_file.write(open(cr_header, encoding='utf-8').read())
+        if os.path.exists(cr_header):
+            with open(cr_header, encoding='utf-8') as hf: tmp_file.write(hf.read())
         tmp_file.write(code)
         tmp_path = tmp_file.name
         tmp_file.close()
@@ -314,7 +315,19 @@ def do_code_run(self, args, response):
         maxlen = self._get_tool_maxlen(10000, args)
         if timeout > 600: result = '[ERROR] Timeout must be <= 600 seconds; code not executed. Run time-consuming code in the background instead of waiting for it to finish in the foreground, verify it started successfully, and monitor it until completion or failure.'
         elif code_type == 'python' and _arg(args, "inline_eval", False, bool):
-            ns = {'handler':self, 'parent':self.parent, 'history':json.dumps(self.parent.llmclient.backend.history)}
+            # SECURITY: Create sandboxed namespace - remove dangerous builtins and history exposure
+            safe_builtins = {
+                'print': print, 'len': len, 'range': range, 'int': int, 'float': float,
+                'str': str, 'bool': bool, 'list': list, 'dict': dict, 'tuple': tuple,
+                'set': set, 'type': type, 'isinstance': isinstance, 'hasattr': hasattr,
+                'getattr': getattr, 'dir': dir, 'vars': vars,
+                'repr': repr, 'abs': abs, 'min': min, 'max': max, 'sum': sum,
+                'enumerate': enumerate, 'zip': zip, 'map': map, 'filter': filter,
+                'sorted': sorted, 'reversed': reversed, 'any': any, 'all': all,
+                'True': True, 'False': False, 'None': None,
+            }
+            ns = {'handler': self, 'parent': self.parent, '__builtins__': safe_builtins}
+            # WARNING: Do NOT expose history - it contains sensitive conversation data
             old_cwd = os.getcwd()
             try:
                 os.chdir(cwd)
@@ -412,8 +425,18 @@ def extract_robust_content(text):
         try:
             new_content = expand_file_refs(content, base_dir=self.cwd)
             if mode == "prepend":
+                # BUG FIX: 原实现先open读再open写，写入异常时原文件已被清空导致数据丢失
+                # 改为先写临时文件再原子rename，保证异常时原文件不受影响
                 old = open(path, 'r', encoding="utf-8").read() if os.path.exists(path) else ""
-                open(path, 'w', encoding="utf-8", newline=_file_newline(path)).write(new_content + old)
+                tmp_fd, tmp_path = tempfile.mkstemp(dir=os.path.dirname(path) or '.', suffix='.tmp')
+                try:
+                    with os.fdopen(tmp_fd, 'w', encoding='utf-8', newline=_file_newline(path)) as tmp_f:
+                        tmp_f.write(new_content + old)
+                    os.replace(tmp_path, path)  # 原子操作，失败时原文件不受影响
+                except Exception:
+                    try: os.unlink(tmp_path)
+                    except OSError: pass
+                    raise
             else:
                 with open(path, 'a' if mode == "append" else 'w', encoding="utf-8", newline=_file_newline(path)) as f: f.write(new_content)
             yield f"[Status] ✅ {mode.capitalize()} 成功 ({len(new_content)} bytes)\n"
@@ -455,7 +478,8 @@ def enter_plan_mode(self, plan_path):
         return plan_path
     def _check_plan_completion(self):
         if not os.path.isfile(p:=self._in_plan_mode() or ''): return None
-        try: return len(re.findall(r'\[ \]', open(p, encoding='utf-8', errors='replace').read()))
+        try:
+            with open(p, encoding='utf-8', errors='replace') as f: return len(re.findall(r'\[ \]', f.read()))
         except: return None
 
     def do_update_working_checkpoint(self, args, response):

llmcore.py

@@ -1030,7 +1030,8 @@ def tryparse(json_str):
     try: return json.loads(json_str[:-1])
     except: pass
     if '}' in json_str: json_str = json_str[:json_str.rfind('}') + 1]
-    return json.loads(json_str)
+    try: return json.loads(json_str)
+    except (json.JSONDecodeError, ValueError): return {'_raw': json_str}
 
 class MixinSession:
     """A Session facade backed by multiple routed transport sessions."""