import type { AgentCheckedChange, AgentToolRequest } from '../shared/agentTools' export function agentRemoteFilesScript(input: { root: string; request: AgentToolRequest } | { changes: Array }): string { const encoded = Buffer.from(JSON.stringify(input)).toString('base64') return `python3 - <<'LMS_AGENT_FILES' import sys sys.argv = ['lms-agent', '${encoded}'] ${agentRemoteFilesPython} LMS_AGENT_FILES` } export const agentRemoteFilesPython = String.raw` import base64, hashlib, json, os, re, stat, sys, tempfile args = json.loads(base64.b64decode(sys.argv[1])) blocked = {'node_modules', 'build', 'install', 'log', 'logs', '__pycache__', 'id_rsa', 'id_ed25519', 'credentials'} text_pattern = re.compile(r'\.(c|cc|cpp|cxx|h|hh|hpp|hxx|py|js|jsx|ts|tsx|json|ya?ml|xml|md|txt|toml|ini|cfg|conf|sh|bash|launch|urdf|xacro|rviz|proto|cmake)$|^(Dockerfile|CMakeLists\.txt)$', re.I) def safe(value): if not isinstance(value, str) or not value or len(value) > 500 or value.startswith('/') or any(c in value for c in '\\\x00\r\n'): raise ValueError('Invalid source path') if any(not p or p.startswith('.') or p.lower() in blocked or re.search(r'\.(pem|key|p12|pfx)$', p, re.I) for p in value.split('/')): raise ValueError('Private or excluded source path') return value def checked(root, path, missing=False): safe(path) root = os.path.realpath(root) current = root parts = path.split('/') for i, part in enumerate(parts): current = os.path.join(current, part) try: info = os.lstat(current) if stat.S_ISLNK(info.st_mode) or (i < len(parts)-1 and not stat.S_ISDIR(info.st_mode)): raise ValueError('Symlink or non-directory source path') except FileNotFoundError: if not (missing and i == len(parts)-1): raise if os.path.commonpath([root, current]) != root: raise ValueError('Source path escapes module') return current def read(root, path): if not text_pattern.search(os.path.basename(path)): raise ValueError('Only text source files are supported') absolute = checked(root, path) info = os.lstat(absolute) if not stat.S_ISREG(info.st_mode) or info.st_size > 1048576: raise ValueError('Only regular source files up to 1 MB are supported') with open(absolute, 'rb') as fp: content = fp.read(1048577) if len(content) > 1048576 or b'\x00' in content: raise ValueError('Binary or oversized file') return content.decode('utf-8') def digest(content): return hashlib.sha256(content.encode('utf-8')).hexdigest() if 'changes' in args: changes = args['changes'] if not changes or len(changes) > 24 or sum(len(c['content']) for c in changes) > 1000000: raise ValueError('Invalid change batch size') prepared, committed, seen = [], [], set() try: for c in changes: path = checked(c['root'], c['path'], True) if path in seen or not text_pattern.search(os.path.basename(path)): raise ValueError('Duplicate or unsupported change file') seen.add(path) try: original = read(c['root'], c['path']) mode = stat.S_IMODE(os.stat(path).st_mode) except FileNotFoundError: original, mode = None, 0o644 if (None if original is None else digest(original)) != c.get('expectedHash', 'missing'): raise ValueError('File changed after analysis: ' + c['moduleName'] + '/' + c['path']) prepared.append({'path': path, 'original': original, 'mode': mode, 'content': c['content'], 'temp': None}) for p in prepared: fd, p['temp'] = tempfile.mkstemp(prefix='.lms-agent-', dir=os.path.dirname(p['path'])) with os.fdopen(fd, 'w', encoding='utf-8', newline='') as fp: fp.write(p['content']) os.chmod(p['temp'], p['mode']) for p in prepared: try: with open(p['path'], 'r', encoding='utf-8', newline='') as fp: latest = fp.read() except FileNotFoundError: latest = None if latest != p['original']: raise ValueError('File changed immediately before apply') for p in prepared: os.replace(p['temp'], p['path']) committed.append(p) except Exception as error: failed = [] for p in reversed(committed): try: if p['original'] is None: os.unlink(p['path']) else: with open(p['path'], 'w', encoding='utf-8', newline='') as fp: fp.write(p['original']) os.chmod(p['path'], p['mode']) except Exception: failed.append(p['path']) if failed: raise ValueError('Rollback failed: ' + ', '.join(failed)) from error raise finally: for p in prepared: if p['temp'] and os.path.exists(p['temp']): os.unlink(p['temp']) print(json.dumps({'ok': True})) else: root = os.path.realpath(args['root']) req = args['request'] options = req.get('args') or {} name = req['name'] if name == 'files.read': path = options['path'] try: content = read(root, path) lines = content.split('\n') partial = 'startLine' in options or 'lineCount' in options or len(content) > 40000 start = options.get('startLine', 1) selected = '\n'.join(lines[start-1:start-1+options.get('lineCount', 200)])[:40000] if partial else content data = {'path': path, 'exists': True, 'hash': digest(content), 'content': selected, 'complete': not partial, 'startLine': start, 'totalLines': len(lines)} except FileNotFoundError: checked(root, path, True) data, partial = {'path': path, 'exists': False, 'hash': None, 'content': '', 'complete': True}, False print(json.dumps({'data': data, 'truncated': partial}, ensure_ascii=False)) else: prefix = options.get('path', '') if prefix: checked(root, prefix) paths, matches = [], [] visited, scanned, truncated = 0, 0, False def fail(error): raise error for directory, dirs, files in os.walk(root, followlinks=False, onerror=fail): depth = len(os.path.relpath(directory, root).split('/')) if directory != root else 0 kept = [] for d in sorted(dirs): rel = os.path.relpath(os.path.join(directory, d), root) if prefix and rel != prefix and not rel.startswith(prefix + '/') and not prefix.startswith(rel + '/'): continue visited += 1 try: safe(rel) if not os.path.islink(os.path.join(directory, d)): kept.append(d) except ValueError: pass dirs[:] = kept if depth < 10 else [] for f in sorted(files): path = os.path.relpath(os.path.join(directory, f), root) if prefix and path != prefix and not path.startswith(prefix + '/'): continue visited += 1 if visited > 5000 or len(paths) >= 250 or len(matches) >= 80 or scanned >= 4000000: truncated = True break try: safe(path) if os.path.islink(os.path.join(directory, f)) or not text_pattern.search(f): continue if prefix and path != prefix and not path.startswith(prefix + '/'): continue if name == 'files.list': paths.append(path) continue content = read(root, path) except (ValueError, OSError, UnicodeError): continue scanned += len(content) for number, line in enumerate(content.split('\n'), 1): if options['query'].lower() in line.lower(): matches.append({'path': path, 'line': number, 'text': line[:500]}) if len(matches) >= 80: truncated = True break if truncated: break data = {'paths': paths} if name == 'files.list' else {'query': options['query'], 'matches': matches} print(json.dumps({'data': data, 'truncated': truncated}, ensure_ascii=False)) `