diff --git a/src/agent_comms/active_route.py b/src/agent_comms/active_route.py index 3fec97cfbc..25a823214a 100644 --- a/src/agent_comms/active_route.py +++ b/src/agent_comms/active_route.py @@ -18,7 +18,7 @@ from pathlib import Path from typing import TYPE_CHECKING, Annotated, ClassVar -from .child_process import Platform +from .child_process import Platform, ProcessIdentity from .coordination_errors import CoordinationError from .declared_family import DeclaredFamily from .errors import RelationViolationError @@ -295,11 +295,6 @@ def detail(self) -> str: return "process exited" -class EmptyCommandLockHolder(IdentifiedLockHolder): - def detail(self) -> str: - return "empty command" - - class UnidentifiedLockHolder(RouteLockHolder): def describe(self) -> str: return "kernel did not identify the owner" @@ -324,11 +319,11 @@ def route_lock_holders(directory: int) -> tuple[RouteLockHolder, ...]: holders.append(UnidentifiedLockHolder()) continue try: - command = Path(f'/proc/{pid}/cmdline').read_bytes().replace(b'\0', b' ').decode('utf-8', 'replace').strip() - except FileNotFoundError: + command = " ".join(ProcessIdentity.capture(pid).arguments()) + except ProcessLookupError: holders.append(ExitedLockHolder(pid)) else: - holders.append(RunningLockHolder(pid, command) if command else EmptyCommandLockHolder(pid)) + holders.append(RunningLockHolder(pid, command)) return tuple(holders) diff --git a/src/agent_comms/child_process.py b/src/agent_comms/child_process.py index 303341f65a..c6b2529cf9 100644 --- a/src/agent_comms/child_process.py +++ b/src/agent_comms/child_process.py @@ -247,6 +247,12 @@ def __eq__(self, other: object) -> bool: def alive(self) -> bool: return Platform.current().matches(self) + def arguments(self) -> tuple[str, ...]: + return Platform.current().arguments(self) + + def environment(self) -> dict[str, str]: + return Platform.current().environment(self) + def require_alive(self) -> ProcessIdentity: """Validate this recorded PID and start time against the live process.""" return Platform.current().require(self) @@ -410,6 +416,12 @@ def force_group(self, identity: ProcessIdentity) -> None: @abstractmethod def identity(self, pid: int) -> ProcessIdentity: ... + @abstractmethod + def arguments(self, process: ProcessIdentity) -> tuple[str, ...]: ... + + @abstractmethod + def environment(self, process: ProcessIdentity) -> dict[str, str]: ... + def parent_identity(self, pid: int) -> ProcessIdentity: raise NotImplementedError('This platform does not expose exact parent identities') @@ -550,6 +562,23 @@ def _stat(pid: int) -> LinuxProcessRecord: def identity(self, pid: int) -> ProcessIdentity: return self._stat(pid).identity + def arguments(self, process: ProcessIdentity) -> tuple[str, ...]: + try: + raw = Path(f"/proc/{process.pid}/cmdline").read_bytes() + except FileNotFoundError: + raise ProcessLookupError(process.pid) from None + if not raw: + raise ProcessLookupError(process.pid) + return tuple(os.fsdecode(word) for word in raw.split(b"\0")[:-1]) + + def environment(self, process: ProcessIdentity) -> dict[str, str]: + try: + raw = Path(f"/proc/{process.pid}/environ").read_bytes() + except FileNotFoundError: + raise ProcessLookupError(process.pid) from None + values = (item.split(b"=", 1) for item in raw.split(b"\0") if b"=" in item) + return {os.fsdecode(key): os.fsdecode(value) for key, value in values} + def parent_identity(self, pid: int) -> ProcessIdentity: return self.identity(self._stat(pid).parent) @@ -683,6 +712,12 @@ def identity(self, pid: int) -> ProcessIdentity: pid, info.start_seconds * 1_000_000 + info.start_microseconds, info.pgid, ) + def arguments(self, process: ProcessIdentity) -> tuple[str, ...]: + raise NotImplementedError(f"{type(self).__name__} does not read process arguments") + + def environment(self, process: ProcessIdentity) -> dict[str, str]: + raise NotImplementedError(f"{type(self).__name__} does not read process environment") + def members_of_group(self, pgid: int) -> tuple[ProcessIdentity, ...]: # proc_listpids returns bytes. Grow until one query fits; never mistake # a full buffer for a complete process-group observation. @@ -875,6 +910,12 @@ def identity(self, pid: int) -> ProcessIdentity: with self.process_handle(pid) as handle: return self._identity(pid, handle) + def arguments(self, process: ProcessIdentity) -> tuple[str, ...]: + raise NotImplementedError(f"{type(self).__name__} does not read process arguments") + + def environment(self, process: ProcessIdentity) -> dict[str, str]: + raise NotImplementedError(f"{type(self).__name__} does not read process environment") + def wait_any_exited( self, identities: tuple[ProcessIdentity, ...], timeout: float, ) -> None: diff --git a/src/agent_comms/host_services/__init__.py b/src/agent_comms/host_services/__init__.py index 5e825675ac..bc71a53ccf 100644 --- a/src/agent_comms/host_services/__init__.py +++ b/src/agent_comms/host_services/__init__.py @@ -19,6 +19,7 @@ from aiohttp.web import TCPSite +from ..child_process import ProcessIdentity from ..declared_family import DeclaredFamily from ..field_codec import PathText from ..launch_runtime import LaunchRuntime @@ -196,15 +197,14 @@ def running(cls) -> tuple[Self, ...]: return () if pid == 0 else (cls.from_process(pid),) @classmethod - def process_arguments(cls, pid: int) -> tuple[str, ...]: - arguments = tuple(os.fsdecode(word) for word in - Path(f"/proc/{pid}/cmdline").read_bytes().split(b"\0")[:-1]) - if not arguments or not arguments[0]: - raise ServiceNotServingError(f"{cls.unit} has no running command") - return arguments + def running_arguments(cls, pid: int) -> tuple[str, ...]: + try: + return ProcessIdentity.capture(pid).arguments() + except ProcessLookupError: + raise ServiceNotServingError(f"{cls.unit} has no running command") from None def runtime(self) -> Path: - return Path(self.process_arguments(self.pid())[0]) + return Path(self.running_arguments(self.pid())[0]) def restart(self) -> None: subprocess.run(("systemctl", "--user", "restart", self.unit), check=True) diff --git a/src/agent_comms/host_services/connector.py b/src/agent_comms/host_services/connector.py index ee3cc449ac..93fae453d3 100644 --- a/src/agent_comms/host_services/connector.py +++ b/src/agent_comms/host_services/connector.py @@ -81,7 +81,7 @@ class SystemdConnectorService(ConnectorService, SystemdUserService): @classmethod def from_process(cls, pid: int) -> Self: # Our installed console script's argv is interpreter, script, verb, options. - endpoint = ServeVerb.parse(cls.process_arguments(pid)[2:]) + endpoint = ServeVerb.parse(cls.running_arguments(pid)[2:]) return cls(launch=LaunchRoot.from_path(endpoint.root), port=endpoint.port) def exec_arguments(self, issuer: str) -> tuple[str, ...]: diff --git a/src/agent_comms/owner_launch.py b/src/agent_comms/owner_launch.py index 9523ff6e84..0887cd12f3 100644 --- a/src/agent_comms/owner_launch.py +++ b/src/agent_comms/owner_launch.py @@ -2,9 +2,7 @@ from __future__ import annotations -import os from dataclasses import dataclass, field -from pathlib import Path from typing import TYPE_CHECKING from .child_process import Platform, ProcessIdentity @@ -37,13 +35,10 @@ def capture( process = selection.process platform = Platform.current() platform.require(process) - command = Path(f"/proc/{process.pid}/cmdline").read_bytes().split(b"\0", 1)[0] - observed_interpreter = os.fsdecode(command) + observed_interpreter = process.arguments()[0] if observed_interpreter != selection.interpreter: raise RelationViolationError("Selected source owner interpreter changed") - raw = Path(f"/proc/{process.pid}/environ").read_bytes() - values = dict(item.split(b"=", 1) for item in raw.split(b"\0") if b"=" in item) - environment = {os.fsdecode(key): os.fsdecode(value) for key, value in values.items()} + environment = process.environment() platform.require(process) route = launch_runtime.ThreadLaunch.current(environment).name if not environment.get(NativeArguments.binary_key): diff --git a/src/agent_comms/owner_lifecycle.py b/src/agent_comms/owner_lifecycle.py index fdac286540..80d8f7e165 100644 --- a/src/agent_comms/owner_lifecycle.py +++ b/src/agent_comms/owner_lifecycle.py @@ -103,8 +103,7 @@ def capture(cls, snapshot: RegistrySnapshot, name: str): process = owner.require_process() platform = Platform.current() platform.require(process) - command = Path(f"/proc/{process.pid}/cmdline").read_bytes().split(b"\0", 1)[0] - interpreter = os.fsdecode(command) + interpreter = process.arguments()[0] if not interpreter: raise RelationViolationError("Selected source owner interpreter is unavailable") platform.require(process)