# # Ouroboros - Copyright (C) 2016 - 2026 # # Python API for Ouroboros - CLI equivalents # # Dimitri Staessens # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public License # version 2.1 as published by the Free Software Foundation. # # This library is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU # Lesser General Public License for more details. # # You should have received a copy of the GNU Lesser General Public # License along with this library; if not, write to the Free Software # Foundation, Inc., http://www.fsf.org/about/contact/. # """ Higher-level wrappers that mirror CLI tool behaviour. The ``irm`` CLI tools perform extra steps that the raw C library API does not, such as resolving program names via ``realpath``, looking up IPCP pids, and the ``autobind`` flag for bootstrapping/enrolling. This module exposes those same patterns as a Python API so that callers do not need to re-implement them. Each wrapper corresponds to a specific ``irm`` sub-command: ========================= ==================================== Python CLI equivalent ========================= ==================================== ``create_ipcp`` ``irm ipcp create`` ``destroy_ipcp`` ``irm ipcp destroy`` ``bootstrap_ipcp`` ``irm ipcp bootstrap [autobind]`` ``enroll_ipcp`` ``irm ipcp enroll [autobind]`` ``connect_ipcp`` ``irm ipcp connect`` ``disconnect_ipcp`` ``irm ipcp disconnect`` ``list_ipcps`` ``irm ipcp list`` ``bind_program`` ``irm bind program`` ``bind_process`` ``irm bind process`` ``bind_ipcp`` ``irm bind ipcp`` ``unbind_program`` ``irm unbind program`` ``unbind_process`` ``irm unbind process`` ``unbind_ipcp`` ``irm unbind ipcp`` ``create_name`` ``irm name create`` ``destroy_name`` ``irm name destroy`` ``reg_name`` ``irm name register`` ``unreg_name`` ``irm name unregister`` ``list_names`` ``irm name list`` ``autoboot`` ``irm ipcp bootstrap autobind`` (with implicit ``create``) ========================= ==================================== Usage:: from ouroboros.cli import create_ipcp, bootstrap_ipcp, enroll_ipcp from ouroboros.cli import bind_program, autoboot """ import shutil from typing import List, Optional from ouroboros.irm import ( DT_COMP, MGMT_COMP, IpcpType, IpcpConfig, IpcpInfo, NameInfo, BindError, IrmError, UnicastConfig, DtConfig, RoutingConfig, LinkStateConfig, LinkStatePolicy, EthConfig, Udp4Config, Udp6Config, bind_program as _irm_bind_program, bind_process as _irm_bind_process, bootstrap_ipcp as _irm_bootstrap_ipcp, connect_ipcp as _irm_connect_ipcp, create_ipcp, create_name as _irm_create_name, destroy_ipcp as _irm_destroy_ipcp, destroy_name, disconnect_ipcp as _irm_disconnect_ipcp, enroll_ipcp as _irm_enroll_ipcp, list_ipcps as _irm_list_ipcps, list_names as _irm_list_names, reg_name as _irm_reg_name, unbind_process as _irm_unbind_process, unbind_program as _irm_unbind_program, unreg_name as _irm_unreg_name, ) from ouroboros.qos import QoSSpec def _pid_of(ipcp_name: str) -> int: """Look up the pid of a running IPCP by its name.""" for info in _irm_list_ipcps(): if info.name == ipcp_name: return info.pid raise ValueError(f"No IPCP named {ipcp_name!r}") def destroy_ipcp(name: str) -> None: """ Destroy an IPCP by name. Mirrors ``irm ipcp destroy name ``. Resolves the IPCP name to a pid, then destroys the IPCP. :param name: Name of the IPCP to destroy. :raises ValueError: If no IPCP with *name* exists. """ _irm_destroy_ipcp(_pid_of(name)) def list_ipcps(name: Optional[str] = None, layer: Optional[str] = None, ipcp_type: Optional[IpcpType] = None) -> List[IpcpInfo]: """ List running IPCPs, optionally filtered. Mirrors ``irm ipcp list [name ] [type ] [layer ]``. :param name: Filter by IPCP name (exact match). :param layer: Filter by layer name (exact match). :param ipcp_type: Filter by IPCP type. :return: List of matching :class:`IpcpInfo` objects. """ result = _irm_list_ipcps() if name is not None: result = [i for i in result if i.name == name] if layer is not None: result = [i for i in result if i.layer == layer] if ipcp_type is not None: result = [i for i in result if i.type == ipcp_type] return result def reg_name(name: str, ipcp: Optional[str] = None, ipcps: Optional[List[str]] = None, layer: Optional[str] = None, layers: Optional[List[str]] = None) -> None: """ Register a name with IPCP(s), creating it first if needed. Mirrors ``irm name register ipcp [ipcp ...] layer [layer ...]``. The C CLI tool resolves IPCP names and layer names to pids, checks whether the name already exists and calls ``irm_create_name`` before ``irm_reg_name`` for each IPCP. The function accepts flexible input: - A single *ipcp* name or list of *ipcps* names. - A single *layer* name or list of *layers* names (registers with every IPCP in each of those layers). - Any combination of the above. :param name: The name to register. :param ipcp: Single IPCP name to register with. :param ipcps: List of IPCP names to register with. :param layer: Single layer name to register with. :param layers: List of layer names to register with. """ existing = {n.name for n in _irm_list_names()} if name not in existing: _irm_create_name(NameInfo(name=name)) pids = set() # Collect IPCP names into a single list ipcp_names = [] if ipcp is not None: ipcp_names.append(ipcp) if ipcps is not None: ipcp_names.extend(ipcps) # Collect layer names into a single list layer_names = [] if layer is not None: layer_names.append(layer) if layers is not None: layer_names.extend(layers) if ipcp_names or layer_names: all_ipcps = _irm_list_ipcps() for ipcp_name in ipcp_names: for i in all_ipcps: if i.name == ipcp_name: pids.add(i.pid) break for lyr in layer_names: for i in all_ipcps: if i.layer == lyr: pids.add(i.pid) for p in pids: _irm_reg_name(name, p) def bind_program(prog: str, name: str, opts: int = 0, argv: Optional[List[str]] = None) -> None: """ Bind a program to a name, resolving bare names to full paths. Mirrors ``irm bind program name ``. The ``irm bind program`` CLI tool calls ``realpath()`` on *prog* before passing it to the library. The raw C function ``irm_bind_program`` contains a ``check_prog_path`` helper that corrupts the ``PATH`` environment variable (writes NUL over ``:`` separators) when given a bare program name. Only the first such call would succeed in a long-running process. This wrapper resolves *prog* via ``shutil.which()`` before calling the library, avoiding the bug entirely. :param prog: Program name or path. Bare names (without ``/``) are resolved on ``PATH`` via ``shutil.which()``. :param name: Name to bind to. :param opts: Bind options (e.g. ``BIND_AUTO``). :param argv: Arguments to pass when the program is auto-started. :raises BindError: If the program cannot be found or the bind call fails. """ if '/' not in prog: resolved = shutil.which(prog) if resolved is None: raise BindError(f"Program {prog!r} not found on PATH") prog = resolved _irm_bind_program(prog, name, opts=opts, argv=argv) def unbind_program(prog: str, name: str) -> None: """ Unbind a program from a name. Mirrors ``irm unbind program name ``. :param prog: Path to the program. :param name: Name to unbind from. """ _irm_unbind_program(prog, name) def bind_process(pid: int, name: str) -> None: """ Bind a running process to a name. Mirrors ``irm bind process name ``. :param pid: PID of the process. :param name: Name to bind to. """ _irm_bind_process(pid, name) def unbind_process(pid: int, name: str) -> None: """ Unbind a process from a name. Mirrors ``irm unbind process name ``. :param pid: PID of the process. :param name: Name to unbind from. """ _irm_unbind_process(pid, name) def bind_ipcp(ipcp: str, name: str) -> None: """ Bind an IPCP to a name. Mirrors ``irm bind ipcp name ``. Resolves the IPCP name to a pid, then calls ``bind_process``. :param ipcp: IPCP instance name. :param name: Name to bind to. :raises ValueError: If no IPCP with *ipcp* exists. """ _irm_bind_process(_pid_of(ipcp), name) def unbind_ipcp(ipcp: str, name: str) -> None: """ Unbind an IPCP from a name. Mirrors ``irm unbind ipcp name ``. Resolves the IPCP name to a pid, then calls ``unbind_process``. :param ipcp: IPCP instance name. :param name: Name to unbind from. :raises ValueError: If no IPCP with *ipcp* exists. """ _irm_unbind_process(_pid_of(ipcp), name) def create_name(name: str, pol_lb: Optional[int] = None, info: Optional[NameInfo] = None) -> None: """ Create a registered name. Mirrors ``irm name create [lb ]``. :param name: The name to create. :param pol_lb: Load-balance policy (optional). :param info: Full :class:`NameInfo` (overrides *name*/*pol_lb* if given). """ if info is not None: _irm_create_name(info) else: ni = NameInfo(name=name) if pol_lb is not None: ni.pol_lb = pol_lb _irm_create_name(ni) def list_names(name: Optional[str] = None) -> List[NameInfo]: """ List all registered names, optionally filtered. Mirrors ``irm name list []``. :param name: Filter by name (exact match). :return: List of :class:`NameInfo` objects. """ result = _irm_list_names() if name is not None: result = [n for n in result if n.name == name] return result def unreg_name(name: str, ipcp: Optional[str] = None, ipcps: Optional[List[str]] = None, layer: Optional[str] = None, layers: Optional[List[str]] = None) -> None: """ Unregister a name from IPCP(s). Mirrors ``irm name unregister ipcp [ipcp ...] layer [layer ...]``. Accepts the same flexible input as :func:`reg_name`. :param name: The name to unregister. :param ipcp: Single IPCP name to unregister from. :param ipcps: List of IPCP names to unregister from. :param layer: Single layer name to unregister from. :param layers: List of layer names to unregister from. """ pids = set() ipcp_names = [] if ipcp is not None: ipcp_names.append(ipcp) if ipcps is not None: ipcp_names.extend(ipcps) layer_names = [] if layer is not None: layer_names.append(layer) if layers is not None: layer_names.extend(layers) if ipcp_names or layer_names: all_ipcps = _irm_list_ipcps() for ipcp_name in ipcp_names: for i in all_ipcps: if i.name == ipcp_name: pids.add(i.pid) break for lyr in layer_names: for i in all_ipcps: if i.layer == lyr: pids.add(i.pid) for p in pids: _irm_unreg_name(name, p) def bootstrap_ipcp(name: str, conf: IpcpConfig, autobind: bool = False) -> None: """ Bootstrap an IPCP, optionally binding it to its name and layer. Mirrors ``irm ipcp bootstrap name layer [autobind]``. When *autobind* is ``True`` and the IPCP type is ``UNICAST`` or ``BROADCAST``, the sequence is:: bind_process(pid, ipcp_name) # accept flows for ipcp name bind_process(pid, layer_name) # accept flows for layer name bootstrap_ipcp(pid, conf) # bootstrap into the layer This matches the C ``irm ipcp bootstrap`` tool exactly. If bootstrap fails after autobind, the bindings are rolled back. :param name: Name of the IPCP. :param conf: IPCP configuration (includes layer name & type). :param autobind: Bind the IPCP process to its name and layer. """ pid = _pid_of(name) layer_name = conf.layer_name if autobind and conf.ipcp_type in (IpcpType.UNICAST, IpcpType.BROADCAST): _irm_bind_process(pid, name) _irm_bind_process(pid, layer_name) try: _irm_bootstrap_ipcp(pid, conf) except Exception: if autobind and conf.ipcp_type in (IpcpType.UNICAST, IpcpType.BROADCAST): _irm_unbind_process(pid, name) _irm_unbind_process(pid, layer_name) raise def enroll_ipcp(name: str, dst: str, autobind: bool = False) -> None: """ Enroll an IPCP, optionally binding it to its name and layer. Mirrors ``irm ipcp enroll name layer [autobind]``. When *autobind* is ``True``, the sequence is:: enroll_ipcp(pid, dst) bind_process(pid, ipcp_name) bind_process(pid, layer_name) # layer learned from enrollment This matches the C ``irm ipcp enroll`` tool exactly. :param name: Name of the IPCP. :param dst: Destination name or layer to enroll with. :param autobind: Bind the IPCP process to its name and layer after successful enrollment. """ pid = _pid_of(name) _irm_enroll_ipcp(pid, dst) if autobind: # Look up enrolled layer from the IPCP list for info in _irm_list_ipcps(): if info.pid == pid: _irm_bind_process(pid, info.name) _irm_bind_process(pid, info.layer) break def connect_ipcp(name: str, dst: str, comp: str = "*", qos: Optional[QoSSpec] = None) -> None: """ Connect IPCP components to a destination. Mirrors ``irm ipcp connect name dst [component ] [qos ]``. When *comp* is ``"*"`` (default), both ``dt`` and ``mgmt`` components are connected, matching the CLI default. :param name: Name of the IPCP. :param dst: Destination IPCP name. :param comp: Component to connect: ``"dt"``, ``"mgmt"``, or ``"*"`` for both (default). :param qos: QoS specification for the dt component. """ pid = _pid_of(name) if comp in ("*", "mgmt"): _irm_connect_ipcp(pid, MGMT_COMP, dst) if comp in ("*", "dt"): _irm_connect_ipcp(pid, DT_COMP, dst, qos=qos) def disconnect_ipcp(name: str, dst: str, comp: str = "*") -> None: """ Disconnect IPCP components from a destination. Mirrors ``irm ipcp disconnect name dst [component ]``. When *comp* is ``"*"`` (default), both ``dt`` and ``mgmt`` components are disconnected, matching the CLI default. :param name: Name of the IPCP. :param dst: Destination IPCP name. :param comp: Component to disconnect: ``"dt"``, ``"mgmt"``, or ``"*"`` for both (default). """ pid = _pid_of(name) if comp in ("*", "mgmt"): _irm_disconnect_ipcp(pid, MGMT_COMP, dst) if comp in ("*", "dt"): _irm_disconnect_ipcp(pid, DT_COMP, dst) def autoboot(name: str, ipcp_type: IpcpType, layer: str, conf: Optional[IpcpConfig] = None) -> None: """ Create, autobind and bootstrap an IPCP in one step. Convenience wrapper equivalent to:: irm ipcp bootstrap name type layer autobind (with an implicit ``create`` if the IPCP does not yet exist). :param name: Name for the IPCP. :param ipcp_type: Type of IPCP to create. :param layer: Layer name to bootstrap into. :param conf: Optional IPCP configuration. If *None*, a default ``IpcpConfig`` is created for the given *ipcp_type* and *layer*. """ create_ipcp(name, ipcp_type) if conf is None: conf = IpcpConfig(ipcp_type=ipcp_type, layer_name=layer) else: conf.layer_name = layer bootstrap_ipcp(name, conf, autobind=True)