diff --git a/core b/core index c1124b5..42361e6 160000 --- a/core +++ b/core @@ -1 +1 @@ -Subproject commit c1124b588d01be470821efa7887f32517c3b0f96 +Subproject commit 42361e6505e607a9ee406226a423e5c1fac82d55 diff --git a/scripts/x/elevate b/scripts/x/elevate index 0c44eae..d56edca 100755 --- a/scripts/x/elevate +++ b/scripts/x/elevate @@ -2,8 +2,14 @@ """Sandbox escape hatch for execute_code elevated steps. Connects to the socket at OLLIE_ELEVATE_SOCKET, sends the command and cwd, -and returns the result. The adapter listening on that socket is responsible -for running the command outside the Landlock sandbox. +and streams output back in real-time. The adapter listening on that socket +is responsible for running the command outside the Landlock sandbox. + +Protocol (streaming): + Client sends: {"cmd": ..., "cwd": ...}\n then shutdown(WR) + Server sends frames: + type 'd' = output data (write to stdout immediately) + type 'x' = exit code (payload is ASCII integer) Usage: elevate -- """ @@ -12,6 +18,7 @@ sys.path = sys.path[1:] # strip script dir — may be a non-navigable fs (e.g. import json import os import socket +import struct def default_socket_path(): @@ -19,6 +26,17 @@ def default_socket_path(): return os.path.join(xdg, "ollie", "elevate.sock") +def recv_exact(sock, n): + """Read exactly n bytes from socket.""" + buf = b"" + while len(buf) < n: + chunk = sock.recv(n - len(buf)) + if not chunk: + return None + buf += chunk + return buf + + def main(): sock_path = os.environ.get("OLLIE_ELEVATE_SOCKET") or default_socket_path() @@ -38,16 +56,26 @@ def main(): with sock: sock.sendall(req) sock.shutdown(socket.SHUT_WR) - data = b"" - while chunk := sock.recv(65536): - data += chunk - resp = json.loads(data) - if resp.get("stdout"): - sys.stdout.write(resp["stdout"]) - if resp.get("stderr"): - sys.stderr.write(resp["stderr"]) - sys.exit(resp.get("exit_code", 0)) + exit_code = 1 + while True: + header = recv_exact(sock, 5) + if header is None: + break + frame_type = chr(header[0]) + length = struct.unpack(">I", header[1:5])[0] + payload = recv_exact(sock, length) if length > 0 else b"" + if payload is None: + break + + if frame_type == "d": + sys.stdout.buffer.write(payload) + sys.stdout.buffer.flush() + elif frame_type == "x": + exit_code = int(payload.decode()) + break + + sys.exit(exit_code) if __name__ == "__main__": diff --git a/scripts/x/elevate-adapter-superpowerd b/scripts/x/elevate-adapter-superpowerd index e5c100f..97fc326 100755 --- a/scripts/x/elevate-adapter-superpowerd +++ b/scripts/x/elevate-adapter-superpowerd @@ -1,9 +1,15 @@ #!/usr/bin/env python3 -"""Elevation adapter: bridges the ollie elevate protocol to superpowerd. +"""Elevation adapter: runs commands outside the Landlock sandbox with streaming I/O. Creates a Unix socket at OLLIE_ELEVATE_SOCKET, accepts {cmd, cwd} requests -from x/elevate, and forwards them to superpowerd via its ExecuteCommand -JSON-RPC interface. Run this outside the Landlock sandbox. +from x/elevate, sends them to superpowerd via ExecuteCommandStreaming with +fd-passing, and streams the output back to the client in real-time. + +Protocol (client-facing, streaming): + Client sends: {"cmd": ..., "cwd": ...}\n then shutdown(WR) + Server sends frames: + type 'd' = output data (combined stdout+stderr) + type 'x' = exit code (payload is ASCII integer) Usage: superpowers run-session -- python3 elevate-adapter-superpowerd @@ -17,10 +23,12 @@ Environment: """ import sys sys.path = sys.path[1:] # strip script dir — may be a non-navigable fs (e.g. 9P) +import array import json import os import signal import socket +import struct import threading @@ -39,6 +47,19 @@ def sp_socket_path(): return f"/run/user/{os.getuid()}/superpowerd/superpowerd.sock" +def send_frame(conn, frame_type, payload): + """Send a framed message: """ + data = payload if isinstance(payload, bytes) else payload.encode() + header = frame_type.encode() + struct.pack(">I", len(data)) + conn.sendall(header + data) + + +def send_error(conn, msg, exit_code=1): + """Send an error message and exit frame.""" + send_frame(conn, "d", msg + "\n") + send_frame(conn, "x", str(exit_code)) + + def handle(conn): with conn: data = b"" @@ -50,51 +71,108 @@ def handle(conn): try: req = json.loads(data) except json.JSONDecodeError as e: - resp = {"stdout": "", "stderr": f"elevate-adapter: bad request: {e}", "exit_code": 1} - conn.sendall(json.dumps(resp).encode()) + send_error(conn, f"elevate-adapter: bad request: {e}") return + cmd = req.get("cmd", "") + cwd = req.get("cwd", os.getcwd()) + token = os.environ.get("SUPERPOWERD_SESSION_TOKEN", "") sp_path = sp_socket_path() + # Connect to superpowerd via SEQPACKET sp = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) try: sp.connect(sp_path) except OSError as e: - resp = {"stdout": "", "stderr": f"elevate-adapter: connect {sp_path}: {e}", "exit_code": 1} - conn.sendall(json.dumps(resp).encode()) + send_error(conn, f"elevate-adapter: connect {sp_path}: {e}") return + # Create pipes for stdout+stderr (merged) and stdin (/dev/null) + stdin_r = os.open("/dev/null", os.O_RDONLY) + stdout_r, stdout_w = os.pipe() + # Use stdout_w for both stdout and stderr (merged output) + + # Build JSON-RPC request sp_req = json.dumps({ "jsonrpc": "2.0", - "method": "ExecuteCommand", + "method": "ExecuteCommandStreaming", "id": 1, "params": { "auth": {"session_token": token}, - "command": req["cmd"], - "working_dir": req["cwd"], + "command": cmd, + "working_dir": cwd, "approval_timeout": 7200, }, }).encode() - with sp: - sp.sendall(sp_req) + # Send request + 3 fds (stdin, stdout, stderr) via SCM_RIGHTS + fds = array.array("i", [stdin_r, stdout_w, stdout_w]) + ancdata = [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fds)] + try: + sp.sendmsg([sp_req], ancdata) + except OSError as e: + os.close(stdin_r) + os.close(stdout_r) + os.close(stdout_w) + sp.close() + send_error(conn, f"elevate-adapter: sendmsg: {e}") + return + + # Close write-ends in this process (superpowerd has them now) + os.close(stdin_r) + os.close(stdout_w) + + # Stream output from the pipe to the elevate client + # Do this in a thread so we can also wait for superpowerd's response + stream_done = threading.Event() + client_broken = threading.Event() + + def stream_output(): + try: + while True: + chunk = os.read(stdout_r, 4096) + if not chunk: + break + send_frame(conn, "d", chunk) + except (BrokenPipeError, OSError): + client_broken.set() + finally: + os.close(stdout_r) + stream_done.set() + + t = threading.Thread(target=stream_output, daemon=True) + t.start() + + # Wait for superpowerd's response (comes after command exits) + try: sp_data = sp.recv(1024 * 1024) + except OSError: + sp_data = b"" + sp.close() - sp_resp = json.loads(sp_data) - rpc_err = sp_resp.get("error") - if rpc_err: - msg = rpc_err.get("message", str(rpc_err)) if isinstance(rpc_err, dict) else str(rpc_err) - resp = {"stdout": "", "stderr": f"elevate-adapter: {msg}", "exit_code": 1} - else: - result = sp_resp.get("result", {}) - resp = { - "stdout": result.get("stdout", ""), - "stderr": result.get("stderr", ""), - "exit_code": result.get("exit_code", 0), - } + # Wait for streaming thread to finish + stream_done.wait() - conn.sendall(json.dumps(resp).encode()) + if client_broken.is_set(): + return + + # Parse exit code from superpowerd response + exit_code = 1 + if sp_data: + try: + sp_resp = json.loads(sp_data) + rpc_err = sp_resp.get("error") + if rpc_err: + msg = rpc_err.get("message", str(rpc_err)) if isinstance(rpc_err, dict) else str(rpc_err) + send_frame(conn, "d", f"elevate-adapter: {msg}\n".encode()) + else: + result = sp_resp.get("result", {}) + exit_code = result.get("exit_code", 0) + except (json.JSONDecodeError, ValueError): + pass + + send_frame(conn, "x", str(exit_code)) def main():