<commit_before>"""
Deputy daemon that provides a service via DBus for performing privileged
operations.

Some operations, such as generating configuration files, sending signals to
other processes etc. need certain privileges. The Deputy service runs as *root*
and provides a very simple service over DBus.
"""
import contextlib
import importlib.resources
import io
import logging
import os
import pathlib
import pwd
import re
import signal
import stat
import string
import subprocess
import tempfile
import textwrap
from functools import partial
from typing import Iterable, Optional, Tuple, Union, overload, Iterator, List

import netaddr
from gi.repository import GLib
from netaddr import EUI, IPAddress
from pydbus import SystemBus
from pydbus.bus import Bus
from sqlalchemy import null
from sqlalchemy.pool import StaticPool

from hades import constants
from hades.common import db
from hades.common.db import (
    auth_dhcp_lease,
    get_dhcp_lease_of_ip,
    unauth_dhcp_lease,
    ObjectsDiff, LeaseInfo,
)
from hades.common.glib import typed_glib_error
from hades.common.privileges import dropped_privileges
from hades.common.signals import install_handler
from hades.config import Config, get_config
from hades.deputy.dhcp import release_dhcp_lease

logger = logging.getLogger(__name__)


def reload_systemd_unit(bus: Bus, unit: str, timeout: int = 100) -> None:
    """Instruct systemd to reload a given unit.

    :param bus: A DBus Bus
    :param unit: The name of the systemd unit
    :param timeout: Timeout in milliseconds
    """
    logger.debug("Instructing systemd to reload unit %s", unit)
    with typed_glib_error():
        systemd = bus.get('org.freedesktop.systemd1', timeout=timeout)
        manager_interface = systemd['org.freedesktop.systemd1.Manager']
        manager_interface.ReloadUnit(unit, 'fail', timeout=timeout)


def restart_systemd_unit(bus: Bus, unit: str, timeout: int = 100) -> None:
    """Instruct systemd to restart a given unit.

    :param bus: A DBus Bus
    :param unit: The name of the systemd unit
    :param timeout: Timeout in milliseconds
    """
    logger.debug("Instructing systemd to restart unit %s", unit)
    with typed_glib_error():
        systemd = bus.get('org.freedesktop.systemd1', timeout=timeout)
        manager_interface = systemd['org.freedesktop.systemd1.Manager']
        manager_interface.RestartUnit(unit, 'fail', timeout=timeout)


def generate_dhcp_host_reservations(
    hosts: Iterable[Tuple[netaddr.EUI, netaddr.IPAddress, Optional[str]]],
) -> Iterable[str]:
    """Generate lines suitable for dnsmasq's ``--dhcp-hostsfile=`` option.

    :param hosts: The MAC address-IP address pairs of the hosts
    """
    for mac, ip, hostname in hosts:
        mac = netaddr.EUI(mac)
        mac.dialect = netaddr.mac_unix_expanded
        if hostname is not None:
            yield "{0},id:*,{1},{2}\n".format(mac, ip, hostname)
        else:
            yield "{0},id:*,{1}\n".format(mac, ip)


@overload
def replace_file(
    path: Union[os.PathLike, str],
    content: Union[Iterable[bytes], bytes],
    *,
    encoding: None = ...,
    owner: Optional[int] = ...,
    group: Optional[int] = ...,
    mode: Optional[int] = ...,
) -> None: ...


@overload
def replace_file(
    path: Union[os.PathLike, str],
    content: Union[Iterable[str], str],
    *,
    encoding: str = ...,
    owner: Optional[int] = ...,
    group: Optional[int] = ...,
    mode: Optional[int] = ...,
) -> None: ...


