136 lines
4.2 KiB
Python
Executable File
136 lines
4.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Elevation adapter: bridges the ollie elevate protocol to superpowerd.
|
|
|
|
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.
|
|
|
|
Usage:
|
|
superpowers run-session -- python3 elevate-adapter-superpowerd
|
|
|
|
Environment:
|
|
OLLIE_ELEVATE_SOCKET path for the listening socket
|
|
(default: $XDG_RUNTIME_DIR/ollie/elevate.sock)
|
|
SUPERPOWERD_SESSION_TOKEN set automatically by superpowers run-session
|
|
SUPERPOWERD_SOCKET override superpowerd socket path
|
|
SUPERPOWERD_SOCKET_DIR directory containing superpowerd.sock
|
|
"""
|
|
import json
|
|
import os
|
|
import signal
|
|
import socket
|
|
import sys
|
|
import threading
|
|
|
|
|
|
def default_socket_path():
|
|
xdg = os.environ.get("XDG_RUNTIME_DIR") or f"/run/user/{os.getuid()}"
|
|
return os.path.join(xdg, "ollie", "elevate.sock")
|
|
|
|
|
|
def sp_socket_path():
|
|
if p := os.environ.get("SUPERPOWERD_SOCKET"):
|
|
return p
|
|
if d := os.environ.get("SUPERPOWERD_SOCKET_DIR"):
|
|
return d + "/superpowerd.sock"
|
|
if x := os.environ.get("XDG_RUNTIME_DIR"):
|
|
return x + "/superpowerd/superpowerd.sock"
|
|
return f"/run/user/{os.getuid()}/superpowerd/superpowerd.sock"
|
|
|
|
|
|
def handle(conn):
|
|
with conn:
|
|
data = b""
|
|
while chunk := conn.recv(65536):
|
|
data += chunk
|
|
if b"\n" in data:
|
|
break
|
|
|
|
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())
|
|
return
|
|
|
|
token = os.environ.get("SUPERPOWERD_SESSION_TOKEN", "")
|
|
sp_path = sp_socket_path()
|
|
|
|
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())
|
|
return
|
|
|
|
sp_req = json.dumps({
|
|
"jsonrpc": "2.0",
|
|
"method": "ExecuteCommand",
|
|
"id": 1,
|
|
"params": {
|
|
"auth": {"session_token": token},
|
|
"command": req["cmd"],
|
|
"working_dir": req["cwd"],
|
|
"approval_timeout": 7200,
|
|
},
|
|
}).encode()
|
|
|
|
with sp:
|
|
sp.sendall(sp_req)
|
|
sp_data = sp.recv(1024 * 1024)
|
|
|
|
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),
|
|
}
|
|
|
|
conn.sendall(json.dumps(resp).encode())
|
|
|
|
|
|
def main():
|
|
if "SUPERPOWERD_SESSION_TOKEN" not in os.environ:
|
|
os.execvp("superpowers", ["superpowers", "run-session", "--",
|
|
sys.executable, __file__] + sys.argv[1:])
|
|
|
|
sock_path = os.environ.get("OLLIE_ELEVATE_SOCKET") or default_socket_path()
|
|
sock_path = os.path.expanduser(sock_path)
|
|
os.makedirs(os.path.dirname(sock_path), exist_ok=True)
|
|
|
|
if os.path.exists(sock_path):
|
|
os.unlink(sock_path)
|
|
|
|
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
srv.bind(sock_path)
|
|
srv.listen()
|
|
|
|
def shutdown(sig, frame):
|
|
srv.close()
|
|
if os.path.exists(sock_path):
|
|
os.unlink(sock_path)
|
|
sys.exit(0)
|
|
|
|
signal.signal(signal.SIGTERM, shutdown)
|
|
signal.signal(signal.SIGINT, shutdown)
|
|
|
|
print(f"elevate-adapter-superpowerd: listening on {sock_path}", file=sys.stderr)
|
|
|
|
while True:
|
|
try:
|
|
conn, _ = srv.accept()
|
|
except OSError:
|
|
break
|
|
threading.Thread(target=handle, args=(conn,), daemon=True).start()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|