path: root/dts/framework/utils.py

# SPDX-License-Identifier: BSD-3-Clause
# Copyright(c) 2010-2014 Intel Corporation
# Copyright(c) 2022-2023 PANTHEON.tech s.r.o.
# Copyright(c) 2022-2023 University of New Hampshire
# Copyright(c) 2024 Arm Limited

"""Various utility classes and functions.

These are used in multiple modules across the framework. They're here because
they provide some non-specific functionality, greatly simplify imports or just don't
fit elsewhere.

Attributes:
    REGEX_FOR_PCI_ADDRESS: The regex representing a PCI address, e.g. ``0000:00:08.0``.
"""

import fnmatch
import json
import os
import random
import tarfile
from enum import Enum, Flag
from pathlib import Path
from typing import Any, Callable

from scapy.layers.inet import IP, TCP, UDP, Ether  # type: ignore[import-untyped]
from scapy.packet import Packet  # type: ignore[import-untyped]

from .exception import InternalError

REGEX_FOR_PCI_ADDRESS: str = r"[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}.[0-9]{1}"
_REGEX_FOR_COLON_OR_HYPHEN_SEP_MAC: str = r"(?:[\da-fA-F]{2}[:-]){5}[\da-fA-F]{2}"
_REGEX_FOR_DOT_SEP_MAC: str = r"(?:[\da-fA-F]{4}.){2}[\da-fA-F]{4}"
REGEX_FOR_MAC_ADDRESS: str = rf"{_REGEX_FOR_COLON_OR_HYPHEN_SEP_MAC}|{_REGEX_FOR_DOT_SEP_MAC}"
REGEX_FOR_BASE64_ENCODING: str = "[-a-zA-Z0-9+\\/]*={0,3}"


def expand_range(range_str: str) -> list[int]:
    """Process `range_str` into a list of integers.

    There are two possible formats of `range_str`:

        * ``n`` - a single integer,
        * ``n-m`` - a range of integers.

    The returned range includes both ``n`` and ``m``. Empty string returns an empty list.

    Args:
        range_str: The range to expand.

    Returns:
        All the numbers from the range.
    """
    expanded_range: list[int] = []
    if range_str:
        range_boundaries = range_str.split("-")
        # will throw an exception when items in range_boundaries can't be converted,
        # serving as type check
        expanded_range.extend(range(int(range_boundaries[0]), int(range_boundaries[-1]) + 1))

    return expanded_range


def get_packet_summaries(packets: list[Packet]) -> str:
    """Format a string summary from `packets`.

    Args:
        packets: The packets to format.

    Returns:
        The summary of `packets`.
    """
    if len(packets) == 1:
        packet_summaries = packets[0].summary()
    else:
        packet_summaries = json.dumps(list(map(lambda pkt: pkt.summary(), packets)), indent=4)
    return f"Packet contents: \n{packet_summaries}"


class StrEnum(Enum):
    """Enum with members stored as strings."""

    @staticmethod
    def _generate_next_value_(name: str, start: int, count: int, last_values: object) -> str:
        return name

    def __str__(self) -> str:
        """The string representation is the name of the member."""
        return self.name


class MesonArgs:
    """Aggregate the arguments needed to build DPDK."""

    _default_library: str

    def __init__(self, default_library: str | None = None, **dpdk_args: str | bool):
        """Initialize the meson arguments.

        Args:
            default_library: The default library type, Meson supports ``shared``, ``static`` and
                ``both``. Defaults to :data:`None`, in which case the argument won't be used.
            dpdk_args: The arguments found in ``meson_options.txt`` in root DPDK directory.
                Do not use ``-D`` with them.

        Example:
            ::

                meson_args = MesonArgs(enable_kmods=True).
        """
        self._default_library = f"--default-library={default_library}" if default_library else ""
        self._dpdk_args = " ".join(
            (
                f"-D{dpdk_arg_name}={dpdk_arg_value}"
                for dpdk_arg_name, dpdk_arg_value in dpdk_args.items()
            )
        )

    def __str__(self) -> str:
        """The actual args."""
        return " ".join(f"{self._default_library} {self._dpdk_args}".split())


class TarCompressionFormat(StrEnum):
    """Compression formats that tar can use.

    Enum names are the shell compression commands
    and Enum values are the associated file extensions.

    The 'none' member represents no compression, only archiving with tar.
    Its value is set to 'tar' to indicate that the file is an uncompressed tar archive.
    """

    none = "tar"
    gzip = "gz"
    compress = "Z"
    bzip2 = "bz2"
    lzip = "lz"
    lzma = "lzma"
    lzop = "lzo"
    xz = "xz"
    zstd = "zst"

    @property
    def extension(self):
        """Return the extension associated with the compression format.

        If the compression format is 'none', the extension will be in the format 'tar'.
        For other compression formats, the extension will be in the format
        'tar.{compression format}'.
        """
        return f"{self.value}" if self == self.none else f"{self.none.value}.{self.value}"


def convert_to_list_of_string(value: Any | list[Any]) -> list[str]:
    """Convert the input to the list of strings."""
    return list(map(str, value) if isinstance(value, list) else str(value))


