"""Sandboxed bot processes for the Chat tab, copied from the cmpatino/chatbot-swarm-chat Space (itself adapted from cmpatino/eliza-chat). Changes: 60 s per reply, each process runs in its own empty temp cwd.""" import os, select, shutil, signal, subprocess, sys, tempfile, threading, time HERE = os.path.dirname(os.path.abspath(__file__)) RUNNER = os.path.join(HERE, "stdio_runner.py") TURN_S = 60.0 # wall clock per reply FIRST_TURN_S = 60.0 # first reply also pays for interpreter start + loading every module CONV_S = 300.0 # wall clock per conversation (the world's contract: 300 s per conversation process) MEM_MB = 256 # address-space limit per bot process CPU_S = 60 # CPU seconds per bot process OUT_CAP = 64 * 1024 # a reply line over this kills the process MAX_RESTARTS = 3 REPLAY_S = 15.0 # budget for replaying earlier messages into a restarted process def _bot_user(): """Run bots as the unprivileged `bot` user when this server may switch uid (root); {} otherwise (e.g. Spaces, which runs the container as uid 1000: then seccomp + a non-dumpable server are the boundary).""" if os.geteuid() != 0: return {} try: import pwd u = pwd.getpwnam(os.environ.get("BOT_USER", "bot")) except (KeyError, ImportError): return {} return dict(user=u.pw_uid, group=u.pw_gid, extra_groups=[]) USER_KW = _bot_user() def _probe_netns(): """Use an empty network namespace too when the container allows it (most hosted containers don't).""" exe = shutil.which("unshare") if not exe: return [] try: r = subprocess.run([exe, "-rn", "true"], capture_output=True, timeout=5, **USER_KW) return [exe, "-rn"] if r.returncode == 0 else [] except Exception: return [] NETNS_PREFIX = _probe_netns() def _probe_seccomp(): """The directory holding pyseccomp.py if a filter loads and is enforced on this host, else None.""" try: import pyseccomp except Exception: return None d = os.path.dirname(os.path.abspath(pyseccomp.__file__)) try: r = subprocess.run([sys.executable, "-I", "-S", RUNNER, "--probe", d], capture_output=True, timeout=20, env={}, **USER_KW) except Exception: return None return d if r.returncode == 0 else None SECCOMP_DIR = _probe_seccomp() # fail closed: without an enforced seccomp filter no bot runs, unless CHAT_REQUIRE_SECCOMP=0 DISABLED = os.environ.get("CHAT_REQUIRE_SECCOMP", "1") != "0" and SECCOMP_DIR is None print(f"chat: seccomp {'on' if SECCOMP_DIR else 'OFF'}, bot uid {USER_KW.get('user', os.geteuid())}, " f"netns {'on' if NETNS_PREFIX else 'off'}{', DISABLED' if DISABLED else ''}", flush=True) class StdioProc: """One bot process: `ask` writes one line and reads one reply line.""" def __init__(self, root, cpu_s=CPU_S): if DISABLED: raise RuntimeError("chat disabled on this host (no seccomp)") cmd = NETNS_PREFIX + [sys.executable, "-I", "-S", "-u", "-B", RUNNER, str(MEM_MB), str(cpu_s), root, SECCOMP_DIR or ""] self.cwd = tempfile.mkdtemp(prefix="bot-cwd-") if USER_KW: # the bot's only writable place (and seccomp denies opening files for writing anyway) os.chown(self.cwd, USER_KW["user"], USER_KW["group"]) self.p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=self.cwd, env={"LANG": "C.UTF-8", "PYTHONHASHSEED": "random"}, start_new_session=True, close_fds=True, **USER_KW) self.out, self.errfd, self.inp = self.p.stdout.fileno(), self.p.stderr.fileno(), self.p.stdin.fileno() for fd in (self.out, self.errfd, self.inp): os.set_blocking(fd, False) self.buf, self.err, self.closed, self.turns = b"", b"", False, 0 def _read_ready(self, timeout, pending=b""): rl = [self.out] + ([self.errfd] if self.errfd is not None else []) r, w, _ = select.select(rl, [self.inp] if pending else [], [], max(0.0, timeout)) if w: try: pending = pending[os.write(self.inp, pending[:65536]):] except BlockingIOError: pass except OSError: pending = b"" eof = False for fd in r: try: chunk = os.read(fd, 65536) except BlockingIOError: continue if fd == self.out: eof |= not chunk self.buf += chunk elif chunk: self.err = (self.err + chunk)[-2000:] else: self.errfd = None return eof, pending def _died(self): try: code = self.p.wait(timeout=1) except subprocess.TimeoutExpired: code = None tail = [l for l in self.err.decode("utf-8", "replace").splitlines() if l.strip()] why = ("was killed (CPU or memory limit?)" if code in (-9, -24, -6, -11, 137) else "closed its output" if code is None else f"exited with code {code}") self.close() return {"fatal": f"crashed: it {why}" + (f" ({tail[-1][:200]})" if tail else "")} def ask(self, line, timeout=None): if self.closed or self.p.poll() is not None: return self._died() if timeout is None: timeout = FIRST_TURN_S if self.turns == 0 else TURN_S # discard stray output from earlier turns (the contract is exactly one line per input line) while select.select([self.out], [], [], 0)[0]: eof, _ = self._read_ready(0) if eof: break self.buf = b"" t0 = time.monotonic() pending = (line + "\n").encode("utf-8", "replace") while b"\n" not in self.buf: if len(self.buf) > OUT_CAP: self.close(); return {"fatal": f"wrote a reply longer than {OUT_CAP >> 10} KB"} left = t0 + timeout - time.monotonic() if left <= 0: self.close(); return {"fatal": f"did not reply within {timeout:.0f} s"} eof, pending = self._read_ready(left, pending) if eof and b"\n" not in self.buf: return self._died() self.turns += 1 reply, _, self.buf = self.buf.partition(b"\n") return {"reply": reply.rstrip(b"\r").decode("utf-8", "replace"), "ms": int((time.monotonic() - t0) * 1000)} def close(self): if self.closed: return self.closed = True try: if self.p.poll() is None: os.killpg(self.p.pid, signal.SIGKILL) except Exception: pass try: self.p.wait(timeout=2) except Exception: pass for f in (self.p.stdin, self.p.stdout, self.p.stderr): try: f.close() except Exception: pass shutil.rmtree(self.cwd, ignore_errors=True) class Conversation: """One conversation = one bot process fed one line per message. On a crash or timeout the process is restarted and the earlier messages are replayed into it (so it keeps its memory when it can). The whole conversation may last CONV_S seconds of wall clock, like the world's contract.""" def __init__(self, root): self.root, self.lines, self.restarts = root, [], 0 self.started = time.monotonic() self.lock = threading.Lock() self.ended = None self.proc = StdioProc(root) def time_left(self): return CONV_S - (time.monotonic() - self.started) def send(self, line): """-> {"reply", "ms"} | {"error", "notice"} (restarted) | {"fatal"} (conversation over)""" with self.lock: if self.ended: return {"fatal": self.ended} left = self.time_left() if left <= 0: return {"fatal": self._end(f"the conversation reached its {CONV_S / 60:.0f}-minute limit")} timeout = min(FIRST_TURN_S if self.proc.turns == 0 else TURN_S, left) r = self.proc.ask(line, timeout=timeout) if "fatal" not in r: self.lines.append(line); return r if self.time_left() <= 0: return {"fatal": self._end(f"the conversation reached its {CONV_S / 60:.0f}-minute limit")} self.restarts += 1 if self.restarts > MAX_RESTARTS: return {"fatal": self._end(f"the bot {r['fatal']}; it was already restarted {MAX_RESTARTS} times")} return {"error": r["fatal"], "notice": self._restart()} def _end(self, why): self.ended = why self.close() return why def _restart(self): self.proc.close() self.proc = StdioProc(self.root) if not self.lines: return "bot restarted" end = time.monotonic() + min(REPLAY_S, max(1.0, self.time_left())) for l in self.lines: first = self.proc.turns == 0 r = self.proc.ask(l, timeout=max(0.5, min(FIRST_TURN_S if first else TURN_S, end - time.monotonic()))) if "fatal" in r: self.proc.close(); self.proc = StdioProc(self.root) n, self.lines = len(self.lines), [] return f"bot restarted fresh (replaying the {n} earlier messages failed, so it forgot them)" return f"bot restarted and the {len(self.lines)} earlier messages were replayed into it" def close(self): if self.proc: self.proc.close()