feat: add ssh module and net/disk/shell helpers

* ssh.py: new module to run commands and copy files over SSH
  (build_options, target, run, scp, rsync)
* net.py: cidr_to_hosts() and get_subnet_hosts() enumerate the usable
  IPv4/IPv6 hosts of a network or interface, with a configurable size cap
* disk.py: copy_file(), copy_dir(), rm_dir(), make_temp_dir(), mkdir()
* shell.py: which()
* add unit tests for the new helpers

Author: markuslf

CHANGELOG.md

@@ -13,11 +13,12 @@
 """
 
 __author__ = 'Linuxfabrik GmbH, Zurich/Switzerland'
-__version__ = '2026060201'
+__version__ = '2026060601'
 
 import csv
 import os
 import re
+import shutil
 import tempfile
 
 from . import shell
@@ -55,6 +56,65 @@ def bd2dmd(device):
     return mapped_device if os.path.islink(mapped_device) else ''
 
 
+def copy_dir(src, dst):
+    """
+    Recursively copy a directory tree.
+
+    Wraps `shutil.copytree()` and reports the outcome in the same
+    `(success, error)` style as the other disk helpers, so callers do not have to
+    handle exceptions themselves.
+
+    ### Parameters
+    - **src** (`str`): Source directory.
+    - **dst** (`str`): Destination directory (must not exist yet).
+
+    ### Returns
+    - **tuple**:
+        - tuple[0] (**bool**): True if the copy succeeded, otherwise False.
+        - tuple[1] (**None or str**): None on success, otherwise an error message.
+
+    ### Example
+    >>> copy_dir('/usr/share/lynis', '/tmp/lynis')
+    (True, None)
+    """
+    try:
+        shutil.copytree(src, dst)
+        return True, None
+    except (OSError, shutil.Error) as e:
+        return False, f'Error copying directory {src} to {dst}: {e}'
+    except Exception as e:
+        return False, f'Unknown error copying directory {src} to {dst}: {e}'
+
+
+def copy_file(src, dst):
+    """
+    Copy a single file, preserving its metadata.
+
+    Wraps `shutil.copy2()` and reports the outcome in the same `(success, error)`
+    style as the other disk helpers.
+
+    ### Parameters
+    - **src** (`str`): Source file.
+    - **dst** (`str`): Destination file or directory.
+
+    ### Returns
+    - **tuple**:
+        - tuple[0] (**bool**): True if the copy succeeded, otherwise False.
+        - tuple[1] (**None or str**): None on success, otherwise an error message.
+
+    ### Example
+    >>> copy_file('/etc/lynis/default.prf', '/tmp/lynis/default.prf')
+    (True, None)
+    """
+    try:
+        shutil.copy2(src, dst)
+        return True, None
+    except OSError as e:
+        return False, f'OS error "{e.strerror}" while copying {src} to {dst}'
+    except Exception as e:
+        return False, f'Unknown error copying {src} to {dst}: {e}'
+
+
 def dir_exists(path):
     """
     Check if a directory exists at the given path.
@@ -375,6 +435,67 @@ def grep_file(filename, pattern):
     return True, ''
 
 
