# 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 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__)
[docs]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 = {}
[docs] 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):
"""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.
"""
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.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):
"""Return a list of domains this component's app is interested in."""
if callable(self._domains):
return self._domains()
return self._domains
[docs] @classmethod
def list(cls):
"""Return a list of all Let's Encrypt components."""
return cls._all.values()
[docs] def setup_certificates(self, app_domains=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:
service_privileged.try_restart(daemon)
[docs] 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
[docs] 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:
service_privileged.try_restart(daemon)
[docs] 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)
[docs] 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:
service_privileged.try_restart(daemon)
[docs] 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)