# SPDX-License-Identifier: AGPL-3.0-or-later
"""App component for other apps to use handle Let's Encrypt certificates."""

import logging
import pathlib
import threading
from typing import ClassVar

from plinth import app
from plinth.modules.names.components import DomainName
from plinth.privileged import service as service_privileged

from . import privileged

logger = logging.getLogger(__name__)

_list_type = list


class LetsEncrypt(app.FollowerComponent):
    """Component to receive Let's Encrypt renewal hooks.

    Performs the following tasks:

    - Listen to certificate change events from the Let's Encrypt app. This
      could be obtain, re-obtain, renewal, revoke and delete events.

    - Filter out events after comparing domains list in certificate events to
      the list of domains the app (to which this component belongs) is
      interested in.

    - If requested, copy certificates from /etc/letsencrypt directory to app's
      directory with the needed permissions and ownership.

    - Query status of copied certificates. Status could be up-to-date,
      self-signed, outdated (/etc/letsencrypt has more recent certificate),
      expired or a test certificate.

    - Optionally restart a bunch of daemons after interesting certificate event
      occurs.

    """

    _all: ClassVar[dict[str, 'LetsEncrypt']] = {}

    def __init__(self, component_id, domains=None, daemons=None,
                 should_copy_certificates=False, private_key_path=None,
                 certificate_path=None, user_owner=None, group_owner=None,
                 managing_app=None, reload_daemons=False):
        """Initialize the Let's Encrypt component.

        component_id should be a unique ID across all components of an app and
        across all components.

        domains is a list of domains that the app is configured for. Any
        certificate operation that is not happening on these domains is ignored
        by this component.

        domains can be the string '*' in which case it means that the app is
        relevant for (or is interested in) all domains configured in the
        system.

        domains can be a callable instead of a list. In this case, the callable
        is called every time an operation needs to be checked against list of
        domains for the app. The callable must return a list of domains that
        the app is configured for.

        daemons is a list of systemd units that need to be restarted (actually
        try-restarted, i.e., only if they are already running) when a relevant
        certificate operation happens.

        should_copy_certificates indicates whether the app needs the Let's
        Encrypt certificates in the directory /etc/letsencrypt to be copied to
        a different location with different permissions to be able to use them.
        If true, private_key_path, certificate_path, user_owner, group_owner
        and managing_app properties must be specified. Otherwise, ValueError is
        raised.

        private_key_path is the string containing the file path to which a
        private key must be copied. It may contain the sub-string '{domain}'
        which will be replaced with the name of the domain for which this
        certificate is being copied. Any non-existing parent directories are
        created with 'root:root' ownership and '0o755' file permissions. The
        permissions for the file will be '0o600' and ownership will be as
        specified in parameters user_owner and group_owner. If both
        private_key_path and certificate_path refer to the same path, the
        target file is created by appending the private key and certificate.

        certificate_path is the string containing the file path to which a
        certificate must be copied. It may contain the sub-string '{domain}'
        which will be replaced with the name of the domain for which this
        certificate is being copied. Any non-existing parent directories are
        created with 'root:root' ownership and '0o755' file permissions. The
        permissions for the file will be '0o644' and ownership will be as
        specified in parameters user_owner and group_owner. If both
        private_key_path and certificate_path refer to the same path, the
        target file is created by appending the private key and certificate.

        user_owner specifies the user who should own the copied certificates.
        This is typically the unprivileged user account that runs the daemon.

        group_owner specifies the group who should own the copied certificates.
        This is typically the unprivileged group account that runs the daemon.

        managing_app is the name of the app that contains this component. This
        is used by the superuser action to verify that this app is allowed to
        copy certificates to the target location. The app must then define a
        property 'managing_paths' which contains a list of pathlib.Path()
        objects to which the app is allowed to write certificates and other
        files to.

        reload_daemons is boolean to indicate if the daemons should be reloaded
        instead of restarted (which is default).
        """
        if should_copy_certificates:
            if (not private_key_path or not certificate_path or not user_owner
                    or not group_owner or not managing_app):
                raise ValueError('Not enough values for copying certificates')

        super().__init__(component_id)
        self._domains = domains
        self.daemons = daemons
        self.reload_daemons = reload_daemons
        self.should_copy_certificates = should_copy_certificates
        self.private_key_path = private_key_path
        self.certificate_path = certificate_path
        self.user_owner = user_owner
        self.group_owner = group_owner
        self.managing_app = managing_app

        self._all[component_id] = self

    @property
    def domains(self) -> list[str] | str:
        """Return a list of domains this component's app is interested in."""
        if callable(self._domains):
            return self._domains()

        return self._domains

    @classmethod
    def list(cls):
        """Return a list of all Let's Encrypt components."""
        return cls._all.values()

    def setup_certificates(
            self, app_domains: str | _list_type[str] | None = None) -> None:
        """Setup app certificates for all interested domains.

        For every domain, a certificate is copied. If a valid certificate is
        not available in Let's Encrypt, a self-signed snakeoil certificate is
        used. Each daemon is restarted if it is running.

        app_domains is the list of domains for which certificates must be
        copied. If it is not provided, the component's list of domains (which
        may be acquired by a callable) is used.
        """
        if not app_domains:
            app_domains = self.domains

        if app_domains == '*':
            # Setup for all domains and not just ones that LE can obtain
            # certificate for. This allows the domains that can't have LE
            # certificate to work with self-signed certificates.
            app_domains = DomainName.list_names()

        domains, status = self._get_letsencrypt_domains()

        if self.should_copy_certificates:
            for domain in app_domains:
                if domain in domains:
                    lineage = status['domains'][domain]['lineage']
                    self._copy_letsencrypt_certificates([domain], lineage)
                else:
                    self._copy_self_signed_certificates([domain])

        for daemon in self.daemons:
            if self.reload_daemons:
                service_privileged.try_reload_or_restart(daemon)
            else:
                service_privileged.try_restart(daemon)

    def get_status(self):
        """Return the status of certificates for all interested domains.

        A dictionary is returned containing a key for each domain and the value
        shall be a simple string with state of certificate for this app.
        Possible values are 'outdated-copy', 'self-signed', 'unknown', 'valid',
        'revoked', 'expired' and 'test_cert'.

        """
        app_domains = self.domains
        domains, le_status = self._get_letsencrypt_domains()

        final_status = {}
        for domain in app_domains:
            if domain in domains:
                status = le_status['domains'][domain]['validity']
                if self.should_copy_certificates:
                    if not self._compare_certificate(
                            domain, le_status['domains'][domain]['lineage']):
                        status = 'outdated-copy'
            else:
                status = 'self-signed'

            final_status[domain] = status

        return final_status

    def on_certificate_obtained(self, domains, lineage):
        """Handle event when a certificate is obtained.

        Filter any domains that are interesting for the app that this component
        is part of. For all interesting domains, copy the certificates if
        requested and then restart the daemons.

        """
        interested_domains = self._get_interested_domains(domains)
        if not interested_domains:
            return

        if self.should_copy_certificates:
            self._copy_letsencrypt_certificates(interested_domains, lineage)

        for daemon in self.daemons:
            if self.reload_daemons:
                service_privileged.try_reload_or_restart(daemon)
            else:
                service_privileged.try_restart(daemon)

    def on_certificate_renewed(self, domains, lineage):
        """Handle event when a certificate is renewed.

        This is typically called when a Cron job triggers certbot renewal
        process which in turn triggers configured renewal hooks after
        successful renewal. It is also called when a valid certificate is
        re-obtained.

        To trigger a renewal event for testing purposes, run `certbot renew
        --force-renewal`.

        For now, performs the same operations as obtaining a fresh certificate.

        """
        return self.on_certificate_obtained(domains, lineage)

    def on_certificate_revoked(self, domains, lineage):
        """Handle event when a certificate is revoked.

        Filter any domains that are interesting for the app that this component
        is part of. For all interesting domains, copy self-signed snakeoil
        certificates if requested and then restart the daemons.

        """
        interested_domains = self._get_interested_domains(domains)
        if not interested_domains:
            return

        if self.should_copy_certificates:
            self._copy_self_signed_certificates(interested_domains)

        for daemon in self.daemons:
            if self.reload_daemons:
                service_privileged.try_reload_or_restart(daemon)
            else:
                service_privileged.try_restart(daemon)

    def on_certificate_deleted(self, domains, lineage):
        """Handle event when a certificate is deleted.

        For now, performs the same operations as obtaining a fresh certificate.

        """
        return self.on_certificate_revoked(domains, lineage)

    @staticmethod
    def _get_letsencrypt_domains():
        """Return the list of domains with LE certificates."""
        from plinth.modules import letsencrypt
        status = letsencrypt.get_status()
        domains = {
            domain
            for domain, domain_status in status['domains'].items()
            if domain_status
        }
        return domains, status

    def _get_interested_domains(self, domains):
        """Return set of domains the component is interested in."""
        app_domains = self.domains
        if not app_domains:
            return set()

        if app_domains == '*':
            return set(domains)

        return set(domains).intersection(set(app_domains))

    def _copy_self_signed_certificates(self, domains):
        """Copy a self-signed certificate for all domains.

        Copy Apache's snake-oil certificate into daemon's directory. The
        self-signed certificate may not really work. It is merely to prevent
        the server from failing to startup because the files are missing.

        """
        source_private_key_path = '/etc/ssl/private/ssl-cert-snakeoil.key'
        source_certificate_path = '/etc/ssl/certs/ssl-cert-snakeoil.pem'
        self._copy_certificates(domains, source_private_key_path,
                                source_certificate_path)

    def _copy_letsencrypt_certificates(self, domains, lineage):
        """Copy a valid Let's Encrypt certificate for given domains."""
        source_private_key_path = pathlib.Path(lineage) / 'privkey.pem'
        source_certificate_path = pathlib.Path(lineage) / 'fullchain.pem'
        self._copy_certificates(domains, source_private_key_path,
                                source_certificate_path)

    def _copy_certificates(self, domains, source_private_key_path,
                           source_certificate_path):
        """Copy certificate for all domains in the certificate.

        The list of domains passed should be subset of domains listed in the
        certificate.

        """
        if '{domain}' in self.private_key_path or \
           '{domain}' in self.certificate_path:
            for domain in domains:
                private_key_path = self.private_key_path.format(domain=domain)
                certificate_path = self.certificate_path.format(domain=domain)
                self._copy_certificate(source_private_key_path,
                                       source_certificate_path,
                                       private_key_path, certificate_path)
        else:
            self._copy_certificate(source_private_key_path,
                                   source_certificate_path,
                                   self.private_key_path,
                                   self.certificate_path)

    def _copy_certificate(self, source_private_key_path,
                          source_certificate_path, private_key_path,
                          certificate_path):
        """Copy certificate for a single domain."""
        privileged.copy_certificate(self.managing_app,
                                    str(source_private_key_path),
                                    str(source_certificate_path),
                                    private_key_path, certificate_path,
                                    self.user_owner, self.group_owner)

    def _compare_certificate(self, domain, lineage):
        """Compare LE certificate with app certificate."""
        source_private_key_path = pathlib.Path(lineage) / 'privkey.pem'
        source_certificate_path = pathlib.Path(lineage) / 'fullchain.pem'
        private_key_path = self.private_key_path.format(domain=domain)
        certificate_path = self.certificate_path.format(domain=domain)
        return privileged.compare_certificate(self.managing_app,
                                              str(source_private_key_path),
                                              str(source_certificate_path),
                                              private_key_path,
                                              certificate_path)


