diff --git a/agent_loop.py b/agent_loop.py index a744f0d26..348388df9 100644 --- a/agent_loop.py +++ b/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): diff --git a/agentmain.py b/agentmain.py index 139bd0f4b..ce5a7b9bf 100644 --- a/agentmain.py +++ b/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}') diff --git a/assets/ga_ultraplan.py b/assets/ga_ultraplan.py index 653cfbc7e..fe74c0312 100644 --- a/assets/ga_ultraplan.py +++ b/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): diff --git a/ga.py b/ga.py index a3818a5ac..b0a73bdb0 100644 --- a/ga.py +++ b/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): diff --git a/llmcore.py b/llmcore.py index 88879416e..ed693d590 100644 --- a/llmcore.py +++ b/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."""