def replace_file(
    path: Union[os.PathLike, str],
    content: Union[Iterable[Union[bytes, str]], Union[bytes, str]],
    *,
    encoding: Optional[str] = None,
    owner: Optional[int] = None,
    group: Optional[int] = None,
    mode: Optional[int] = None,
) -> None:
    """
    Atomically replace a file with the given content.

    The directory of the file must exist and must be writeable. The content may
    either be a str or bytes object or an Iterable of such objects, in which
    case the content will be written via :func:`io.IO.writelines`.

    :param path: Path to the file
    :param content: The new content of the file
    :param encoding: The encoding for :class:`str` content
    :param owner: File owner
    :param group: File group
    :param mode: File mode
    :raises OSError: if file system operations fail
    """
    open_mode = "wb" if encoding is None else "w"
    path = pathlib.Path(path)
    parent = path.parent
    if encoding is None and isinstance(content, str):
        raise ValueError("encoding required for writing str content")
    with tempfile.NamedTemporaryFile(
        open_mode, encoding=encoding, dir=parent, delete=False
    ) as file:
        dir_fd = None
        try:
            dir_fd = os.open(
                parent, os.O_DIRECTORY | os.O_CLOEXEC | os.O_RDONLY, 0
            )
            fd = file.fileno()
            if (owner is None) ^ (group is None):
                stat_result = os.fstat(fd)
                if owner is None:
                    owner = stat_result.st_uid
                if group is None:
                    group = stat_result.st_gid
            if owner is not None:
                os.fchown(fd, owner, group)
            if mode is not None:
                os.fchmod(fd, mode)
            if isinstance(content, (bytes, str)):
                file.write(content)
            else:
                file.writelines(content)
            file.flush()
            os.fsync(fd)
            os.rename(file.name, path)
            os.fsync(dir_fd)
        except BaseException:
            os.unlink(file)
            raise
        finally:
            if dir_fd is not None:
                os.close(dir_fd)


