mirror of
https://github.com/freedombox/FreedomBox.git
synced 2026-01-28 08:03:36 +00:00
254b8a98a9
During boot or in other situations when FreedomBox Service is offline, Let's Encrypt certificate renewals might happen. When FreedomBox Service starts, check on such certificates and run certificate setup mechanism in each app to use the latest renewed certificate. Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org> Reviewed-by: Joseph Nuthalapati <njoseph@thoughtworks.com>
405 lines
16 KiB
Python
405 lines
16 KiB
Python
#
|
|
# This file is part of FreedomBox.
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as
|
|
# published by the Free Software Foundation, either version 3 of the
|
|
# License, or (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
#
|
|
"""
|
|
App component for other apps to use handle Let's Encrypt certificates.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import pathlib
|
|
import threading
|
|
|
|
from plinth import actions, app
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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 = {}
|
|
|
|
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
|
|
|
|
@classmethod
|
|
def list(cls):
|
|
"""Return a list of all Let's Encrypt components."""
|
|
return cls._all.values()
|
|
|
|
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
|
|
|
|
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:
|
|
actions.superuser_run('service', ['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:
|
|
actions.superuser_run('service', ['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:
|
|
actions.superuser_run('service', ['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."""
|
|
actions.superuser_run('letsencrypt', [
|
|
'copy-certificate', '--managing-app', self.managing_app,
|
|
'--user-owner', self.user_owner, '--group-owner', self.group_owner,
|
|
'--source-private-key-path',
|
|
str(source_private_key_path), '--source-certificate-path',
|
|
str(source_certificate_path), '--private-key-path',
|
|
private_key_path, '--certificate-path', certificate_path
|
|
])
|
|
|
|
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)
|
|
output = actions.superuser_run('letsencrypt', [
|
|
'compare-certificate', '--managing-app', self.managing_app,
|
|
'--source-private-key-path',
|
|
str(source_private_key_path), '--source-certificate-path',
|
|
str(source_certificate_path), '--private-key-path',
|
|
private_key_path, '--certificate-path', certificate_path
|
|
])
|
|
return json.loads(output)['result']
|
|
|
|
|
|
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)
|