1"""Bounded subprocess execution for the fixed Sigil and systemd interfaces."""23from __future__ import annotations45import asyncio6import json7import os8import stat9from dataclasses import dataclass10from pathlib import Path11from typing import Any, Sequence1213from .errors import ControllerError141516MAX_OUTPUT_BYTES = 128 * 102417MAX_REQUEST_BYTES = 16 * 102418192021class CommandResult:22 returncode: int23 stdout: bytes24 stderr: bytes252627def _strict_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:28 result: dict[str, Any] = {}29 for key, value in pairs:30 if key in result:31 raise ValueError("duplicate JSON key")32 result[key] = value33 return result343536def strict_json(raw: bytes, schema_version: int) -> dict[str, Any]:37 if not raw or len(raw) > MAX_OUTPUT_BYTES:38 raise ControllerError("invalid_response")39 try:40 value = json.loads(raw, object_pairs_hook=_strict_object)41 except (UnicodeDecodeError, ValueError, json.JSONDecodeError) as error:42 raise ControllerError("invalid_response") from error43 if not isinstance(value, dict) or value.get("schema_version") != schema_version:44 raise ControllerError("incompatible_sigil")45 return value464748class Runner:49 """Runs only absolute, caller-assembled argv without invoking a shell."""5051 def __init__(self, user_home: str | Path, *, uid: int | None = None):52 self.home = Path(user_home)53 self.uid = os.geteuid() if uid is None else uid54 self.sigil = self.home / ".local/libexec/sigil-spark/current/sigil"55 self.config = self.home / ".config/sigil-spark/host.toml"56 self.unit = "sigil-host.service"57 self.systemctl = Path("/usr/bin/systemctl")5859 def environment(self) -> dict[str, str]:60 if self.uid == 0:61 raise ControllerError("unsafe_user")62 runtime = Path(f"/run/user/{self.uid}")63 bus = runtime / "bus"64 try:65 runtime_stat = runtime.stat()66 bus_stat = bus.stat()67 except OSError as error:68 raise ControllerError("user_session_unavailable") from error69 if (70 not stat.S_ISDIR(runtime_stat.st_mode)71 or runtime_stat.st_uid != self.uid72 or runtime_stat.st_mode & 0o07773 or not stat.S_ISSOCK(bus_stat.st_mode)74 or bus_stat.st_uid != self.uid75 ):76 raise ControllerError("unsafe_user_session")77 return {78 "HOME": str(self.home),79 "XDG_RUNTIME_DIR": str(runtime),80 "DBUS_SESSION_BUS_ADDRESS": f"unix:path={bus}",81 "LANG": "C.UTF-8",82 }8384 async def run(85 self,86 argv: Sequence[str | Path],87 *,88 stdin: bytes | None = None,89 timeout: float = 10.0,90 check: bool = True,91 ) -> CommandResult:92 if not argv or not Path(argv[0]).is_absolute():93 raise ControllerError("unsafe_command")94 if stdin is not None and len(stdin) > MAX_REQUEST_BYTES:95 raise ControllerError("invalid_request")96 process = await asyncio.create_subprocess_exec(97 *(str(value) for value in argv),98 stdin=asyncio.subprocess.PIPE if stdin is not None else asyncio.subprocess.DEVNULL,99 stdout=asyncio.subprocess.PIPE,100 stderr=asyncio.subprocess.PIPE,101 env=self.environment(),102 )103104 async def stop_child() -> None:105 if process.returncode is not None:106 return107 process.terminate()108 try:109 await asyncio.wait_for(process.wait(), timeout=1.0)110 except TimeoutError:111 process.kill()112 await process.wait()113114 async def read_bounded(stream: asyncio.StreamReader | None) -> bytes:115 if stream is None:116 return b""117 chunks: list[bytes] = []118 total = 0119 while True:120 chunk = await stream.read(min(4096, MAX_OUTPUT_BYTES + 1 - total))121 if not chunk:122 return b"".join(chunks)123 chunks.append(chunk)124 total += len(chunk)125 if total > MAX_OUTPUT_BYTES:126 raise ControllerError("output_limit")127128 try:129 if stdin is not None and process.stdin is not None:130 process.stdin.write(stdin)131 await process.stdin.drain()132 process.stdin.close()133 stdout, stderr, _ = await asyncio.wait_for(134 asyncio.gather(135 read_bounded(process.stdout),136 read_bounded(process.stderr),137 process.wait(),138 ),139 timeout=timeout,140 )141 except (TimeoutError, ControllerError) as error:142 await stop_child()143 if isinstance(error, ControllerError):144 raise145 raise ControllerError("command_timeout") from error146 except BaseException:147 await stop_child()148 raise149150 result = CommandResult(process.returncode or 0, stdout, stderr)151 if check and result.returncode != 0:152 raise ControllerError("command_failed")153 return result154155 async def sigil_json(156 self,157 arguments: Sequence[str],158 *,159 stdin_value: dict[str, Any] | None = None,160 schema_version: int = 1,161 timeout: float = 10.0,162 ) -> dict[str, Any]:163 if not self.sigil.is_file() or not os.access(self.sigil, os.X_OK):164 raise ControllerError("sigil_not_installed")165 stdin = None166 if stdin_value is not None:167 try:168 stdin = json.dumps(169 stdin_value, separators=(",", ":"), ensure_ascii=True170 ).encode("ascii")171 except (TypeError, ValueError) as error:172 raise ControllerError("invalid_request") from error173 result = await self.run(174 [self.sigil, *arguments], stdin=stdin, timeout=timeout175 )176 return strict_json(result.stdout, schema_version)177178 async def systemctl_command(179 self, action: str, *, check: bool = True, timeout: float = 15.0180 ) -> CommandResult:181 if action not in {"start", "stop", "restart", "is-active", "show"}:182 raise ControllerError("unsafe_command")183 arguments = [self.systemctl, "--user", action, self.unit]184 if action == "show":185 arguments.extend(186 [187 "--property=LoadState",188 "--property=ActiveState",189 "--property=SubState",190 "--property=UnitFileState",191 "--no-pager",192 ]193 )194 return await self.run(arguments, timeout=timeout, check=check)