#!/usr/bin/env python3 import json import os import signal import socket import struct import traceback from registry import Registry, RegistryError SOCKET_PATH = os.environ.get( "MANAGED_CONTAINERD_SOCKET", "/run/managed-containerd/managed-containerd.sock", ) server_socket = None registry = None pacemaker_resource_id = None pacemaker_peer_uid = None pacemaker_peer_gid = None PROTOCOL_VERSION = 1 def protocol_response(payload): """ Return a daemon response carrying the wire-protocol version. protocol_version is daemon-owned metadata and cannot be overridden by registry or plugin output. """ response = dict(payload) response["protocol_version"] = PROTOCOL_VERSION return response def load_pacemaker_resource_binding( current_registry, environ=None, ): """ Load and validate the Pacemaker resource identity accepted by this daemon instance. RESOURCE_ID identifies the registry entry. PACEMAKER_RESOURCE_ID is a separate control-plane binding and must resolve to that registry. """ if environ is None: environ = os.environ resource_id = environ.get( "PACEMAKER_RESOURCE_ID" ) if not resource_id: raise RegistryError( "PACEMAKER_RESOURCE_ID is not configured" ) if resource_id not in current_registry.resources: raise RegistryError( "PACEMAKER_RESOURCE_ID does not identify " f"a registered resource: {resource_id}" ) return resource_id def load_pacemaker_peer_authorization( environ=None, ): """ Load the peer credentials that authorize Pacemaker resource actions. In the rootless keep-id user namespace, host root is intentionally unmapped and therefore appears as the configured overflow UID/GID. Socket path permissions remain part of this authorization boundary. """ if environ is None: environ = os.environ raw_uid = environ.get("PACEMAKER_PEER_UID") raw_gid = environ.get("PACEMAKER_PEER_GID") if raw_uid is None or raw_gid is None: raise RegistryError( "PACEMAKER_PEER_UID and " "PACEMAKER_PEER_GID must both be configured" ) try: peer_uid = int(raw_uid) peer_gid = int(raw_gid) except ValueError as exc: raise RegistryError( "PACEMAKER_PEER_UID and " "PACEMAKER_PEER_GID must be integers" ) from exc if peer_uid < 0 or peer_gid < 0: raise RegistryError( "PACEMAKER peer credentials must be non-negative" ) return peer_uid, peer_gid def get_peer_credentials(conn): raw = conn.getsockopt( socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i"), ) pid, uid, gid = struct.unpack("3i", raw) return { "pid": pid, "uid": uid, "gid": gid, } def handle_request( req, peer_credentials=None, ): if not isinstance(req, dict): return protocol_response({ "ok": False, "error_code": "invalid_request", "error": "request must be a JSON object", }) requested_version = req.get("protocol_version") # Backward compatibility for the pre-versioned proxy: # absence is accepted during the rolling upgrade, but an explicitly # incompatible version is rejected. if ( requested_version is not None and requested_version != PROTOCOL_VERSION ): return protocol_response({ "ok": False, "error_code": "unsupported_protocol_version", "error": ( "unsupported protocol version: " f"{requested_version}; " f"supported={PROTOCOL_VERSION}" ), }) action = req.get("action") if action == "ping": return protocol_response({ "ok": True, "daemon": "managed-containerd", "pacemaker_resource": ( pacemaker_resource_id ), }) if action == "registry.list": return protocol_response({ "ok": True, "pacemaker_resource": ( pacemaker_resource_id ), "resources": registry.list_resources(), }) resource = req.get("resource") if not resource: return protocol_response({ "ok": False, "error_code": "invalid_request", "error": "resource is required", }) if resource != pacemaker_resource_id: return protocol_response({ "ok": False, "error_code": "resource_binding_mismatch", "error": ( "requested resource does not match " "the configured Pacemaker binding" ), "resource": resource, "action": action, }) if peer_credentials is None: return protocol_response({ "ok": False, "error_code": "peer_credentials_unavailable", "error": ( "peer credentials are required for " "Pacemaker resource actions" ), "resource": resource, "action": action, }) if ( peer_credentials.get("uid") != pacemaker_peer_uid or peer_credentials.get("gid") != pacemaker_peer_gid ): return protocol_response({ "ok": False, "error_code": "peer_credential_mismatch", "error": ( "peer credentials are not authorized " "for Pacemaker resource actions" ), "resource": resource, "action": action, }) return protocol_response( registry.execute( resource, action, ) ) def cleanup_socket(): try: os.unlink(SOCKET_PATH) except FileNotFoundError: pass def shutdown(signum, frame): global server_socket print( f"managed-containerd received signal {signum}; shutting down", flush=True, ) if server_socket is not None: try: server_socket.close() except OSError: pass cleanup_socket() raise SystemExit(0) def serve(): global server_socket global registry global pacemaker_resource_id global pacemaker_peer_uid global pacemaker_peer_gid registry = Registry() pacemaker_resource_id = ( load_pacemaker_resource_binding(registry) ) ( pacemaker_peer_uid, pacemaker_peer_gid, ) = load_pacemaker_peer_authorization() socket_directory = os.path.dirname( SOCKET_PATH ) os.makedirs( socket_directory, exist_ok=True, ) os.chmod(socket_directory, 0o700) cleanup_socket() signal.signal(signal.SIGTERM, shutdown) signal.signal(signal.SIGINT, shutdown) server_socket = socket.socket( socket.AF_UNIX, socket.SOCK_STREAM, ) server_socket.bind(SOCKET_PATH) os.chmod(SOCKET_PATH, 0o600) server_socket.listen(16) print( f"managed-containerd listening on {SOCKET_PATH}", flush=True, ) print( "registered resources: " + ", ".join(registry.resources), flush=True, ) print( "Pacemaker resource binding: " + pacemaker_resource_id, flush=True, ) print( "Pacemaker peer authorization: " f"uid={pacemaker_peer_uid} " f"gid={pacemaker_peer_gid}", flush=True, ) try: while True: conn, _ = server_socket.accept() with conn: try: peer_credentials = ( get_peer_credentials(conn) ) raw = b"" while b"\n" not in raw: chunk = conn.recv(65536) if not chunk: break raw += chunk req = json.loads( raw.decode("utf-8").strip() ) response = handle_request( req, peer_credentials, ) except ( RegistryError, ValueError, KeyError, ) as exc: response = protocol_response({ "ok": False, "error_code": "request_error", "error": str(exc), }) except Exception as exc: traceback.print_exc() response = protocol_response({ "ok": False, "error_code": "internal_error", "error": str(exc), }) conn.sendall( json.dumps( response, default=str, ).encode("utf-8") + b"\n" ) finally: cleanup_socket() if __name__ == "__main__": serve()