diff options
Diffstat (limited to 'ouroboros/cli.py')
| -rw-r--r-- | ouroboros/cli.py | 302 |
1 files changed, 114 insertions, 188 deletions
diff --git a/ouroboros/cli.py b/ouroboros/cli.py index 4b9ebb7..e887410 100644 --- a/ouroboros/cli.py +++ b/ouroboros/cli.py @@ -1,64 +1,13 @@ # -# Ouroboros - Copyright (C) 2016 - 2026 -# -# Python API for Ouroboros - CLI equivalents -# -# Dimitri Staessens <dimitri@ouroboros.rocks> -# -# 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/. +# SPDX-FileCopyrightText: 2016 - 2026 Dimitri Staessens +# SPDX-License-Identifier: LGPL-2.1-only # -"""Higher-level wrappers that mirror the ``irm`` CLI tool behaviour. - -This is the Python counterpart of the C project's ``ouroboros/tools/irm`` -command-line tool. The ``irm`` CLI performs steps that the raw C -library API does not, such as resolving program names via ``realpath``, -looking up IPCP pids, and handling the ``autobind`` flag during -bootstrap/enroll. This module exposes those same patterns as a Python -API so 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 +""" +Higher-level wrappers that mirror the ``irm`` CLI tool behaviour. + +Adds program path resolution, IPCP name-to-pid lookup and ``autobind`` +handling on top of the raw IRM API. """ from __future__ import annotations @@ -110,7 +59,7 @@ __all__ = [ "unbind_program", "unbind_process", "unbind_ipcp", "create_name", "destroy_name", "reg_name", "unreg_name", "list_names", "autoboot", - # Re-exported configuration types (convenience) + # Re-exported configuration types "DhtConfig", "DirConfig", "DtConfig", "EthConfig", "IpcpConfig", "IpcpInfo", "IpcpType", "LinkStateConfig", "NameInfo", "RoutingConfig", @@ -119,10 +68,14 @@ __all__ = [ def _pid_of(ipcp_name: str) -> int: - """Look up the pid of a running IPCP by its name.""" + """ + 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}") @@ -130,68 +83,83 @@ def _collect_pids(ipcp: Optional[str], ipcps: Optional[List[str]], layer: Optional[str], layers: Optional[List[str]]) -> Set[int]: - """Collect the set of pids matching the given name/layer filters.""" + """ + Collect the set of pids matching the given name/layer filters. + """ + ipcp_names: List[str] = [] + if ipcp is not None: ipcp_names.append(ipcp) + if ipcps is not None: ipcp_names.extend(ipcps) layer_names: List[str] = [] + if layer is not None: layer_names.append(layer) + if layers is not None: layer_names.extend(layers) pids: Set[int] = set() + if not ipcp_names and not layer_names: return pids 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) + return pids def destroy_ipcp(name: str) -> None: - """Destroy an IPCP by name. - - Mirrors ``irm ipcp destroy name <name>``. - - Resolves the IPCP name to a pid, then destroys the IPCP. + """ + Destroy an IPCP by name. :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. + """ + List running IPCPs, optionally filtered. - Mirrors ``irm ipcp list [name <n>] [type <t>] [layer <l>]``. + Filters are exact matches. :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 @@ -200,21 +168,10 @@ def reg_name(name: str, 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 <name> ipcp <ipcp> [ipcp ...] - layer <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: + """ + Register a name with IPCP(s), creating it first if needed. - - 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. + Registers with every IPCP in *layer* and *layers*. :param name: The name to register. :param ipcp: Single IPCP name to register with. @@ -222,6 +179,7 @@ def reg_name(name: str, :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)) @@ -234,129 +192,120 @@ 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 <prog> name <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. + """ + Bind a program to a name. :param prog: Program name or path. Bare names (without ``/``) are - resolved on ``PATH`` via ``shutil.which()``. + resolved on ``PATH``. :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. + :param argv: Arguments passed when the program is auto-started. + :raises BindError: If the program is not found on ``PATH``. """ + 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 <prog> name <name>``. + """ + Unbind a program from a 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 <pid> name <name>``. + """ + Bind a running process to a 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 <pid> name <name>``. + """ + Unbind a process from a 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 <ipcp> name <name>``. - - Resolves the IPCP name to a pid, then calls :func:`bind_process`. + """ + Bind an IPCP to a name. :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 <ipcp> name <name>``. - - Resolves the IPCP name to a pid, then calls :func:`unbind_process`. + """ + Unbind an IPCP from a name. :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 <name> [lb <policy>]``. + """ + Create a registered name. :param name: The name to create. - :param pol_lb: Load-balance policy (optional). - :param info: Full :class:`NameInfo` (overrides *name* / *pol_lb* - if given). + :param pol_lb: Load-balance policy. + :param info: Full :class:`NameInfo`; overrides *name* and + *pol_lb*. """ + 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 [<name>]``. + """ + List all registered names, optionally filtered on exact name. :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 @@ -365,12 +314,10 @@ def unreg_name(name: str, 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 <name> ipcp <ipcp> [ipcp ...] - layer <layer> [layer ...]``. + """ + Unregister a name from IPCP(s). - Accepts the same flexible input as :func:`reg_name`. + Unregisters from every IPCP in *layer* and *layers*. :param name: The name to unregister. :param ipcp: Single IPCP name to unregister from. @@ -378,6 +325,7 @@ def unreg_name(name: str, :param layer: Single layer name to unregister from. :param layers: List of layer names to unregister from. """ + for pid in _collect_pids(ipcp, ipcps, layer, layers): _irm_unreg_name(name, pid) @@ -385,28 +333,21 @@ def unreg_name(name: str, 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 <n> layer <l> [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 + """ + Bootstrap an IPCP, optionally binding it to its name and layer. - This matches the C ``irm ipcp bootstrap`` tool exactly. If - bootstrap fails after autobind, the bindings are rolled back. + Autobind applies to UNICAST and BROADCAST IPCPs only, and is rolled + back if the bootstrap fails. :param name: Name of the IPCP. - :param conf: IPCP configuration (includes layer name & type). + :param conf: IPCP configuration, including layer name and type. :param autobind: Bind the IPCP process to its name and layer. """ + pid = _pid_of(name) layer_name = conf.layer_name - autobind_types = (IpcpType.UNICAST, IpcpType.BROADCAST) + autobind_types = (IpcpType.UNICAST, IpcpType.BROADCAST) if autobind and conf.ipcp_type in autobind_types: _irm_bind_process(pid, name) _irm_bind_process(pid, layer_name) @@ -417,33 +358,26 @@ def bootstrap_ipcp(name: str, if autobind and conf.ipcp_type in autobind_types: _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 <n> layer <dst> [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. + """ + Enroll an IPCP, optionally binding it to its name and layer. :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. + :param autobind: Bind the IPCP process to its name and to the layer + learned during 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) @@ -453,44 +387,38 @@ def enroll_ipcp(name: str, dst: str, 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 <n> dst <dst> [component <c>] - [qos <qos>]``. - - When *comp* is ``"*"`` (default), both ``dt`` and ``mgmt`` - components are connected, matching the CLI default. + """ + Connect IPCP components to a destination. :param name: Name of the IPCP. :param dst: Destination IPCP name. - :param comp: Component to connect: ``"dt"``, ``"mgmt"``, or - ``"*"`` for both (default). + :param comp: ``"dt"``, ``"mgmt"``, or ``"*"`` for both. :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 <n> dst <dst> - [component <c>]``. - - When *comp* is ``"*"`` (default), both ``dt`` and ``mgmt`` - components are disconnected, matching the CLI default. + """ + Disconnect IPCP components from a destination. :param name: Name of the IPCP. :param dst: Destination IPCP name. - :param comp: Component to disconnect: ``"dt"``, ``"mgmt"``, or - ``"*"`` for both (default). + :param comp: ``"dt"``, ``"mgmt"``, or ``"*"`` for both. """ + 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) @@ -499,24 +427,22 @@ 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 <name> type <type> layer <layer> autobind - - (with an implicit ``create`` if the IPCP does not yet exist). + """ + Create, autobind and bootstrap an IPCP in one step. :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 - :class:`IpcpConfig` is created for the given - *ipcp_type* and *layer*. + :param conf: IPCP configuration. If *None*, a default + :class:`IpcpConfig` is built from *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) |