"""C01 v7 optional serial channel. No device locator is supplied by a manager. The separately installed radio owner has already opened the externally appointed TTY. This connector does not open hardware, change permissions, or start services. """ import os from pathlib import Path import socket import struct import c01_runtime as wire PATH = Path('/run/archipelago-radio/control.sock') def application_fds(prepared, channel): """Retain the endpoint through Children.spawn's PRE-callback FD sweep.""" return (*prepared, *((channel.fileno(),) if channel is not None else ())) def connect(): if not os.path.lexists(PATH): return None return wire.root_connect(PATH) def bind(channel, pid, uid, gid): wire.send(channel, {'bind': [pid, uid, gid]}) reply, _ = wire.receive(channel) wire.require(reply == {'bound': [pid, uid, gid]}, 'radio caller binding refused') return struct.unpack('3i', channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))[0]