lkn
/
ollie
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

elevate: streaming protocol for real-time output 

Replace the batch JSON request/response protocol with a streaming
frame protocol. The adapter now uses superpowerd ExecuteCommandStreaming
with SCM_RIGHTS fd-passing to stream stdout/stderr in real-time.

Protocol: client sends {cmd,cwd}+newline, server streams frames
(type byte + 4-byte length + payload). This enables elevated+detach
to capture live output in the ring buffer.

Also updates core submodule with elevated+detach support.
This commit is contained in:
Ollie Agent 2026-07-18 00:16:32 +02:00
parent 500388cd26
commit 83ca936c8c
3 changed files with 143 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
core

@ -1 +1 @@
Subproject commit c1124b588d01be470821efa7887f32517c3b0f96
Subproject commit 42361e6505e607a9ee406226a423e5c1fac82d55

50
scripts/x/elevate
View File

@ -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:1><length:4 big-endian><payload>
type 'd' = output data (write to stdout immediately)
type 'x' = exit code (payload is ASCII integer)
Usage: elevate -- <cmd>
"""
@ -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__":

128
scripts/x/elevate-adapter-superpowerd
View File

@ -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:1><length:4 big-endian><payload>
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: <type:1><length:4 big-endian><payload>"""
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():