def create_tarball(
    dir_path: Path,
    compress_format: TarCompressionFormat = TarCompressionFormat.none,
    exclude: Any | list[Any] | None = None,
) -> Path:
    """Create a tarball from the contents of the specified directory.

    This method creates a tarball containing all files and directories within `dir_path`.
    The tarball will be saved in the directory of `dir_path` and will be named based on `dir_path`.

    Args:
        dir_path: The directory path.
        compress_format: The compression format to use. Defaults to no compression.
        exclude: Patterns for files or directories to exclude from the tarball.
                These patterns are used with `fnmatch.fnmatch` to filter out files.

    Returns:
        The path to the created tarball.
    """

    def create_filter_function(exclude_patterns: str | list[str] | None) -> Callable | None:
        """Create a filter function based on the provided exclude patterns.

        Args:
            exclude_patterns: Patterns for files or directories to exclude from the tarball.
                These patterns are used with `fnmatch.fnmatch` to filter out files.

        Returns:
            The filter function that excludes files based on the patterns.
        """
        if exclude_patterns:
            exclude_patterns = convert_to_list_of_string(exclude_patterns)

            def filter_func(tarinfo: tarfile.TarInfo) -> tarfile.TarInfo | None:
                file_name = os.path.basename(tarinfo.name)
                if any(fnmatch.fnmatch(file_name, pattern) for pattern in exclude_patterns):
                    return None
                return tarinfo

            return filter_func
        return None

    target_tarball_path = dir_path.with_suffix(f".{compress_format.extension}")
    with tarfile.open(target_tarball_path, f"w:{compress_format.value}") as tar:
        tar.add(dir_path, arcname=dir_path.name, filter=create_filter_function(exclude))

    return target_tarball_path


def extract_tarball(tar_path: str | Path):
    """Extract the contents of a tarball.

    The tarball will be extracted in the same path as `tar_path` parent path.

    Args:
        tar_path: The path to the tarball file to extract.
    """
    with tarfile.open(tar_path, "r") as tar:
        tar.extractall(path=Path(tar_path).parent)


class PacketProtocols(Flag):
    """Flag specifying which protocols to use for packet generation."""

    #:
    IP = 1
    #:
    TCP = 2 | IP
    #:
    UDP = 4 | IP
    #:
    ALL = TCP | UDP


def generate_random_packets(
    number_of: int,
    payload_size: int = 1500,
    protocols: PacketProtocols = PacketProtocols.ALL,
    ports_range: range = range(1024, 49152),
    mtu: int = 1500,
) -> list[Packet]:
    """Generate a number of random packets.

    The payload of the packets will consist of random bytes. If `payload_size` is too big, then the
    maximum payload size allowed for the specific packet type is used. The size is calculated based
    on the specified `mtu`, therefore it is essential that `mtu` is set correctly to match the MTU
    of the port that will send out the generated packets.

    If `protocols` has any L4 protocol enabled then all the packets are generated with any of
    the specified L4 protocols chosen at random. If only :attr:`~PacketProtocols.IP` is set, then
    only L3 packets are generated.

    If L4 packets will be generated, then the TCP/UDP ports to be used will be chosen at random from
    `ports_range`.

    Args:
        number_of: The number of packets to generate.
        payload_size: The packet payload size to generate, capped based on `mtu`.
        protocols: The protocols to use for the generated packets.
        ports_range: The range of L4 port numbers to use. Used only if `protocols` has L4 protocols.
        mtu: The MTU of the NIC port that will send out the generated packets.

    Raises:
        InternalError: If the `payload_size` is invalid.

    Returns:
        A list containing the randomly generated packets.
    """
    if payload_size < 0:
        raise InternalError(f"An invalid payload_size of {payload_size} was given.")

    l4_factories = []
    if protocols & PacketProtocols.TCP:
        l4_factories.append(TCP)
    if protocols & PacketProtocols.UDP:
        l4_factories.append(UDP)

    def _make_packet() -> Packet:
        packet = Ether()

        if protocols & PacketProtocols.IP:
            packet /= IP()

        if len(l4_factories) > 0:
            src_port, dst_port = random.choices(ports_range, k=2)
            packet /= random.choice(l4_factories)(sport=src_port, dport=dst_port)

        max_payload_size = mtu - len(packet)
        usable_payload_size = payload_size if payload_size < max_payload_size else max_payload_size
        return packet / random.randbytes(usable_payload_size)

    return [_make_packet() for _ in range(number_of)]


class MultiInheritanceBaseClass:
    """A base class for classes utilizing multiple inheritance.

    This class enables it's subclasses to support both single and multiple inheritance by acting as
    a stopping point in the tree of calls to the constructors of superclasses. This class is able
    to exist at the end of the Method Resolution Order (MRO) so that subclasses can call
    :meth:`super.__init__` without repercussion.
    """

    def __init__(self, *args, **kwargs) -> None:
        """Call the init method of :class:`object`."""
        super().__init__()


def to_pascal_case(text: str) -> str:
    """Convert `text` from snake_case to PascalCase."""
    return "".join([seg.capitalize() for seg in text.split("_")])