+def make_temp_dir(prefix=''):
+    """
+    Create a unique temporary directory and return its path.
+
+    Wraps `tempfile.mkdtemp()` and reports the outcome in the same
+    `(success, result)` style as the other disk helpers.
+
+    ### Parameters
+    - **prefix** (`str`, optional): Prefix for the directory name. Defaults to ''.
+
+    ### Returns
+    - **tuple**:
+        - tuple[0] (**bool**): True on success, otherwise False.
+        - tuple[1] (**str**): The created directory path on success, otherwise an
+          error message.
+
+    ### Example
+    >>> make_temp_dir(prefix='myapp-')
+    (True, '/tmp/myapp-abcd1234')
+    """
+    try:
+        return True, tempfile.mkdtemp(prefix=prefix)
+    except OSError as e:
+        return False, f'OS error "{e.strerror}" while creating a temporary directory'
+    except Exception as e:
+        return False, f'Unknown error creating a temporary directory: {e}'
+
+
+def mkdir(path, mode=0o755, exist_ok=True):
+    """
+    Create a directory, including any missing parent directories.
+
+    Wraps `os.makedirs()` and reports the outcome in the same `(success, error)`
+    style as the other disk helpers, so callers do not have to handle exceptions
+    themselves.
+
+    ### Parameters
+    - **path** (`str`): Directory path to create.
+    - **mode** (`int`, optional): Permission bits for newly created directories.
+      Defaults to `0o755`.
+    - **exist_ok** (`bool`, optional): If `True`, an already existing directory is
+      not an error. Defaults to `True`.
+
+    ### Returns
+    - **tuple**:
+        - tuple[0] (**bool**): True if the directory exists afterwards, else False.
+        - tuple[1] (**None or str**): None on success, otherwise an error message.
+
+    ### Example
+    >>> mkdir('/tmp/example/sub')
+    (True, None)
+    """
+    try:
+        os.makedirs(path, mode=mode, exist_ok=exist_ok)
+        return True, None
+    except OSError as e:
+        return False, f'OS error "{e.strerror}" while creating {path}'
+    except Exception as e:
+        return False, f'Unknown error creating {path}: {e}'
+
+
 def read_csv(
     filename,
     delimiter=',',
@@ -517,6 +638,34 @@ def read_file(filename):
         return False, f'Unknown error opening or reading {filename}: {e}'
 
 
+def rm_dir(path):
+    """
+    Recursively delete a directory tree.
+
+    Wraps `shutil.rmtree()` and reports the outcome in the same `(success, error)`
+    style as `rm_file()`.
+
+    ### Parameters
+    - **path** (`str`): Directory tree to delete.
+
+    ### Returns
+    - **tuple**:
+        - tuple[0] (**bool**): True if deletion succeeded, otherwise False.
+        - tuple[1] (**None or str**): None on success, otherwise an error message.
+
+    ### Example
+    >>> rm_dir('/tmp/lynis')
+    (True, None)
+    """
+    try:
+        shutil.rmtree(path)
+        return True, None
+    except OSError as e:
+        return False, f'OS error "{e.strerror}" while deleting {path}'
+    except Exception as e:
+        return False, f'Unknown error deleting {path}: {e}'
+
+
 def rm_file(filename):
     """
     Delete or remove a file.

disk.py

@@ -12,8 +12,9 @@
 """Provides network related functions and variables."""
 
 __author__ = 'Linuxfabrik GmbH, Zurich/Switzerland'
-__version__ = '2026050901'
+__version__ = '2026060601'
 
+import ipaddress
 import random
 import re
 import socket
@@ -128,7 +129,11 @@
 
 
 def _socket_fetch(
-    open_socket_func, connect_args, payload=None, dialog=None, timeout=3,
+    open_socket_func,
+    connect_args,
+    payload=None,
+    dialog=None,
+    timeout=3,
     socket_name='socket',
 ):
     """
@@ -192,7 +197,10 @@ def _socket_fetch(
                         try:
                             s.sendall(send_bytes)
                         except Exception as e:
-                            return False, f'Could not send payload on {socket_name}: {e}'
+                            return (
+                                False,
+                                f'Could not send payload on {socket_name}: {e}',
+                            )
 
                     if expect is None:
                         results.append('')
@@ -293,7 +301,8 @@ def fetch(host, port, msg=None, dialog=None, timeout=3, ipv6=False, tls=False):
     ### Example
     >>> success, response = fetch('example.com', 80)
     >>> ok, [hello, vars_block, _] = fetch(
-    ...     '127.0.0.1', 3493,
+    ...     '127.0.0.1',
+    ...     3493,
     ...     dialog=[
     ...         (None, r'^OK\\b|^ERR\\b'),
     ...         (b'LIST VAR myups\\n', r'END LIST VAR myups'),
@@ -548,6 +557,110 @@ def get_public_ip(services, insecure=False, no_proxy=False, timeout=2):
     return False, None
 
 
+def cidr_to_hosts(cidr, max_hosts=65536):
+    """
+    Return the usable host IP addresses of a network in CIDR notation.
+
+    Enumerates the usable host addresses of an IPv4 or IPv6 network given as a
+    CIDR string (e.g. `10.1.1.0/24` or `2001:db8::/120`). The network and
+    broadcast address are excluded. Host bits set in the input are tolerated
+    (`strict=False`).
+
+    ### Parameters
+    - **cidr** (`str`): Network in CIDR notation, e.g. `10.1.1.0/24`.
+    - **max_hosts** (`int`, optional): Refuse to enumerate networks with more
+      addresses than this, which protects against accidentally expanding a large
+      range such as an IPv6 `/64` (2^64 addresses). Set to `None` to disable the
+      limit. Defaults to `65536` (an IPv4 `/16`).
+
+    ### Returns
+    - **tuple** (`bool`, `list` or `str`):
+      - `True` and the list of host IP address strings (IPv4 or IPv6) on success.
+      - `False` and an error message on failure.
+
+    ### Example
+    >>> cidr_to_hosts('10.1.1.0/30')
+    (True, ['10.1.1.1', '10.1.1.2'])
+    >>> cidr_to_hosts('2001:db8::/126')
+    (True, ['2001:db8::1', '2001:db8::2', '2001:db8::3'])
+    """
+    try:
+        network = ipaddress.ip_network(cidr, strict=False)
+    except ValueError as e:
+        return (False, f'Invalid network "{cidr}": {e}')
+    if max_hosts is not None and network.num_addresses > max_hosts:
+        return (
+            False,
+            f'Network "{cidr}" is too large to enumerate '
+            f'({network.num_addresses} addresses, limit {max_hosts}).',
+        )
+    return (True, [str(host) for host in network.hosts()])
+
+
+def get_subnet_hosts(interface=None, max_hosts=65536):
+    """
+    Return the usable host IP addresses of an interface's subnet.
+
+    Resolves the address and netmask of the given interface (or the default
+    interface, the one carrying the default route, if none is given), derives the
+    subnet and enumerates its usable host addresses (network and broadcast
+    address excluded). IPv4 is preferred; if an interface has no IPv4 address its
+    IPv6 subnet is used. Note that an IPv6 `/64` exceeds `max_hosts` and is
+    therefore reported as too large rather than enumerated.
+
+    ### Parameters
+    - **interface** (`str`, optional): Network interface name (e.g. `eth0`). If
+      `None`, the default interface is used.
+    - **max_hosts** (`int`, optional): Passed through to `cidr_to_hosts()` to cap
+      enumeration. Defaults to `65536`.
+
+    ### Returns
+    - **tuple** (`bool`, `list` or `str`):
+      - `True` and the list of host IP address strings on success.
+      - `False` and an error message on failure.
+
+    ### Notes
+    - Requires the `netifaces` library.
+
+    ### Example
+    >>> get_subnet_hosts('eth0')
+    (True, ['192.168.1.1', '192.168.1.2', ..., '192.168.1.254'])
+    """
+    if not HAVE_NETIFACES:
+        return (False, 'Python module "netifaces" is not installed.')
+
+    if interface is None:
+        netinfo = get_netinfo()
+        if not netinfo or not netinfo.get('address'):
+            return (False, 'Unable to determine the default interface subnet.')
+        return cidr_to_hosts(
+            f'{netinfo["address"]}/{netinfo["mask_cidr"]}', max_hosts=max_hosts
+        )
+
+    try:
+        iface_addrs = netifaces.ifaddresses(interface)
+    except ValueError:
+        return (False, f'No such interface "{interface}".')
+
+    # IPv4 first, then IPv6. netifaces stores the IPv6 netmask as
+    # "ffff:ffff:ffff:ffff::/64" and may suffix the address with a "%scope".
+    for family in (netifaces.AF_INET, netifaces.AF_INET6):
+        entries = iface_addrs.get(family)
+        if not entries:
+            continue
+        address = (entries[0].get('addr') or '').split('%')[0]
+        netmask = entries[0].get('netmask') or ''
+        if not address:
+            continue
+        if family == netifaces.AF_INET:
+            cidr = ip_to_cidr(netmask)
+        else:
+            cidr = netmask.split('/')[-1] if '/' in netmask else '128'
+        return cidr_to_hosts(f'{address}/{cidr}', max_hosts=max_hosts)
+
+    return (False, f'No IPv4 or IPv6 address found on interface "{interface}".')
+
+
 def ip_to_cidr(ip):
     """
     Convert an IPv4 netmask to CIDR notation.

net.py

@@ -11,12 +11,13 @@
 """Communicates with the Shell on Linux and Windows."""
 
 __author__ = 'Linuxfabrik GmbH, Zurich/Switzerland'
-__version__ = '2026041201'
+__version__ = '2026060601'
 
 
 import os
 import re
 import shlex
+import shutil
 import subprocess  # nosec B404 - this library is the subprocess helper
 
 from . import txt
@@ -211,3 +212,24 @@ def shell_exec(
         return False, f'Timeout after {timeout} seconds.'
 
     return True, (txt.to_text(stdout), txt.to_text(stderr), p.returncode)
+
+
+def which(name):
+    """
+    Locate an executable in the system PATH, like the `which` command.
+
+    Thin wrapper around `shutil.which()` so callers do not need to import it
+    directly and the lookup stays consistent across consumers.
+
+    ### Parameters
+    - **name** (`str`): Program name to look for (e.g. `lynis`).
+
+    ### Returns
+    - **str or None**: The absolute path to the executable, or `None` if it is
+      not found in PATH.
+
+    ### Example
+    >>> which('sh')
+    '/usr/bin/sh'
+    """
+    return shutil.which(name)