def generate_auth_dhcp_hosts_file(
    hosts: Iterable[Tuple[netaddr.EUI, netaddr.IPAddress, Optional[str]]],
) -> None:
    """Generate the dnsmasq hosts file for authenticated users"""
    file = pathlib.Path(constants.AUTH_DHCP_HOSTS_FILE)
    logger.info("Generating DHCP hosts file %s", file)
    # Use hades-auth-dhcp as owner not as root
    auth_dhcp_pwd = pwd.<PASSWORD>(constants.AUTH_DHCP_USER)
    try:
        replace_file(
            file,
            generate_dhcp_host_reservations(hosts),<commit_msg>Expand docstring of `generate_auth_dhcp_hosts_file`<commit_after>"""
Deputy daemon that provides a service via DBus for performing privileged
operations.

Some operations, such as generating configuration files, sending signals to
other processes etc. need certain privileges. The Deputy service runs as *root*
and provides a very simple service over DBus.
"""
import contextlib
import importlib.resources
import io
import logging
import os
import pathlib
import pwd
import re
import signal
import stat
import string
import subprocess
import tempfile
import textwrap
from functools import partial
from typing import Iterable, Optional, Tuple, Union, overload, Iterator, List

import netaddr
from gi.repository import GLib
from netaddr import EUI, IPAddress
from pydbus import SystemBus
from pydbus.bus import Bus
from sqlalchemy import null
from sqlalchemy.pool import StaticPool

from hades import constants
from hades.common import db
from hades.common.db import (
    auth_dhcp_lease,
    get_dhcp_lease_of_ip,
    unauth_dhcp_lease,
    ObjectsDiff, LeaseInfo,
)
from hades.common.glib import typed_glib_error
from hades.common.privileges import dropped_privileges
from hades.common.signals import install_handler
from hades.config import Config, get_config
from hades.deputy.dhcp import release_dhcp_lease

logger = logging.getLogger(__name__)


def reload_systemd_unit(bus: Bus, unit: str, timeout: int = 100) -> None:
    """Instruct systemd to reload a given unit.

    :param bus: A DBus Bus
    :param unit: The name of the systemd unit
    :param timeout: Timeout in milliseconds
    """
    logger.debug("Instructing systemd to reload unit %s", unit)
    with typed_glib_error():
        systemd = bus.get('org.freedesktop.systemd1', timeout=timeout)
        manager_interface = systemd['org.freedesktop.systemd1.Manager']
        manager_interface.ReloadUnit(unit, 'fail', timeout=timeout)


def restart_systemd_unit(bus: Bus, unit: str, timeout: int = 100) -> None:
    """Instruct systemd to restart a given unit.

    :param bus: A DBus Bus
    :param unit: The name of the systemd unit
    :param timeout: Timeout in milliseconds
    """
    logger.debug("Instructing systemd to restart unit %s", unit)
    with typed_glib_error():
        systemd = bus.get('org.freedesktop.systemd1', timeout=timeout)
        manager_interface = systemd['org.freedesktop.systemd1.Manager']
        manager_interface.RestartUnit(unit, 'fail', timeout=timeout)


def generate_dhcp_host_reservations(
    hosts: Iterable[Tuple[netaddr.EUI, netaddr.IPAddress, Optional[str]]],
) -> Iterable[str]:
    """Generate lines suitable for dnsmasq's ``--dhcp-hostsfile=`` option.

    :param hosts: The MAC address-IP address pairs of the hosts
    """
    for mac, ip, hostname in hosts:
        mac = netaddr.EUI(mac)
        mac.dialect = netaddr.mac_unix_expanded
        if hostname is not None:
            yield "{0},id:*,{1},{2}\n".format(mac, ip, hostname)
        else:
            yield "{0},id:*,{1}\n".format(mac, ip)


@overload
def replace_file(
    path: Union[os.PathLike, str],
    content: Union[Iterable[bytes], bytes],
    *,
    encoding: None = ...,
    owner: Optional[int] = ...,
    group: Optional[int] = ...,
    mode: Optional[int] = ...,
) -> None: ...


@overload
def replace_file(
    path: Union[os.PathLike, str],
    content: Union[Iterable[str], str],
    *,
    encoding: str = ...,
    owner: Optional[int] = ...,
    group: Optional[int] = ...,
    mode: Optional[int] = ...,
) -> None: ...


def replace_file(
    path: Union[os.PathLike, str],
    content: Union[Iterable[Union[bytes, str]], Union[bytes, str]],
    *,
    encoding: Optional[str] = None,
    owner: Optional[int] = None,
    group: Optional[int] = None,
    mode: Optional[int] = None,
) -> None:
    """
    Atomically replace a file with the given content.

    The directory of the file must exist and must be writeable. The content may
    either be a str or bytes object or an Iterable of such objects, in which
    case the content will be written via :func:`io.IO.writelines`.

    :param path: Path to the file
    :param content: The new content of the file
    :param encoding: The encoding for :class:`str` content
    :param owner: File owner
    :param group: File group
    :param mode: File mode
    :raises OSError: if file system operations fail
    """
    open_mode = "wb" if encoding is None else "w"
    path = pathlib.Path(path)
    parent = path.parent
    if encoding is None and isinstance(content, str):
        raise ValueError("encoding required for writing str content")
    with tempfile.NamedTemporaryFile(
        open_mode, encoding=encoding, dir=parent, delete=False
    ) as file:
        dir_fd = None
        try:
            dir_fd = os.open(
                parent, os.O_DIRECTORY | os.O_CLOEXEC | os.O_RDONLY, 0
            )
            fd = file.fileno()
            if (owner is None) ^ (group is None):
                stat_result = os.fstat(fd)
                if owner is None:
                    owner = stat_result.st_uid
                if group is None:
                    group = stat_result.st_gid
            if owner is not None:
                os.fchown(fd, owner, group)
            if mode is not None:
                os.fchmod(fd, mode)
            if isinstance(content, (bytes, str)):
                file.write(content)
            else:
                file.writelines(content)
            file.flush()
            os.fsync(fd)
            os.rename(file.name, path)
            os.fsync(dir_fd)
        except BaseException:
            os.unlink(file)
            raise
        finally:
            if dir_fd is not None:
                os.close(dir_fd)


def generate_auth_dhcp_hosts_file(
    hosts: Iterable[Tuple[netaddr.EUI, netaddr.IPAddress, Optional[str]]],
) -> None:
    """Generate the dnsmasq hosts file for authenticated users.

    This file is passed toh the dnsmasq via the ``--dhcp-hostsfile`` option.
    The lines are generated by :func:`generate_dhcp_host_reservations`.
    """
    file = pathlib.Path(constants.AUTH_DHCP_HOSTS_FILE)
    logger.info("Generating DHCP hosts file %s", file)
    # Use hades-auth-dhcp as owner not as root
    auth_dhcp_pwd = <PASSWORD>(constants.AUTH_DHCP_USER)
    try:
        replace_file(
            file,
            generate_dhcp_host_reservations(hosts),
{'fraction_non_alphanumeric': 0.09743875278396437, 'fraction_numerical': 0.002545338848234171, 'mean_word_length': 2.798489425981873, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1974750', 'n_tokens_mistral': 3870, 'n_tokens_neox': 3740, 'n_words': 1189}