def on_certificate_event(event, domains, lineage):
    """Start a new thread to handle a LE certificate event.

    Run in a new thread because:

    - We don't want to block the thread running Glib main loop when called from
      dbus handler in case of renewal.

    - We don't want to delay the a certificate operation with copying
      certificates to all apps.

    - We don't want to cause a page load error when restarting Apache due to
      certificate changes.

    """
    threading.Thread(target=on_certificate_event_sync,
                     args=(event, domains, lineage)).start()


def on_certificate_event_sync(event, domains, lineage):
    """Trigger certificate event hooks across all apps."""
    if isinstance(domains, str):
        domains = domains.split()

    if not lineage:
        # XXX: A better to find lineage is by reading the renewal/{domain}.conf
        lineage = str(pathlib.Path('/etc/letsencrypt/live/') / domains[0])

    assert event in ('obtained', 'renewed', 'revoked', 'deleted')

    for component in LetsEncrypt.list():
        logger.info('Handling certificate event for %s: %s, %s, %s',
                    component.component_id, event, domains, lineage)
        try:
            getattr(component, 'on_certificate_' + event)(domains, lineage)
        except Exception as exception:
            logger.exception(
                'Error executing certificate hook for %s: %s, %s, %s: %s',
                component.component_id, event, domains, lineage, exception)

    if event in ('obtained', 'renewed'):
        from plinth.modules import letsencrypt
        letsencrypt.certificate_set_last_seen_modified_time(lineage)