diff options
Diffstat (limited to 'ouroboros')
| -rw-r--r-- | ouroboros/__init__.py | 77 | ||||
| -rw-r--r-- | ouroboros/_qosspec.py | 61 | ||||
| -rw-r--r-- | ouroboros/_timespec.py | 59 | ||||
| -rw-r--r-- | ouroboros/cli.py | 448 | ||||
| -rw-r--r-- | ouroboros/dev.py | 596 | ||||
| -rw-r--r-- | ouroboros/errors.py | 284 | ||||
| -rw-r--r-- | ouroboros/event.py | 226 | ||||
| -rw-r--r-- | ouroboros/irm.py | 653 | ||||
| -rw-r--r-- | ouroboros/qos.py | 183 |
9 files changed, 2193 insertions, 394 deletions
diff --git a/ouroboros/__init__.py b/ouroboros/__init__.py new file mode 100644 index 0000000..cbca95d --- /dev/null +++ b/ouroboros/__init__.py @@ -0,0 +1,77 @@ +# +# SPDX-FileCopyrightText: 2016 - 2026 Dimitri Staessens +# SPDX-License-Identifier: LGPL-2.1-only +# + +""" +Public Python API for Ouroboros. + +The package surface re-exports the pure-Python helpers only: errors and +QoS. The rest of the API is imported from its submodule: flows from +:mod:`ouroboros.dev`, the event loop from :mod:`ouroboros.event`, the +control plane from :mod:`ouroboros.irm` (or :mod:`ouroboros.cli`). +""" + +from __future__ import annotations + +from importlib.metadata import PackageNotFoundError, version + +from ouroboros.errors import ( + BindError, + FlowAuthError, + FlowCryptError, + FlowDeallocWarning, + FlowDownError, + FlowError, + FlowEventError, + FlowPeerError, + FlowPermissionError, + FlowReplayError, + InvalidNameError, + IpcpBootstrapError, + IpcpConnectError, + IpcpCreateError, + IpcpEnrollError, + IpcpStateError, + IpcpTypeError, + IpcpdError, + IrmError, + IrmdError, + NameExistsError, + NameNotFoundError, + OuroborosError, +) +from ouroboros.qos import ( + QOS_MSG, + QOS_RAW, + QOS_RAW_SAFE, + QOS_RT, + QOS_RT_SAFE, + QOS_STREAM, + QoSService, + QoSSpec, +) + +try: + __version__ = version("PyOuroboros") +except PackageNotFoundError: + __version__ = "0.0.0+unknown" + +__all__ = [ + "__version__", + # qos + "QoSService", "QoSSpec", + "QOS_MSG", "QOS_RAW", "QOS_RAW_SAFE", + "QOS_RT", "QOS_RT_SAFE", "QOS_STREAM", + # errors + "OuroborosError", + "IrmError", "IrmdError", "IpcpdError", + "IpcpCreateError", "IpcpBootstrapError", "IpcpEnrollError", + "IpcpConnectError", "IpcpTypeError", "IpcpStateError", + "BindError", + "NameNotFoundError", "NameExistsError", "InvalidNameError", + "FlowError", "FlowDownError", "FlowPeerError", + "FlowPermissionError", "FlowEventError", + "FlowCryptError", "FlowAuthError", "FlowReplayError", + "FlowDeallocWarning", +] diff --git a/ouroboros/_qosspec.py b/ouroboros/_qosspec.py new file mode 100644 index 0000000..5c8b245 --- /dev/null +++ b/ouroboros/_qosspec.py @@ -0,0 +1,61 @@ +# +# SPDX-FileCopyrightText: 2016 - 2026 Dimitri Staessens +# SPDX-License-Identifier: LGPL-2.1-only +# + +""" +Shared QoSSpec <-> qosspec_t conversion helpers. + +CFFI types are not interchangeable across extension modules, so the +caller passes its own ``ffi`` instance. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Optional + +from ouroboros.qos import QoSSpec + +if TYPE_CHECKING: + from cffi import FFI + + +def qos_to_qosspec(ffi: FFI, qos: Optional[QoSSpec]) -> Any: + """ + Convert a :class:`QoSSpec` to a freshly-allocated ``qosspec_t *``. + + Returns ``ffi.NULL`` when *qos* is ``None``. + """ + + if qos is None: + return ffi.NULL + + return ffi.new("qosspec_t *", + [qos.service, + qos.delay, + qos.bandwidth, + qos.availability, + qos.loss, + qos.ber, + qos.max_gap, + qos.timeout]) + + +def qosspec_to_qos(ffi: FFI, _qos: Any) -> Optional[QoSSpec]: + """ + Convert a ``qosspec_t *`` back to a :class:`QoSSpec`. + + Returns ``None`` when *_qos* is ``ffi.NULL``. + """ + + if _qos == ffi.NULL: + return None + + return QoSSpec(service=_qos.service, + delay=_qos.delay, + bandwidth=_qos.bandwidth, + availability=_qos.availability, + loss=_qos.loss, + ber=_qos.ber, + max_gap=_qos.max_gap, + timeout=_qos.timeout) diff --git a/ouroboros/_timespec.py b/ouroboros/_timespec.py new file mode 100644 index 0000000..1656eb2 --- /dev/null +++ b/ouroboros/_timespec.py @@ -0,0 +1,59 @@ +# +# SPDX-FileCopyrightText: 2016 - 2026 Dimitri Staessens +# SPDX-License-Identifier: LGPL-2.1-only +# + +""" +Shared float-seconds <-> ``struct timespec *`` conversions. + +The caller passes its own ``ffi`` instance so the returned CData is +bound to the right CFFI extension module. +""" + +from __future__ import annotations + +from math import modf +from typing import TYPE_CHECKING, Any, Optional + +if TYPE_CHECKING: + from cffi import FFI + +BILLION = 1000 * 1000 * 1000 + + +def fl_to_timespec(ffi: FFI, timeo: Optional[float]) -> Any: + """ + Convert *timeo* (seconds, or None) to a ``struct timespec *``. + + *None* yields ``ffi.NULL`` (block forever). A non-positive *timeo* + yields a zeroed timespec (async / poll). + """ + + if timeo is None: + return ffi.NULL + + if timeo <= 0: + return ffi.new("struct timespec *", [0, 0]) + + frac, whole = modf(timeo) + _timeo = ffi.new("struct timespec *") + _timeo.tv_sec = int(whole) + _timeo.tv_nsec = int(frac * BILLION) + + return _timeo + + +def timespec_to_fl(ffi: FFI, _timeo: Any) -> Optional[float]: + """ + Convert a ``struct timespec *`` to seconds-as-float. + + ``ffi.NULL`` yields *None* (block forever). + """ + + if _timeo == ffi.NULL: + return None + + if _timeo.tv_sec <= 0 and _timeo.tv_nsec == 0: + return 0.0 + + return _timeo.tv_sec + _timeo.tv_nsec / BILLION diff --git a/ouroboros/cli.py b/ouroboros/cli.py new file mode 100644 index 0000000..e887410 --- /dev/null +++ b/ouroboros/cli.py @@ -0,0 +1,448 @@ +# +# SPDX-FileCopyrightText: 2016 - 2026 Dimitri Staessens +# SPDX-License-Identifier: LGPL-2.1-only +# + +""" +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 + +import shutil +from typing import List, Optional, Set + +from ouroboros.errors import BindError +from ouroboros.irm import ( + DT_COMP, + MGMT_COMP, + DhtConfig, + DirConfig, + DtConfig, + EthConfig, + IpcpConfig, + IpcpInfo, + IpcpType, + LinkStateConfig, + NameInfo, + RoutingConfig, + Udp4Config, + Udp6Config, + UnicastConfig, + 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 + +__all__ = [ + # Functions + "create_ipcp", "destroy_ipcp", "bootstrap_ipcp", "enroll_ipcp", + "connect_ipcp", "disconnect_ipcp", "list_ipcps", + "bind_program", "bind_process", "bind_ipcp", + "unbind_program", "unbind_process", "unbind_ipcp", + "create_name", "destroy_name", "reg_name", "unreg_name", "list_names", + "autoboot", + # Re-exported configuration types + "DhtConfig", "DirConfig", "DtConfig", "EthConfig", + "IpcpConfig", "IpcpInfo", "IpcpType", + "LinkStateConfig", "NameInfo", "RoutingConfig", + "Udp4Config", "Udp6Config", "UnicastConfig", +] + + +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 _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. + """ + + 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. + + :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. + + 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 + + +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. + + Registers with every IPCP in *layer* and *layers*. + + :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)) + + for pid in _collect_pids(ipcp, ipcps, layer, layers): + _irm_reg_name(name, pid) + + +def bind_program(prog: str, + name: str, + opts: int = 0, + argv: Optional[List[str]] = None) -> None: + """ + Bind a program to a name. + + :param prog: Program name or path. Bare names (without ``/``) are + resolved on ``PATH``. + :param name: Name to bind to. + :param opts: Bind options (e.g. ``BIND_AUTO``). + :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. + + :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. + + :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. + + :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. + + :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. + + :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. + + :param name: The name to create. + :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 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 + + +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). + + Unregisters from every IPCP in *layer* and *layers*. + + :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. + """ + + for pid in _collect_pids(ipcp, ipcps, layer, layers): + _irm_unreg_name(name, pid) + + +def bootstrap_ipcp(name: str, + conf: IpcpConfig, + autobind: bool = False) -> None: + """ + Bootstrap an IPCP, optionally binding it to its name and layer. + + 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, 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) + if autobind and conf.ipcp_type in autobind_types: + _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 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. + + :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 to the layer + learned during enrollment. + """ + + pid = _pid_of(name) + + _irm_enroll_ipcp(pid, dst) + + if autobind: + 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. + + :param name: Name of the IPCP. + :param dst: Destination IPCP name. + :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. + + :param name: Name of the IPCP. + :param dst: Destination IPCP name. + :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) + + +def autoboot(name: str, + ipcp_type: IpcpType, + layer: str, + conf: Optional[IpcpConfig] = None) -> None: + """ + 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: 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) diff --git a/ouroboros/dev.py b/ouroboros/dev.py index 7d29624..1397bca 100644 --- a/ouroboros/dev.py +++ b/ouroboros/dev.py @@ -1,398 +1,582 @@ # -# Ouroboros - Copyright (C) 2016 - 2020 -# -# Python API for applications -# -# 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 # -from _ouroboros_cffi import ffi, lib -import errno +""" +Flow allocation, acceptance and I/O for the Ouroboros dev API. +""" + +from __future__ import annotations + +import warnings from enum import IntFlag -from ouroboros.qos import * -from ouroboros.qos import _qos_to_qosspec, _fl_to_timespec, _qosspec_to_qos, _timespec_to_fl +from typing import TYPE_CHECKING, Optional, Type -# Some constants -MILLION = 1000 * 1000 -BILLION = 1000 * 1000 * 1000 +from _ouroboros_dev_cffi import ffi, lib +from ouroboros._qosspec import qos_to_qosspec, qosspec_to_qos +from ouroboros._timespec import fl_to_timespec, timespec_to_fl +from ouroboros.errors import ( + FlowAlreadyAllocatedError, + FlowDeallocWarning, + FlowNotAllocatedError, + check_ouroboros_version, + raise_errno, +) +from ouroboros.qos import QoSSpec -# ouroboros exceptions -class FlowAllocatedException(Exception): - pass +if TYPE_CHECKING: + from types import TracebackType +check_ouroboros_version(lib.OUROBOROS_VERSION_MAJOR, + lib.OUROBOROS_VERSION_MINOR) -class FlowNotAllocatedException(Exception): - pass +class FrctFlags(IntFlag): + """ + FRCT-level feature flags. + """ -class FlowDownException(Exception): - pass + RETRANSMIT = 0o1 + RESCNTL = 0o2 + LINGER = 0o4 -class FlowPermissionException(Exception): - pass +class FlowProperties(IntFlag): + """ + Flags describing flow properties and blocking behaviour. + """ + READ_ONLY = 0o0 + WRITE_ONLY = 0o1 + READ_WRITE = 0o2 + DOWN = 0o4 + NON_BLOCKING_READ = 0o1000 + NON_BLOCKING_WRITE = 0o2000 + NON_BLOCKING = NON_BLOCKING_READ | NON_BLOCKING_WRITE + NO_PARTIAL_READ = 0o10000 + NO_PARTIAL_WRITE = 0o20000 -class FlowException(Exception): - pass +class Flow: + """ + Represents an allocated Ouroboros flow. + """ -class FlowDeallocWarning(Warning): - pass + def __init__(self, fd: int = -1) -> None: + """ + Construct a Flow wrapper. + :param fd: Existing flow descriptor, or -1 for an unallocated + flow. + """ -def _raise(e: int) -> None: - if e >= 0: - return + self._fd: int = fd - print("error: " + str(e)) - if e == -errno.ETIMEDOUT: - raise TimeoutError() - if e == -errno.EINVAL: - raise ValueError() - if e == -errno.ENOMEM: - raise MemoryError() - else: - raise FlowException() + def __enter__(self) -> Flow: + return self + def __exit__(self, exc_type: Optional[Type[BaseException]], + exc_value: Optional[BaseException], + tb: Optional[TracebackType]) -> None: + self.dealloc() -class FlowProperties(IntFlag): - ReadOnly = 0o0 - WriteOnly = 0o1 - ReadWrite = 0o2 - Down = 0o4 - NonBlockingRead = 0o1000 - NonBlockingWrite = 0o2000 - NonBlocking = NonBlockingRead | NonBlockingWrite - NoPartialRead = 0o10000 - NoPartialWrite = 0o200000 + def __del__(self) -> None: + """ + Deallocate on collection; errors are swallowed as the library + may already be torn down at interpreter shutdown. + """ + try: + self.dealloc() + except Exception: # pylint: disable=broad-exception-caught + pass -class Flow: + def __repr__(self) -> str: + return f"Flow(fd={self._fd})" - def __init__(self): - self.__fd: int = -1 + def fileno(self) -> int: + """ + Return the underlying ouroboros flow descriptor. - def __enter__(self): - return self + :return: The ouroboros flow descriptor, or -1 if unallocated. + """ - def __exit__(self, exc_type, exc_value, tb): - lib.flow_dealloc(self.__fd) + return self._fd def alloc(self, dst: str, - qos: QoSSpec = None, - timeo: float = None) -> Optional[QoSSpec]: + qos: Optional[QoSSpec] = None, + timeo: Optional[float] = None) -> Optional[QoSSpec]: """ - Allocates a flow with a certain QoS to a destination + Allocate a flow with a certain QoS to a destination. - :param dst: The destination name (string) - :param qos: The QoS for the requested flow (QoSSpec) - :param timeo: The timeout for the flow allocation (None -> forever, 0->async) - :return: The QoS for the new flow + :param dst: Destination name. + :param qos: Requested QoS. + :param timeo: Allocation timeout (None blocks forever, 0 is + async). + :return: The QoS the IRM granted for the new flow. + :raises FlowAlreadyAllocatedError: If this Flow is already + allocated. + :raises FlowError: On a negative ouroboros return code. """ - if self.__fd >= 0: - raise FlowAllocatedException() + if self._fd >= 0: + raise FlowAlreadyAllocatedError() - _qos = _qos_to_qosspec(qos) + _qos = qos_to_qosspec(ffi, qos) + _timeo = fl_to_timespec(ffi, timeo) - _timeo = _fl_to_timespec(timeo) + rc = lib.flow_alloc(dst.encode(), _qos, _timeo) - self.__fd = lib.flow_alloc(dst.encode(), _qos, _timeo) + raise_errno(rc) - _raise(self.__fd) + self._fd = rc - return _qosspec_to_qos(_qos) + return qosspec_to_qos(ffi, _qos) def accept(self, - timeo: float = None) -> QoSSpec: + timeo: Optional[float] = None) -> Optional[QoSSpec]: """ - Accepts new flows and returns the QoS + Accept an incoming flow and return its QoS. - :param timeo: The timeout for the flow allocation (None -> forever, 0->async) - :return: The QoS for the new flow + :param timeo: Accept timeout (None blocks forever, 0 is async). + :return: The QoS of the accepted flow. + :raises FlowAlreadyAllocatedError: If this Flow is already + allocated. + :raises FlowError: On a negative ouroboros return code. """ - if self.__fd >= 0: - raise FlowAllocatedException() + if self._fd >= 0: + raise FlowAlreadyAllocatedError() _qos = ffi.new("qosspec_t *") + _timeo = fl_to_timespec(ffi, timeo) - _timeo = _fl_to_timespec(timeo) + rc = lib.flow_accept(_qos, _timeo) - self.__fd = lib.flow_accept(_qos, _timeo) + raise_errno(rc) - _raise(self.__fd) + self._fd = rc - return _qosspec_to_qos(_qos) + return qosspec_to_qos(ffi, _qos) def join(self, dst: str, - qos: QoSSpec = None, - timeo: float = None) -> Optional[QoSSpec]: + timeo: Optional[float] = None) -> None: """ - Join a broadcast layer + Join a broadcast layer. - :param dst: The destination broadcast layer name (string) - :param qos: The QoS for the requested flow (QoSSpec) - :param timeo: The timeout for the flow allocation (None -> forever, 0->async) - :return: The QoS for the flow + :param dst: Broadcast layer name. + :param timeo: Join timeout (None blocks forever, 0 is async). + :raises FlowAlreadyAllocatedError: If this Flow is already + allocated. + :raises FlowError: On a negative ouroboros return code. """ - if self.__fd >= 0: - raise FlowAllocatedException() - - _qos = _qos_to_qosspec(qos) + if self._fd >= 0: + raise FlowAlreadyAllocatedError() - _timeo = _fl_to_timespec(timeo) + _timeo = fl_to_timespec(ffi, timeo) - self.__fd = lib.flow_join(dst.encode(), _qos, _timeo) + rc = lib.flow_join(dst.encode(), _timeo) - _raise(self.__fd) + raise_errno(rc) - return _qosspec_to_qos(_qos) + self._fd = rc - def dealloc(self): + def dealloc(self) -> None: """ - Deallocate a flow - + Deallocate this flow; a no-op on an unallocated flow. """ - self.__fd = lib.flow_dealloc(self.__fd) + if self._fd < 0: + return - if self.__fd < 0: - raise FlowDeallocWarning + rc = lib.flow_dealloc(self._fd) + self._fd = -1 - self.__fd = -1 + if rc < 0: + warnings.warn(f"flow_dealloc returned {rc}", + FlowDeallocWarning, stacklevel=2) def write(self, buf: bytes, - count: int = None) -> int: + count: Optional[int] = None) -> int: """ - Attempt to write <count> bytes to a flow + Write up to *count* bytes to the flow. - :param buf: Buffer to write from - :param count: Number of bytes to write from the buffer - :return: Number of bytes written + :param buf: Buffer to write from. + :param count: Number of bytes to write (defaults to + ``len(buf)``). + :return: Number of bytes written. + :raises FlowError: On a negative ouroboros return code. + :raises FlowNotAllocatedError: If the flow is not allocated. """ - if self.__fd < 0: - raise FlowNotAllocatedException() + if self._fd < 0: + raise FlowNotAllocatedError() if count is None: - return lib.flow_write(self.__fd, ffi.from_buffer(buf), len(buf)) - else: - return lib.flow_write(self.__fd, ffi.from_buffer(buf), count) + count = len(buf) + + rc = lib.flow_write(self._fd, ffi.from_buffer(buf), count) + + return raise_errno(rc) def writeline(self, ln: str) -> int: """ - Attempt to write a string to a flow + Encode *ln* as UTF-8 and write it to the flow. - :param ln: String to write - :return: Number of bytes written + :param ln: String to write. + :return: Number of bytes written. + :raises FlowNotAllocatedError: If the flow is not allocated. """ - if self.__fd < 0: - raise FlowNotAllocatedException() + if self._fd < 0: + raise FlowNotAllocatedError() - return self.write(ln.encode(), len(ln)) + data = ln.encode() + + return self.write(data, len(data)) def read(self, - count: int = None) -> bytes: + count: Optional[int] = None) -> bytes: """ - Attempt to read bytes from a flow + Read up to *count* bytes from the flow. - :param count: Maximum number of bytes to read - :return: Bytes read + :param count: Maximum number of bytes to read (default 2048). + :return: Bytes read; empty when a previous read fully + consumed the SDU. + :raises FlowError: On a negative ouroboros return code. + :raises FlowNotAllocatedError: If the flow is not allocated. """ - if self.__fd < 0: - raise FlowNotAllocatedException() + if self._fd < 0: + raise FlowNotAllocatedError() if count is None: count = 2048 _buf = ffi.new("char []", count) - result = lib.flow_read(self.__fd, _buf, count) + rc = lib.flow_read(self._fd, _buf, count) + + raise_errno(rc) - return ffi.unpack(_buf, result) + return ffi.unpack(_buf, rc) - def readline(self): + def readline(self) -> str: """ + Read from the flow and decode the result as UTF-8. - :return: A string + :return: Decoded UTF-8 string read from the flow. + :raises FlowNotAllocatedError: If the flow is not allocated. """ - if self.__fd < 0: - raise FlowNotAllocatedException() + + if self._fd < 0: + raise FlowNotAllocatedError() return self.read().decode() - # flow manipulation - def set_snd_timeout(self, - timeo: float): + def set_snd_timeout(self, timeo: float) -> None: """ - Set the timeout for blocking writes + Set the timeout for blocking writes (seconds). + + :param timeo: Write timeout in seconds. + :raises FlowError: On a negative ouroboros return code. """ - _timeo = _fl_to_timespec(timeo) - if lib.flow_set_snd_timout(self.__fd, _timeo) != 0: - raise FlowPermissionException() + _timeo = fl_to_timespec(ffi, timeo) - def get_snd_timeout(self) -> float: + raise_errno(lib.flow_set_snd_timeout(self._fd, _timeo)) + + def get_snd_timeout(self) -> Optional[float]: """ - Get the timeout for blocking writes + Return the timeout for blocking writes (seconds). - :return: timeout for blocking writes + :return: Timeout for blocking writes, in seconds. + :raises FlowError: On a negative ouroboros return code. """ + _timeo = ffi.new("struct timespec *") - if lib.flow_get_snd_timeout(self.__fd, _timeo) != 0: - raise FlowPermissionException() + raise_errno(lib.flow_get_snd_timeout(self._fd, _timeo)) - return _timespec_to_fl(_timeo) + return timespec_to_fl(ffi, _timeo) - def set_rcv_timeout(self, - timeo: float): + def set_rcv_timeout(self, timeo: float) -> None: """ - Set the timeout for blocking writes + Set the timeout for blocking reads (seconds). + + :param timeo: Read timeout in seconds. + :raises FlowError: On a negative ouroboros return code. """ - _timeo = _fl_to_timespec(timeo) - if lib.flow_set_rcv_timout(self.__fd, _timeo) != 0: - raise FlowPermissionException() + _timeo = fl_to_timespec(ffi, timeo) - def get_rcv_timeout(self) -> float: + raise_errno(lib.flow_set_rcv_timeout(self._fd, _timeo)) + + def get_rcv_timeout(self) -> Optional[float]: """ - Get the timeout for blocking reads + Return the timeout for blocking reads (seconds). - :return: timeout for blocking writes + :return: Timeout for blocking reads, in seconds. + :raises FlowError: On a negative ouroboros return code. """ + _timeo = ffi.new("struct timespec *") - if lib.flow_get_rcv_timeout(self.__fd, _timeo) != 0: - raise FlowPermissionException() + raise_errno(lib.flow_get_rcv_timeout(self._fd, _timeo)) - return _timespec_to_fl(_timeo) + return timespec_to_fl(ffi, _timeo) - def get_qos(self) -> QoSSpec: + def get_qos(self) -> Optional[QoSSpec]: """ + Return the current QoS in effect on the flow. - :return: Current QoS on the flow + :return: The QoS currently in effect on the flow. + :raises FlowError: On a negative ouroboros return code. """ + _qos = ffi.new("qosspec_t *") - if lib.flow_get_qos(self.__fd, _qos) != 0: - raise FlowPermissionException() + raise_errno(lib.flow_get_qos(self._fd, _qos)) - return _qosspec_to_qos(_qos) + return qosspec_to_qos(ffi, _qos) def get_rx_queue_len(self) -> int: """ + Return the receive queue length (bytes). - :return: + :return: Receive queue length in bytes. + :raises FlowError: On a negative ouroboros return code. """ size = ffi.new("size_t *") - if lib.flow_get_rx_qlen(self.__fd, size) != 0: - raise FlowPermissionException() + raise_errno(lib.flow_get_rx_qlen(self._fd, size)) - return int(size) + return int(size[0]) def get_tx_queue_len(self) -> int: """ + Return the transmit queue length (bytes). - :return: + :return: Transmit queue length in bytes. + :raises FlowError: On a negative ouroboros return code. """ size = ffi.new("size_t *") - if lib.flow_get_tx_qlen(self.__fd, size) != 0: - raise FlowPermissionException() + raise_errno(lib.flow_get_tx_qlen(self._fd, size)) + + return int(size[0]) + + def get_mtu(self) -> int: + """ + Return the maximum user payload that fits in one n-1 PDU after + crypto headers (bytes), or 0 if unknown. + + :return: Maximum user payload in bytes, or 0 if unknown. + :raises FlowError: On a negative ouroboros return code. + """ + + mtu = ffi.new("size_t *") + + raise_errno(lib.flow_get_mtu(self._fd, mtu)) + + return int(mtu[0]) + + def set_flags(self, flags: FlowProperties) -> None: + """ + Replace the full set of flags for this flow. + + :param flags: Bitmask of :class:`FlowProperties` values. + :raises FlowError: On a negative ouroboros return code. + """ + + raise_errno(lib.flow_set_flags(self._fd, int(flags))) + + def add_flags(self, flags: FlowProperties) -> None: + """ + OR *flags* into the current flags, leaving other bits set. + + :param flags: :class:`FlowProperties` bits to set. + :raises FlowError: On a negative ouroboros return code. + """ + + current = self.get_flags() - return int(size) + self.set_flags(current | FlowProperties(int(flags))) - def set_flags(self, flags: FlowProperties): + def remove_flags(self, flags: FlowProperties) -> None: """ - Set flags for this flow. - :param flags: + Clear *flags* from the current flags, leaving other bits set. + + :param flags: :class:`FlowProperties` bits to clear. + :raises FlowError: On a negative ouroboros return code. """ - _flags = ffi.new("uint32_t *", int(flags)) + current = self.get_flags() - if lib.flow_set_flag(self.__fd, _flags): - raise FlowPermissionException() + self.set_flags(current & ~FlowProperties(int(flags))) def get_flags(self) -> FlowProperties: """ - Get the flags for this flow + Return the current flags for this flow. + + :return: The current :class:`FlowProperties` bitmask. + :raises FlowError: On a negative ouroboros return code. """ - flags = lib.flow_get_flag(self.__fd) - if flags < 0: - raise FlowPermissionException() + flags = raise_errno(lib.flow_get_flags(self._fd)) return FlowProperties(int(flags)) + def set_frct_flags(self, flags: FrctFlags) -> None: + """ + Set FRCT flags for this flow. + + :param flags: Bitmask of :class:`FrctFlags`. + :raises FlowError: On a negative ouroboros return code. + """ + + raise_errno(lib.flow_set_frct_flags(self._fd, int(flags))) + + def get_frct_flags(self) -> FrctFlags: + """ + Return the FRCT flags for this flow. + + :return: The current :class:`FrctFlags` bitmask. + :raises FlowError: On a negative ouroboros return code. + """ + + flags = raise_errno(lib.flow_get_frct_flags(self._fd)) + + return FrctFlags(int(flags)) + + def set_frct_max_sdu(self, size: int) -> None: + """ + Set the maximum reassembly SDU size for FRCT (bytes). + + :param size: Maximum reassembly SDU size in bytes. + :raises FlowError: On a negative ouroboros return code. + """ + + raise_errno(lib.flow_set_frct_max_sdu(self._fd, size)) + + def get_frct_max_sdu(self) -> int: + """ + Return the maximum reassembly SDU size for FRCT (bytes). + + :return: Maximum reassembly SDU size in bytes. + :raises FlowError: On a negative ouroboros return code. + """ + + size = ffi.new("size_t *") + + raise_errno(lib.flow_get_frct_max_sdu(self._fd, size)) + + return int(size[0]) + + def set_frct_rcv_ring_size(self, size: int) -> None: + """ + Set the stream receive ring size (bytes, power of two). + + :param size: Stream receive ring size in bytes (power of two). + :raises FlowError: On a negative ouroboros return code. + """ + + raise_errno(lib.flow_set_frct_rcv_ring_sz(self._fd, size)) + + def get_frct_rcv_ring_size(self) -> int: + """ + Return the stream receive ring size (bytes). + + :return: Stream receive ring size in bytes. + :raises FlowError: On a negative ouroboros return code. + """ + + size = ffi.new("size_t *") + + raise_errno(lib.flow_get_frct_rcv_ring_sz(self._fd, size)) + + return int(size[0]) + + def set_frct_rto_min(self, rto_ns: int) -> None: + """ + Set the FRCT RTO floor (nanoseconds). + + :param rto_ns: FRCT RTO floor in nanoseconds. + :raises FlowError: On a negative ouroboros return code. + """ + + raise_errno(lib.flow_set_frct_rto_min(self._fd, rto_ns)) + + def get_frct_rto_min(self) -> int: + """ + Return the FRCT RTO floor (nanoseconds). + + :return: FRCT RTO floor in nanoseconds. + :raises FlowError: On a negative ouroboros return code. + """ + + rto = ffi.new("time_t *") + + raise_errno(lib.flow_get_frct_rto_min(self._fd, rto)) + + return int(rto[0]) + def flow_alloc(dst: str, - qos: QoSSpec = None, - timeo: float = None) -> Flow: + qos: Optional[QoSSpec] = None, + timeo: Optional[float] = None) -> Flow: """ + Allocate a new flow and return the :class:`Flow` wrapper. - :param dst: Destination name - :param qos: Requested QoS - :param timeo: Timeout to wait for the allocation - :return: A new Flow() + :param dst: Destination name. + :param qos: Requested QoS. + :param timeo: Allocation timeout (None blocks forever, 0 is async). + :return: The allocated flow. """ f = Flow() + f.alloc(dst, qos, timeo) + return f -def flow_accept(timeo: float = None) -> Flow: +def flow_accept(timeo: Optional[float] = None) -> Flow: """ + Accept an incoming flow and return the :class:`Flow` wrapper. - :param timeo: Timeout to wait for the allocation - :return: A new Flow() + :param timeo: Accept timeout (None blocks forever, 0 is async). + :return: The accepted flow. """ f = Flow() + f.accept(timeo) + return f def flow_join(dst: str, - qos: QoSSpec = None, - timeo: float = None) -> Flow: + timeo: Optional[float] = None) -> Flow: """ + Join a broadcast layer and return the :class:`Flow` wrapper. - :param dst: Broadcast layer name - :param qos: Requested QoS - :param timeo: Timeout to wait for the allocation - :return: A new Flow() + :param dst: Broadcast layer name. + :param timeo: Join timeout (None blocks forever, 0 is async). + :return: The joined broadcast flow. """ f = Flow() - f.join(dst, qos, timeo) - return f + f.join(dst, timeo) + return f diff --git a/ouroboros/errors.py b/ouroboros/errors.py new file mode 100644 index 0000000..6ab6053 --- /dev/null +++ b/ouroboros/errors.py @@ -0,0 +1,284 @@ +# +# SPDX-FileCopyrightText: 2016 - 2026 Dimitri Staessens +# SPDX-License-Identifier: LGPL-2.1-only +# + +""" +Exception hierarchy and errno translation for pyouroboros. + +The ``E*`` constants mirror the Ouroboros-specific errno values in +``include/ouroboros/errno.h``. +""" + +from __future__ import annotations + +import errno +import os +from importlib.metadata import PackageNotFoundError, version + +ENOTALLOC = 1000 # Flow is not allocated +EIPCPTYPE = 1001 # Unknown IPCP type +EIRMD = 1002 # Failed to communicate with IRMD +EIPCP = 1003 # Failed to communicate with IPCP +EIPCPSTATE = 1004 # Target in wrong state +EFLOWDOWN = 1005 # Flow is down +EFLOWPEER = 1006 # Flow is down (peer timed out) +ENAME = 1007 # Naming error +ECRYPT = 1008 # Encryption error +EAUTH = 1009 # Authentication error +EREPLAY = 1010 # OAP replay detected + +_O7S_ERRNO_STR: dict[int, str] = { + ENOTALLOC: "flow is not allocated", + EIPCPTYPE: "unknown IPCP type", + EIRMD: "failed to communicate with IRMd", + EIPCP: "failed to communicate with IPCP", + EIPCPSTATE: "target in wrong state", + EFLOWDOWN: "flow is down", + EFLOWPEER: "flow peer timed out", + ENAME: "naming error", + ECRYPT: "encryption error", + EAUTH: "authentication error", + EREPLAY: "OAP replay detected", +} + + +class OuroborosError(Exception): + """ + Base class for all pyouroboros exceptions. + """ + + +class IrmError(OuroborosError): + """ + Base class for IRM (control plane) errors. + """ + + +class IpcpCreateError(IrmError): + """ + Raised when IPCP creation fails. + """ + + +class IpcpBootstrapError(IrmError): + """ + Raised when IPCP bootstrapping fails. + """ + + +class IpcpEnrollError(IrmError): + """ + Raised when IPCP enrollment fails. + """ + + +class IpcpConnectError(IrmError): + """ + Raised when IPCP connection or disconnection fails. + """ + + +class IpcpTypeError(IrmError, ValueError): + """ + Raised when an unknown IPCP type is encountered. + """ + + +class IpcpStateError(IrmError): + """ + Raised when an IPCP/IRMd is in the wrong state for the request. + """ + + +class IrmdError(IrmError): + """ + Raised when the IRM daemon is unreachable. + """ + + +class IpcpdError(IrmError): + """ + Raised when the IPCP daemon is unreachable. + """ + + +class BindError(IrmError): + """ + Raised when binding a program/process/IPCP to a name fails. + """ + + +class NameNotFoundError(IrmError): + """ + Raised when a name lookup or unregister target is missing. + """ + + +class NameExistsError(IrmError): + """ + Raised when a name already exists. + """ + + +class InvalidNameError(IrmError, ValueError): + """ + Raised when a name is malformed or otherwise invalid. + """ + + +class FlowError(OuroborosError): + """ + Base class for flow-plane errors. + """ + + +class FlowAlreadyAllocatedError(FlowError): + """ + Raised when allocating on a Flow that already holds an fd. + """ + + +class FlowNotAllocatedError(FlowError): + """ + Raised when operating on a Flow that has no fd. + """ + + +class FlowDownError(FlowError, ConnectionError): + """ + Raised when the flow has gone down. + """ + + +class FlowPeerError(FlowDownError): + """ + Raised when the flow's peer has timed out. + """ + + +class FlowPermissionError(FlowError, PermissionError): + """ + Raised when an operation is not permitted on the flow. + """ + + +class FlowTimeout(FlowError, TimeoutError): + """ + Raised when a flow operation times out. + """ + + +class FlowCryptError(FlowError): + """ + Raised on flow encryption errors. + """ + + +class FlowAuthError(FlowError): + """ + Raised on flow authentication errors. + """ + + +class FlowReplayError(FlowError): + """ + Raised when OAP replay is detected. + """ + + +class FlowEventError(FlowError): + """ + Raised on errors from the flow event subsystem. + """ + + +class FlowDeallocWarning(Warning): + """ + Warning issued when a deallocation reports a non-fatal problem. + """ + + +_ERRNO_MAP: dict[int, type[BaseException]] = { + errno.ETIMEDOUT: TimeoutError, + errno.EAGAIN: BlockingIOError, + errno.EWOULDBLOCK: BlockingIOError, + errno.ENOMEM: MemoryError, + errno.EACCES: PermissionError, + errno.ENOTCONN: ConnectionError, + errno.ECONNRESET: ConnectionResetError, + ENOTALLOC: FlowNotAllocatedError, + EIPCPTYPE: IpcpTypeError, + EIRMD: IrmdError, + EIPCP: IpcpdError, + EIPCPSTATE: IpcpStateError, + EFLOWDOWN: FlowDownError, + EFLOWPEER: FlowPeerError, + ENAME: NameNotFoundError, + ECRYPT: FlowCryptError, + EAUTH: FlowAuthError, + EREPLAY: FlowReplayError, +} + + +def _strerror(err: int) -> str: + if err >= 1000: + return _O7S_ERRNO_STR.get(err, f"ouroboros errno {err}") + + return os.strerror(err) + + +def raise_errno(rc: int, default: type[OuroborosError] = FlowError) -> int: + """ + Translate a non-negative-or-negative-errno return code. + + ``_ERRNO_MAP`` holds errnos in their canonical positive form. + + :param rc: The return code to translate; a negative + value is a negated errno. + :param default: Exception class to raise when the errno + has no mapping; a mapped errno overrides + it. + :return: *rc* unchanged when it is non-negative. + :raises OuroborosError: (or the builtin mapped for ``-rc``) when + *rc* is negative. + """ + + if rc >= 0: + return rc + + err = -rc + exc_cls = _ERRNO_MAP.get(err, default) + raise exc_cls(_strerror(err)) + + +def check_ouroboros_version(o7s_major: int, o7s_minor: int) -> None: + """ + Verify that the installed pyouroboros matches the linked library. + + Compares the linked library version against the installed + pyouroboros distribution metadata. Silently returns when running + from a source tree without distribution metadata. + + :param o7s_major: Major version of the linked library, from + ``OUROBOROS_VERSION_MAJOR`` in the CFFI + module. + :param o7s_minor: Minor version of the linked library, from + ``OUROBOROS_VERSION_MINOR`` in the CFFI + module. + :raises RuntimeError: If the library and pyouroboros ``major.minor`` + differ. + """ + + try: + pyo7s_parts = version("PyOuroboros").split(".") + except PackageNotFoundError: + return + + if o7s_major != int(pyo7s_parts[0]) or \ + o7s_minor != int(pyo7s_parts[1]): + raise RuntimeError( + f"Ouroboros version mismatch: library is " + f"{o7s_major}.{o7s_minor}, pyouroboros is " + f"{pyo7s_parts[0]}.{pyo7s_parts[1]}" + ) diff --git a/ouroboros/event.py b/ouroboros/event.py index b707c1b..24b58b8 100644 --- a/ouroboros/event.py +++ b/ouroboros/event.py @@ -1,147 +1,197 @@ # -# Ouroboros - Copyright (C) 2016 - 2020 -# -# Python API for applications -# -# 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 # -from ouroboros.dev import * -from ouroboros.qos import _fl_to_timespec +""" +Asynchronous flow event monitoring for Ouroboros. +""" +from __future__ import annotations -# async API -class FlowEventError(Exception): - pass +from enum import IntFlag +from typing import TYPE_CHECKING, Any, List, Optional, Tuple, Type + +from _ouroboros_dev_cffi import ffi, lib + +from ouroboros._timespec import fl_to_timespec +from ouroboros.dev import Flow +from ouroboros.errors import FlowEventError, raise_errno + +if TYPE_CHECKING: + from types import TracebackType class FEventType(IntFlag): - FlowPkt = lib.FLOW_PKT - FlowDown = lib.FLOW_DOWN - FlowUp = lib.FLOW_UP - FlowAlloc = lib.FLOW_ALLOC - FlowDealloc = lib.FLOW_DEALLOC + """ + Types of flow events. + """ + + FLOW_PKT = lib.FLOW_PKT + FLOW_DOWN = lib.FLOW_DOWN + FLOW_UP = lib.FLOW_UP + FLOW_ALLOC = lib.FLOW_ALLOC + FLOW_DEALLOC = lib.FLOW_DEALLOC + FLOW_PEER = lib.FLOW_PEER class FEventQueue: """ - A queue of events waiting to be handled + A queue of flow events waiting to be drained. """ - def __init__(self): - self.__fq = lib.fqueue_create() - if self.__fq is ffi.NULL: + def __init__(self) -> None: + self._fq = lib.fqueue_create() + if self._fq == ffi.NULL: raise MemoryError("Failed to create FEventQueue") - def __del__(self): - lib.fqueue_destroy(self.__fq) + def __enter__(self) -> FEventQueue: + return self + + def __exit__(self, exc_type: Optional[Type[BaseException]], + exc_value: Optional[BaseException], + tb: Optional[TracebackType]) -> None: + self.destroy() - def next(self): + def __del__(self) -> None: + self.destroy() + + def destroy(self) -> None: """ - Get the next event - :return: Flow and eventtype on that flow + Destroy the underlying queue. Idempotent. """ - f = Flow() - f._Flow__fd = lib.fqueue_next(self.__fq) - if f._Flow__fd < 0: - raise FlowEventError - _type = lib.fqueue_type(self.__fq) - if _type < 0: - raise FlowEventError + if self._fq != ffi.NULL: + lib.fqueue_destroy(self._fq) + + self._fq = ffi.NULL + + def next(self) -> Tuple[Flow, FEventType]: + """ + Return the next ``(flow, event_type)`` pair from the queue. + + :return: The :class:`~ouroboros.dev.Flow` the event occurred + on, and its :class:`FEventType`. + :raises FlowEventError: On a negative ouroboros return code. + """ - return f, _type + fd = lib.fqueue_next(self._fq) + + raise_errno(fd, default=FlowEventError) + + _type = lib.fqueue_type(self._fq) + + raise_errno(_type, default=FlowEventError) + + return Flow(fd), FEventType(_type) + + @property + def handle(self) -> Any: + """ + Return the underlying ``fqueue_t *``. + + :return: The underlying ``fqueue_t *`` CFFI pointer. + """ + + return self._fq class FlowSet: """ - A set of flows that can be monitored for events + A set of flows that can be monitored for events. """ - def __init__(self, - flows: [Flow] = None): - self.__set = lib.fset_create() - if self.__set is ffi.NULL: + def __init__(self, flows: Optional[List[Flow]] = None) -> None: + self._set = lib.fset_create() + if self._set == ffi.NULL: raise MemoryError("Failed to create FlowSet") if flows is not None: for flow in flows: - if lib.fset_add(self.__set, flow._Flow__fd) != 0: - lib.fset_destroy(self.__set) - self.__set = ffi.NULL - raise MemoryError("Failed to add flow " + str(flow._Flow__fd) + ".") + if lib.fset_add(self._set, flow.fileno()) != 0: + lib.fset_destroy(self._set) - def __enter__(self): + self._set = ffi.NULL + raise MemoryError(f"Failed to add flow {flow.fileno()}.") + + def __enter__(self) -> FlowSet: return self - def __exit__(self, exc_type, exc_value, tb): - lib.fset_destroy(self.__set) + def __exit__(self, exc_type: Optional[Type[BaseException]], + exc_value: Optional[BaseException], + tb: Optional[TracebackType]) -> None: + self.destroy() + + def __del__(self) -> None: + self.destroy() - def add(self, - flow: Flow): + def destroy(self) -> None: + """ + Destroy the underlying set. Idempotent. """ - Add a Flow - :param flow: The flow object to add + if self._set != ffi.NULL: + lib.fset_destroy(self._set) + + self._set = ffi.NULL + + def add(self, flow: Flow) -> None: """ + Add *flow* to this set. - if self.__set is ffi.NULL: - raise ValueError + :param flow: The flow to add to this set. + :raises ValueError: If the FlowSet has been destroyed. + :raises MemoryError: If the flow could not be added. + """ + + if self._set == ffi.NULL: + raise ValueError("FlowSet has been destroyed") - if lib.fset_add(self.__set, flow._Flow___fd) != 0: - raise MemoryError("Failed to add flow") + if lib.fset_add(self._set, flow.fileno()) != 0: + raise MemoryError(f"Failed to add flow {flow.fileno()}") - def zero(self): + def zero(self) -> None: """ - Remove all Flows from this set + Remove all flows from this set. + + :raises ValueError: If the FlowSet has been destroyed. """ - if self.__set is ffi.NULL: - raise ValueError + if self._set == ffi.NULL: + raise ValueError("FlowSet has been destroyed") - lib.fset_zero(self.__set) + lib.fset_zero(self._set) - def remove(self, - flow: Flow): + def remove(self, flow: Flow) -> None: """ - Remove a flow from a set + Remove *flow* from this set. - :param flow: + :param flow: The flow to remove from this set. + :raises ValueError: If the FlowSet has been destroyed. """ - if self.__set is ffi.NULL: - raise ValueError + if self._set == ffi.NULL: + raise ValueError("FlowSet has been destroyed") - lib.fset_del(self.__set, flow._Flow__fd) + lib.fset_del(self._set, flow.fileno()) def wait(self, - fq: FEventType, - timeo: float = None): + fq: FEventQueue, + timeo: Optional[float] = None) -> None: """ - Wait for at least one event on one of the monitored flows + Wait for at least one event on one of the monitored flows. + + :param fq: The :class:`FEventQueue` to drain events into. + :param timeo: Wait timeout in seconds (None blocks forever). + :raises ValueError: If the FlowSet has been destroyed. + :raises FlowEventError: On a negative ouroboros return code. """ - if self.__set is ffi.NULL: - raise ValueError + if self._set == ffi.NULL: + raise ValueError("FlowSet has been destroyed") - _timeo = _fl_to_timespec(timeo) + _timeo = fl_to_timespec(ffi, timeo) - ret = lib.fevent(self.__set, fq._FEventQueue__fq, _timeo) - if ret < 0: - raise FlowEventError + rc = lib.fevent(self._set, fq.handle, _timeo) - def destroy(self): - lib.fset_destroy(self.__set) + raise_errno(rc, default=FlowEventError) diff --git a/ouroboros/irm.py b/ouroboros/irm.py new file mode 100644 index 0000000..ea7d665 --- /dev/null +++ b/ouroboros/irm.py @@ -0,0 +1,653 @@ +# +# SPDX-FileCopyrightText: 2016 - 2026 Dimitri Staessens +# SPDX-License-Identifier: LGPL-2.1-only +# + +""" +IRM (IPC Resource Manager) bindings for Ouroboros. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import IntEnum +from typing import Any, List, Optional + +from _ouroboros_irm_cffi import ffi, lib + +from ouroboros._qosspec import qos_to_qosspec +from ouroboros.errors import ( + BindError, + IpcpBootstrapError, + IpcpConnectError, + IpcpCreateError, + IpcpEnrollError, + IrmError, + NameExistsError, + NameNotFoundError, + check_ouroboros_version, + raise_errno, +) +from ouroboros.qos import QoSSpec + +check_ouroboros_version(lib.OUROBOROS_VERSION_MAJOR, + lib.OUROBOROS_VERSION_MINOR) + +BIND_AUTO = lib.BIND_AUTO + +DT_COMP = "Data Transfer" +MGMT_COMP = "Management" + + +class IpcpType(IntEnum): + """ + IPCP types available in Ouroboros. + """ + + LOCAL = lib.IPCP_LOCAL + UNICAST = lib.IPCP_UNICAST + BROADCAST = lib.IPCP_BROADCAST + ETH_LLC = lib.IPCP_ETH_LLC + ETH_DIX = lib.IPCP_ETH_DIX + UDP4 = lib.IPCP_UDP4 + UDP6 = lib.IPCP_UDP6 + + +class AddressAuthPolicy(IntEnum): + """ + Address authority policies for unicast IPCPs. + """ + + FLAT_RANDOM = lib.ADDR_AUTH_FLAT_RANDOM + + +class LinkStatePolicy(IntEnum): + """ + Link state routing policies. + """ + + SIMPLE = lib.LS_SIMPLE + LFA = lib.LS_LFA + ECMP = lib.LS_ECMP + + +class RoutingPolicy(IntEnum): + """ + Routing policies. + """ + + LINK_STATE = lib.ROUTING_LINK_STATE + + +class CongestionAvoidPolicy(IntEnum): + """ + Congestion avoidance policies. + """ + + NONE = lib.CA_NONE + MB_ECN = lib.CA_MB_ECN + + +class DirectoryPolicy(IntEnum): + """ + Directory policies. + """ + + DHT = lib.DIR_DHT + + +class DirectoryHashAlgo(IntEnum): + """ + Directory hash algorithms. + """ + + SHA3_224 = lib.DIR_HASH_SHA3_224 + SHA3_256 = lib.DIR_HASH_SHA3_256 + SHA3_384 = lib.DIR_HASH_SHA3_384 + SHA3_512 = lib.DIR_HASH_SHA3_512 + + +class LoadBalancePolicy(IntEnum): + """ + Load balancing policies for names. + """ + + ROUND_ROBIN = lib.LB_RR + SPILL = lib.LB_SPILL + + +@dataclass +class LinkStateConfig: + """ + Configuration for link state routing. + """ + + pol: LinkStatePolicy = LinkStatePolicy.SIMPLE + t_recalc: int = 4 + t_update: int = 15 + t_timeo: int = 60 + + +@dataclass +class RoutingConfig: + """ + Routing configuration. + """ + + pol: RoutingPolicy = RoutingPolicy.LINK_STATE + ls: LinkStateConfig = field(default_factory=LinkStateConfig) + + +@dataclass +class DtConfig: + """ + Data transfer configuration for unicast IPCPs. + """ + + addr_size: int = 4 + eid_size: int = 8 + max_ttl: int = 60 + routing: RoutingConfig = field(default_factory=RoutingConfig) + + +@dataclass +class DhtConfig: + """ + DHT directory configuration. + """ + + alpha: int = 3 + k: int = 8 + t_expire: int = 86400 + t_refresh: int = 900 + t_replicate: int = 900 + + +@dataclass +class DirConfig: + """ + Directory configuration. + """ + + pol: DirectoryPolicy = DirectoryPolicy.DHT + dht: DhtConfig = field(default_factory=DhtConfig) + + +@dataclass +class UnicastConfig: + """ + Configuration for unicast IPCPs. + """ + + dt: DtConfig = field(default_factory=DtConfig) + dir: DirConfig = field(default_factory=DirConfig) + addr_auth: AddressAuthPolicy = AddressAuthPolicy.FLAT_RANDOM + cong_avoid: CongestionAvoidPolicy = CongestionAvoidPolicy.MB_ECN + + +@dataclass +class EthConfig: + """ + Configuration for Ethernet IPCPs (LLC or DIX). + """ + + dev: str = "" + ethertype: int = 0xA000 + + +@dataclass +class Udp4Config: + """ + Configuration for UDP over IPv4 IPCPs. + """ + + ip_addr: str = "0.0.0.0" + dns_addr: str = "0.0.0.0" + port: int = 3435 + + +@dataclass +class Udp6Config: + """ + Configuration for UDP over IPv6 IPCPs. + """ + + ip_addr: str = "::" + dns_addr: str = "::" + port: int = 3435 + + +@dataclass +class IpcpConfig: + """ + Configuration for bootstrapping an IPCP. + """ + + ipcp_type: IpcpType + layer_name: str = "" + dir_hash_algo: DirectoryHashAlgo = DirectoryHashAlgo.SHA3_256 + unicast: Optional[UnicastConfig] = None + eth: Optional[EthConfig] = None + udp4: Optional[Udp4Config] = None + udp6: Optional[Udp6Config] = None + + +@dataclass +class NameSecPaths: + """ + Security paths for a name. + """ + + sec: str = "" + key: str = "" + crt: str = "" + + +@dataclass +class NameInfo: + """ + Information about a registered name. + """ + + name: str + pol_lb: LoadBalancePolicy = LoadBalancePolicy.ROUND_ROBIN + server_sec: NameSecPaths = field(default_factory=NameSecPaths) + client_sec: NameSecPaths = field(default_factory=NameSecPaths) + + +@dataclass +class IpcpInfo: + """ + Information about a running IPCP. + """ + + pid: int + type: IpcpType + name: str + layer: str + + +def _ipcp_config_to_c(conf: IpcpConfig) -> Any: + """ + Convert an :class:`IpcpConfig` to a C ``struct ipcp_config *``. + + The layer name is truncated to the 255 bytes the C struct holds. + """ + + _conf = ffi.new("struct ipcp_config *") + + layer_name = conf.layer_name.encode() + + ffi.memmove(_conf.layer_info.name, layer_name, + min(len(layer_name), 255)) + + _conf.layer_info.dir_hash_algo = conf.dir_hash_algo + + _conf.type = conf.ipcp_type + + if conf.ipcp_type == IpcpType.UNICAST: + uc = conf.unicast or UnicastConfig() + _conf.unicast.dt.addr_size = uc.dt.addr_size + _conf.unicast.dt.eid_size = uc.dt.eid_size + _conf.unicast.dt.max_ttl = uc.dt.max_ttl + _conf.unicast.dt.routing.pol = uc.dt.routing.pol + _conf.unicast.dt.routing.ls.pol = uc.dt.routing.ls.pol + _conf.unicast.dt.routing.ls.t_recalc = uc.dt.routing.ls.t_recalc + _conf.unicast.dt.routing.ls.t_update = uc.dt.routing.ls.t_update + _conf.unicast.dt.routing.ls.t_timeo = uc.dt.routing.ls.t_timeo + _conf.unicast.dir.pol = uc.dir.pol + _conf.unicast.dir.dht.params.alpha = uc.dir.dht.alpha + _conf.unicast.dir.dht.params.k = uc.dir.dht.k + _conf.unicast.dir.dht.params.t_expire = uc.dir.dht.t_expire + _conf.unicast.dir.dht.params.t_refresh = uc.dir.dht.t_refresh + _conf.unicast.dir.dht.params.t_replicate = uc.dir.dht.t_replicate + _conf.unicast.addr_auth_type = uc.addr_auth + _conf.unicast.cong_avoid = uc.cong_avoid + + elif conf.ipcp_type in (IpcpType.ETH_LLC, IpcpType.ETH_DIX): + ec = conf.eth or EthConfig() + dev = ec.dev.encode() + + ffi.memmove(_conf.eth.dev, dev, min(len(dev), 255)) + + _conf.eth.ethertype = ec.ethertype + + elif conf.ipcp_type == IpcpType.UDP4: + uc4 = conf.udp4 or Udp4Config() + _conf.udp4.port = uc4.port + if lib.ipcp_config_udp4_set_ip(_conf, uc4.ip_addr.encode()) != 0: + raise ValueError(f"Invalid IPv4 address: {uc4.ip_addr}") + + if lib.ipcp_config_udp4_set_dns(_conf, uc4.dns_addr.encode()) != 0: + raise ValueError(f"Invalid IPv4 DNS address: {uc4.dns_addr}") + + elif conf.ipcp_type == IpcpType.UDP6: + uc6 = conf.udp6 or Udp6Config() + _conf.udp6.port = uc6.port + if lib.ipcp_config_udp6_set_ip(_conf, uc6.ip_addr.encode()) != 0: + raise ValueError(f"Invalid IPv6 address: {uc6.ip_addr}") + + if lib.ipcp_config_udp6_set_dns(_conf, uc6.dns_addr.encode()) != 0: + raise ValueError(f"Invalid IPv6 DNS address: {uc6.dns_addr}") + + return _conf + + +def _name_info_to_c(info: NameInfo) -> Any: + """ + Convert a :class:`NameInfo` to a C ``struct name_info *``. + """ + + _info = ffi.new("struct name_info *") + + name = info.name.encode() + + ffi.memmove(_info.name, name, min(len(name), 255)) + + _info.pol_lb = info.pol_lb + + for attr, sec in (('s', info.server_sec), ('c', info.client_sec)): + sec_paths = getattr(_info, attr) + + for fld in ('sec', 'key', 'crt'): + val = getattr(sec, fld).encode() + + ffi.memmove(getattr(sec_paths, fld), val, + min(len(val), 511)) + + return _info + + +def create_ipcp(name: str, + ipcp_type: IpcpType) -> int: + """ + Create a new IPCP. + + :param name: Name for the IPCP. + :param ipcp_type: Type of IPCP to create. + :return: PID of the created IPCP. + :raises IpcpCreateError: If creation fails or the IPCP is not found. + """ + + raise_errno(lib.irm_create_ipcp(name.encode(), ipcp_type), + default=IpcpCreateError) + + for info in list_ipcps(): + if info.name == name: + return info.pid + + raise IpcpCreateError(f"IPCP '{name}' created but not found in list") + + +def destroy_ipcp(pid: int) -> None: + """ + Destroy an IPCP. + + :param pid: PID of the IPCP to destroy. + :raises IrmError: If the IPCP could not be destroyed. + """ + + raise_errno(lib.irm_destroy_ipcp(pid), default=IrmError) + + +def list_ipcps() -> List[IpcpInfo]: + """ + List all running IPCPs. + + :return: List of :class:`IpcpInfo` objects. + :raises IrmError: If the IPCPs could not be listed. + """ + + _ipcps = ffi.new("struct ipcp_list_info **") + n = raise_errno(lib.irm_list_ipcps(_ipcps), default=IrmError) + + result = [] + + for i in range(n): + info = _ipcps[0][i] + + result.append(IpcpInfo( + pid=info.pid, + type=IpcpType(info.type), + name=ffi.string(info.name).decode(), + layer=ffi.string(info.layer).decode(), + )) + + if n > 0: + lib.free(_ipcps[0]) + + return result + + +def enroll_ipcp(pid: int, dst: str) -> None: + """ + Enroll an IPCP in a layer. + + :param pid: PID of the IPCP to enroll. + :param dst: Name of a member of the layer to enroll with. + :raises IpcpEnrollError: If the IPCP could not enroll with *dst*. + """ + + raise_errno(lib.irm_enroll_ipcp(pid, dst.encode()), + default=IpcpEnrollError) + + +def bootstrap_ipcp(pid: int, conf: IpcpConfig) -> None: + """ + Bootstrap an IPCP. + + :param pid: PID of the IPCP to bootstrap. + :param conf: Configuration for the IPCP. + :raises IpcpBootstrapError: If the IPCP could not be bootstrapped. + """ + + _conf = _ipcp_config_to_c(conf) + + raise_errno(lib.irm_bootstrap_ipcp(pid, _conf), + default=IpcpBootstrapError) + + +def connect_ipcp(pid: int, + component: str, + dst: str, + qos: Optional[QoSSpec] = None) -> None: + """ + Connect an IPCP component to a destination. + + :param pid: PID of the IPCP. + :param component: :data:`DT_COMP` or :data:`MGMT_COMP`. + :param dst: Destination name. + :param qos: QoS specification for the connection. + :raises IpcpConnectError: If *component* could not be + connected to *dst*. + """ + + _qos = qos_to_qosspec(ffi, qos) + if _qos == ffi.NULL: + _qos = ffi.new("qosspec_t *") + + raise_errno( + lib.irm_connect_ipcp(pid, dst.encode(), component.encode(), + _qos[0]), + default=IpcpConnectError, + ) + + +def disconnect_ipcp(pid: int, + component: str, + dst: str) -> None: + """ + Disconnect an IPCP component from a destination. + + :param pid: PID of the IPCP. + :param component: :data:`DT_COMP` or :data:`MGMT_COMP`. + :param dst: Destination name. + :raises IpcpConnectError: If *component* could not be + disconnected from *dst*. + """ + + raise_errno( + lib.irm_disconnect_ipcp(pid, dst.encode(), component.encode()), + default=IpcpConnectError, + ) + + +def bind_program(prog: str, + name: str, + opts: int = 0, + argv: Optional[List[str]] = None) -> None: + """ + Bind a program to a name. + + :param prog: Path to the program. + :param name: Name to bind to. + :param opts: Bind options (e.g. :data:`BIND_AUTO`). + :param argv: Arguments passed when the IRM starts the program. + :raises BindError: If *prog* could not be bound to *name*. + """ + + if argv: + argc = len(argv) + _args = [] + + for a in argv: + _args.append(ffi.new("char[]", a.encode())) + + _argv = ffi.new("char *[]", _args) + else: + argc = 0 + _argv = ffi.NULL + + raise_errno( + lib.irm_bind_program(prog.encode(), name.encode(), + opts, argc, _argv), + default=BindError, + ) + + +def unbind_program(prog: str, name: str) -> None: + """ + Unbind a program from a name. + + :param prog: Path to the program. + :param name: Name to unbind from. + :raises BindError: If *prog* could not be unbound from *name*. + """ + + raise_errno(lib.irm_unbind_program(prog.encode(), name.encode()), + default=BindError) + + +def bind_process(pid: int, name: str) -> None: + """ + Bind a running process to a name. + + :param pid: PID of the process. + :param name: Name to bind to. + :raises BindError: If the process could not be bound to *name*. + """ + + raise_errno(lib.irm_bind_process(pid, name.encode()), + default=BindError) + + +def unbind_process(pid: int, name: str) -> None: + """ + Unbind a process from a name. + + :param pid: PID of the process. + :param name: Name to unbind from. + :raises BindError: If the process could not be unbound from *name*. + """ + + raise_errno(lib.irm_unbind_process(pid, name.encode()), + default=BindError) + + +def create_name(info: NameInfo) -> None: + """ + Create a name in the IRM. + + :param info: :class:`NameInfo` describing the name to create. + :raises NameExistsError: If the name already exists. + """ + + _info = _name_info_to_c(info) + + raise_errno(lib.irm_create_name(_info), default=NameExistsError) + + +def destroy_name(name: str) -> None: + """ + Destroy a name in the IRM. + + :param name: The name to destroy. + :raises NameNotFoundError: If the name is unknown. + """ + + raise_errno(lib.irm_destroy_name(name.encode()), + default=NameNotFoundError) + + +def list_names() -> List[NameInfo]: + """ + List all registered names. + + :return: List of :class:`NameInfo` objects. + :raises IrmError: If the names could not be listed. + """ + + _names = ffi.new("struct name_info **") + n = raise_errno(lib.irm_list_names(_names), default=IrmError) + + result = [] + + for i in range(n): + info = _names[0][i] + + result.append(NameInfo( + name=ffi.string(info.name).decode(), + pol_lb=LoadBalancePolicy(info.pol_lb), + server_sec=NameSecPaths( + sec=ffi.string(info.s.sec).decode(), + key=ffi.string(info.s.key).decode(), + crt=ffi.string(info.s.crt).decode(), + ), + client_sec=NameSecPaths( + sec=ffi.string(info.c.sec).decode(), + key=ffi.string(info.c.key).decode(), + crt=ffi.string(info.c.crt).decode(), + ), + )) + + if n > 0: + lib.free(_names[0]) + + return result + + +def reg_name(name: str, pid: int) -> None: + """ + Register an IPCP to a name. + + :param name: The name to register. + :param pid: PID of the IPCP to register. + :raises NameNotFoundError: If the name is unknown. + """ + + raise_errno(lib.irm_reg_name(name.encode(), pid), + default=NameNotFoundError) + + +def unreg_name(name: str, pid: int) -> None: + """ + Unregister an IPCP from a name. + + :param name: The name to unregister. + :param pid: PID of the IPCP to unregister. + :raises NameNotFoundError: If the name or the registration + is unknown. + """ + + raise_errno(lib.irm_unreg_name(name.encode(), pid), + default=NameNotFoundError) diff --git a/ouroboros/qos.py b/ouroboros/qos.py index f437ee2..c42a62e 100644 --- a/ouroboros/qos.py +++ b/ouroboros/qos.py @@ -1,110 +1,93 @@ # -# Ouroboros - Copyright (C) 2016 - 2020 -# -# Python API for applications - QoS -# -# 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 # -from _ouroboros_cffi import ffi -from math import modf -from typing import Optional +""" +QoS specification for Ouroboros flows. +""" + +from __future__ import annotations -# Some constants -MILLION = 1000 * 1000 -BILLION = 1000 * 1000 * 1000 +from dataclasses import dataclass +from enum import IntEnum + +DEFAULT_PEER_TIMEOUT = 120000 +UINT32_MAX = 0xFFFFFFFF +UINT64_MAX = 0xFFFFFFFFFFFFFFFF + + +class QoSService(IntEnum): + """ + Mirrors ``enum qos_service`` in ``ouroboros/qos.h``. + """ + RAW = 0 # No FRCT; best-effort raw messages + MESSAGE = 1 # FRCT, reliable ordered messages + STREAM = 2 # FRCT, reliable ordered byte stream + +@dataclass(frozen=True) class QoSSpec: """ - delay: In ms, default 1000s - bandwidth: In bits / s, default 0 - availability: Class of 9s, default 0 - loss: Packet loss in ppm, default MILLION - ber: Bit error rate, errors per billion bits. default BILLION - in_order: In-order delivery, enables FRCT, default 0 - max_gap: Maximum interruption in ms, default MILLION - cypher_s: Requested encryption strength in bits + QoS specification for a flow. + + :ivar service: :class:`QoSService` (gates FRCT when > 0). + :ivar delay: Maximum one-way delay in ms. + :ivar bandwidth: Required bandwidth in bits/s. + :ivar availability: Class of 9s (number of nines of uptime). + :ivar loss: Maximum packet loss (per million). + :ivar ber: Maximum bit error rate (errors per billion). + :ivar max_gap: Maximum interruption in ms. + :ivar timeout: Peer timeout in ms (default 120000 ms). """ - def __init__(self, - delay: int = MILLION, - bandwidth: int = 0, - availability: int = 0, - loss: int = 1, - ber: int = MILLION, - in_order: int = 0, - max_gap: int = MILLION, - cypher_s: int = 0): - self.delay = delay - self.bandwidth = bandwidth - self.availability = availability - self.loss = loss - self.ber = ber - self.in_order = in_order - self.max_gap = max_gap - self.cypher_s = cypher_s - - -def _fl_to_timespec(timeo: float): - if timeo is None: - return ffi.NULL - elif timeo <= 0: - return ffi.new("struct timespec *", [0, 0]) - else: - frac, whole = modf(timeo) - _timeo = ffi.new("struct timespec *") - _timeo.tv_sec = whole - _timeo.tv_nsec = frac * BILLION - return _timeo - - -def _timespec_to_fl(_timeo) -> Optional[float]: - if _timeo is ffi.NULL: - return None - elif _timeo.tv_sec <= 0 and _timeo.tv_nsec == 0: - return 0 - else: - return _timeo.tv_sec + _timeo.tv_nsec / BILLION - - -def _qos_to_qosspec(qos: QoSSpec): - if qos is None: - return ffi.NULL - else: - return ffi.new("qosspec_t *", - [qos.delay, - qos.bandwidth, - qos.availability, - qos.loss, - qos.ber, - qos.in_order, - qos.max_gap, - qos.cypher_s]) - - -def _qosspec_to_qos(_qos) -> Optional[QoSSpec]: - if _qos is ffi.NULL: - return None - else: - return QoSSpec(delay=_qos.delay, - bandwidth=_qos.bandwidth, - availability=_qos.availability, - loss=_qos.loss, - ber=_qos.ber, - in_order=_qos.in_order, - max_gap=_qos.max_gap, - cypher_s=_qos.cypher_s) + service: QoSService = QoSService.RAW + delay: int = UINT32_MAX + bandwidth: int = 0 + availability: int = 0 + loss: int = 1 + ber: int = 1 + max_gap: int = UINT32_MAX + timeout: int = DEFAULT_PEER_TIMEOUT + + +# Predefined QoS specs, mirroring the static const qosspec_t values in +# ouroboros/qos.h. "_safe" sets ber=0 (integrity check); "rt" trades +# reliability for latency. + +QOS_RAW = QoSSpec( + service=QoSService.RAW, + delay=UINT32_MAX, bandwidth=0, availability=0, + loss=1, ber=1, + max_gap=UINT32_MAX, timeout=0) + +QOS_RAW_SAFE = QoSSpec( + service=QoSService.RAW, + delay=UINT32_MAX, bandwidth=0, availability=0, + loss=1, ber=0, + max_gap=UINT32_MAX, timeout=0) + +QOS_RT = QoSSpec( + service=QoSService.MESSAGE, + delay=100, bandwidth=UINT64_MAX, availability=3, + loss=1, ber=1, + max_gap=100, timeout=DEFAULT_PEER_TIMEOUT) + +QOS_RT_SAFE = QoSSpec( + service=QoSService.MESSAGE, + delay=100, bandwidth=UINT64_MAX, availability=3, + loss=1, ber=0, + max_gap=100, timeout=DEFAULT_PEER_TIMEOUT) + +QOS_MSG = QoSSpec( + service=QoSService.MESSAGE, + delay=1000, bandwidth=0, availability=0, + loss=0, ber=0, + max_gap=2000, timeout=DEFAULT_PEER_TIMEOUT) + +QOS_STREAM = QoSSpec( + service=QoSService.STREAM, + delay=1000, bandwidth=0, availability=0, + loss=0, ber=0, + max_gap=2000, timeout=DEFAULT_PEER_TIMEOUT) |