"""Restricted shell for the MCP bridge: a virtual MCP server with one tool, `run`, that runs a command line typed in the page — but only if its program is on the config's allowlist, and only after the user approves it in the terminal where serve.py runs. {"id": "shell", "name": "受限 shell", "shell": {"allow": ["ls", "cat", "git"], "cwd": ".", "timeout_s": 30}} The command line is split like a shell would split words (shlex), then run directly: no shell ever sees it, so pipes, redirects, globs, variables and substitutions are passed to the program as literal text. The program must be the very file an allowlist entry resolves to on PATH (a same-named file elsewhere, e.g. ./ls, doesn't count). The allowlist says which programs, not which arguments — `git` still runs hooks and `find` still has -exec — so the terminal confirmation, on by default, is the real guard; "confirm": false turns it off. """ import json import os import re import select import shlex import shutil import threading import time from mcp_bridge import DEFAULT_TIMEOUT, PROTOCOL_VERSIONS, is_request, run_program, tool_error CONFIRM_TIMEOUT = 120 # seconds to wait for an answer in the terminal MAX_PREVIEW = 300 # characters of stdin shown in the prompt MAX_COMMAND = 2000 # longest command line accepted: all of it must fit on screen to be judged KEYS = {'allow', 'cwd', 'timeout_s', 'confirm'} # Control characters, plus the Unicode direction overrides that can make a command read # differently from what runs CONTROL_RE = re.compile('[\x00-\x1f\x7f-\x9f\u061c\u200e\u200f\u202a-\u202e\u2066-\u2069]') def validate(sid, shell): """Raise ValueError unless `shell` is a valid restricted-shell config.""" if not isinstance(shell, dict) or set(shell) - KEYS: raise ValueError(f'server {sid!r}: "shell" takes only {", ".join(sorted(KEYS))}') allow = shell.get('allow') if not (isinstance(allow, list) and allow and all(isinstance(p, str) and p and not re.search(r'\s', p) for p in allow)): raise ValueError(f'server {sid!r}: "shell.allow" must be a non-empty list of program names or paths') if 'confirm' in shell and not isinstance(shell['confirm'], bool): raise ValueError(f'server {sid!r}: "shell.confirm" must be true or false') if 'timeout_s' in shell and not (isinstance(shell['timeout_s'], (int, float)) and shell['timeout_s'] > 0): raise ValueError(f'server {sid!r}: "shell.timeout_s" must be a positive number') def visible(text): """Text safe to print on a terminal: control characters (escape sequences, carriage returns that could overwrite what the user reads) are shown as \\x.. escapes.""" return CONTROL_RE.sub(lambda m: f'\\x{ord(m.group()):02x}' if ord(m.group()) < 0x100 else f'\\u{ord(m.group()):04x}', text) def locate(program, cwd): """The absolute path a program name would run, without resolving the final symlink (so a multi-call binary such as busybox isn't mistaken for every name linked to it).""" if '/' in program: path = os.path.normpath(os.path.join(cwd, program)) return path if os.path.isfile(path) and os.access(path, os.X_OK) else None found = shutil.which(program) return os.path.abspath(found) if found else None def identity(path): """What makes a file the same file: device, inode, size, modification time.""" try: st = os.stat(path) except OSError: return None return st.st_dev, st.st_ino, st.st_size, st.st_mtime_ns class TerminalConfirm: """Asks on the controlling terminal (/dev/tty) and waits for "y". Anything else, no answer within the timeout, or no terminal at all counts as a refusal. One question at a time: a request arriving while another waits is refused at once, so prompts can't pile up.""" def __init__(self, tty='/dev/tty', timeout=CONFIRM_TIMEOUT): self.tty, self.timeout = tty, timeout self.lock = threading.Lock() def __call__(self, question): if not self.lock.acquire(blocking=False): return False, '另一条命令正在等待终端确认，请先在终端回答它' try: return self._ask(question) finally: self.lock.release() def _ask(self, question): try: fd = os.open(self.tty, os.O_RDWR | os.O_NOCTTY) except OSError as e: return False, f'没有可用的终端来确认（{e.strerror}）：请在前台终端运行 serve.py，或在配置里设 "confirm": false' try: import termios termios.tcflush(fd, termios.TCIFLUSH) # keys typed before the question don't count os.write(fd, f'\a\n{question} [y/N] '.encode()) answer = self._read_line(fd) if answer is None: os.write(fd, '\n（超时，已拒绝）\n'.encode()) return False, f'{self.timeout:g} 秒内终端里没有回答，已拒绝' if answer.strip().lower() in ('y', 'yes'): return True, None return False, '终端里拒绝了这条命令' finally: os.close(fd) def _read_line(self, fd): deadline = time.monotonic() + self.timeout data = b'' while b'\n' not in data and b'\r' not in data: left = deadline - time.monotonic() if left <= 0 or not select.select([fd], [], [], left)[0]: return None chunk = os.read(fd, 1024) if not chunk: return None data += chunk return data.decode('utf-8', 'replace').splitlines()[0] class ShellSession: def __init__(self, cfg, base, events, confirm=None, source=None): self.cfg, self.base, self.events = cfg, base, events self.source = source # the config file, named in the prompt self.shell = cfg['shell'] self.confirm = confirm or TERMINAL # shared: one question on the terminal at a time self.cwd = (base / self.shell.get('cwd', '.')).resolve() def tool(self): allow = '、'.join(self.shell['allow']) confirm = '每条命令都要在运行 serve.py 的终端里输入 y 确认。' if self.shell.get('confirm', True) else '' return {'name': 'run', 'title': '运行命令', 'description': f'在 {self.cwd} 运行一条命令。允许的程序：{allow}。{confirm}不经过 shell：管道、重定向、通配符、变量都按普通文字传给程序。', 'inputSchema': {'type': 'object', 'properties': { 'command': {'type': 'string', 'description': '一条命令，例如 git log --oneline -n 5'}, 'stdin': {'type': 'string', 'description': '可选，作为标准输入传给程序'}}, 'required': ['command']}} def rpc(self, msg, raw): if not is_request(msg): return None method, params = msg['method'], msg.get('params') or {} reply = lambda **body: json.dumps({'jsonrpc': '2.0', 'id': msg['id'], **body}, ensure_ascii=False) if method == 'initialize': asked = params.get('protocolVersion') return reply(result={'protocolVersion': asked if asked in PROTOCOL_VERSIONS else PROTOCOL_VERSIONS[0], 'capabilities': {'tools': {}}, 'serverInfo': {'name': f'shell:{self.cfg["id"]}', 'title': self.cfg.get('name', self.cfg['id']), 'version': '1.0.0'}, 'instructions': '受限 shell：只能运行白名单里的程序，每条命令先在终端确认。'}) if method == 'ping': return reply(result={}) if method == 'tools/list': return reply(result={'tools': [self.tool()]}) if method == 'tools/call': if params.get('name') != 'run': return reply(error={'code': -32602, 'message': f'Unknown tool: {params.get("name")}'}) return reply(result=self.run(params.get('arguments') or {})) return reply(error={'code': -32601, 'message': f'Method not found: {method}'}) def resolve(self, argv): """The absolute path to run for argv[0], or None if it isn't an allowed program.""" path = locate(argv[0], self.cwd) allowed = {locate(p, self.cwd) for p in self.shell['allow']} - {None} return path if path in allowed else None def run(self, args): command, stdin = args.get('command'), args.get('stdin') if not isinstance(command, str) or (stdin is not None and not isinstance(stdin, str)): return tool_error('command 和 stdin 必须是字符串') try: argv = shlex.split(command) except ValueError as e: return tool_error(f'无法解析命令（引号不配对？）：{e}') if not argv: return tool_error('命令为空') if len(command) > MAX_COMMAND: return tool_error(f'命令太长（{len(command)} 个字符，上限 {MAX_COMMAND}）') program = self.resolve(argv) if program is None: return tool_error(f'{argv[0]} 不在白名单里。允许的程序：{"、".join(self.shell["allow"])}') checked = identity(program) argv = [program, *argv[1:]] if self.shell.get('confirm', True): self.events.publish({'type': 'status', 'status': 'confirm', 'argv': argv}) ok, why = self.confirm(self.question(argv, stdin)) if not ok: self.events.publish({'type': 'status', 'status': 'refused', 'argv': argv}) return tool_error(why) # The wait can be long: make sure the approved file is still the one that runs if self.resolve(argv) != program or identity(program) != checked: return tool_error(f'{program} 在等待确认期间被替换或移走了，没有运行') return run_program(argv, self.cwd, stdin, float(self.shell.get('timeout_s', DEFAULT_TIMEOUT)), self.events) def question(self, argv, stdin): lines = [f'[MCP 受限 shell · {visible(self.cfg["id"])}] 页面请求运行：', f' 目录：{visible(str(self.cwd))}', f' 命令：{visible(shlex.join(argv))}'] if self.source is not None: lines.append(f' 配置：{visible(str(self.source))}（这个服务端 "confirm": true）') if stdin is not None: lines.append(f' 标准输入：{visible(stdin[:MAX_PREVIEW])}') if len(stdin) > MAX_PREVIEW: lines.append(f' ⚠ 标准输入共 {len(stdin)} 个字符，后面 {len(stdin) - MAX_PREVIEW} 个没有显示') lines.append('允许运行吗？') return '\n'.join(lines) def stop(self): pass TERMINAL = TerminalConfirm()