991 lines
45 KiB
Python
991 lines
45 KiB
Python
"""The router's PKI, for vpn-router-setup, the web service, vpn-router-ca and the user module.
|
|
|
|
The root CA (ROOT_DIR) signs the issuing CAs, and only them: p2s, which signs the router's VPN server certificate and
|
|
the users' certificates, s2s, which signs the router's site-to-site certificate and those of its peers, and the
|
|
general-purpose CAs of pki.issuing_cas. Each CA is a directory with cert.pem, key.pem and crl.pem. While the root key is
|
|
not on the router, a new issuing CA has key.pem and csr.pem and is pending until import_ca takes its certificate, signed
|
|
offline. The router's own certificates are in ROUTER_DIR and the users' in USER_DIR, with their keys, which the OpenVPN
|
|
profile embeds; a certificate issued for anything else is kept without its key. STATE records every certificate issued,
|
|
its revocation, the CRL numbers, where the root CRL came from and the retired CAs.
|
|
|
|
The certificates carry the URLs of pki.publish for their issuer's certificate (AIA) and CRL (CRL distribution point);
|
|
PUBLISH_DIR holds the DER copies served there. distribute copies what strongSwan and OpenVPN read.
|
|
|
|
Callers hold locked() around every operation that writes, which is the lock of the user module's sessions too. A request
|
|
that cannot be done raises Failure, whose status says why: 2 the request is invalid, 3 something it names does not
|
|
exist, 4 the state does not allow it, 1 anything else.
|
|
"""
|
|
|
|
import base64
|
|
import contextlib
|
|
import datetime
|
|
import fcntl
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
|
|
from cryptography import x509
|
|
from cryptography.exceptions import InvalidSignature
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import ec, rsa
|
|
from cryptography.hazmat.primitives.serialization import pkcs12
|
|
from cryptography.x509.oid import AuthorityInformationAccessOID, ExtendedKeyUsageOID, NameOID
|
|
|
|
import vpnrouter
|
|
|
|
PKI_DIR = pathlib.Path('/etc/vpn-router/pki')
|
|
ROOT_DIR = PKI_DIR / 'root'
|
|
CA_DIR = PKI_DIR / 'ca'
|
|
ROUTER_DIR = PKI_DIR / 'router'
|
|
USER_DIR = PKI_DIR / 'users'
|
|
RETIRED_DIR = PKI_DIR / 'retired'
|
|
STATE = PKI_DIR / 'state.json'
|
|
|
|
#: The DER copies of the CA certificates and CRLs, <name>.crt and <name>.crl, which the web service and Caddy serve.
|
|
PUBLISH_DIR = pathlib.Path('/var/lib/vpn-router/pki-publish')
|
|
|
|
#: The lock of the PKI and of the user module's sessions.
|
|
LOCK_FILE = pathlib.Path('/run/vpn-router/users.lock')
|
|
|
|
#: The copies strongSwan reads: one CA certificate per file, as strongSwan reads only the first of a file.
|
|
X509CA_DIR = pathlib.Path('/etc/swanctl/x509ca')
|
|
X509_DIR = pathlib.Path('/etc/swanctl/x509')
|
|
PRIVATE_DIR = pathlib.Path('/etc/swanctl/private')
|
|
X509CRL_DIR = pathlib.Path('/etc/swanctl/x509crl')
|
|
SWANCTL_CA_FILES = {'root': 'root.pem', 'p2s': 'p2s.pem', 's2s': 's2s.pem'}
|
|
VPN_SERVER_CERT = X509_DIR / 'vpn-server.pem'
|
|
VPN_SERVER_KEY = PRIVATE_DIR / 'vpn-server.key'
|
|
S2S_CERT = X509_DIR / 'site-to-site.pem'
|
|
S2S_KEY = PRIVATE_DIR / 'site-to-site.key'
|
|
|
|
#: What OpenVPN reads: the root and the P2S CA, and their CRLs, which it needs for every CA of a client's chain.
|
|
OPENVPN_CA = pathlib.Path('/etc/openvpn/server/vpn-router-ca.pem')
|
|
OPENVPN_CRL = pathlib.Path('/etc/openvpn/server/vpn-router-crl.pem')
|
|
|
|
#: The root CA in the system's trust store.
|
|
TRUST_ANCHOR = pathlib.Path('/usr/local/share/ca-certificates/vpn-router-ca.crt')
|
|
|
|
#: The router's own certificates: name -> the issuing CA and the template.
|
|
ROUTER_CERTS = {'vpn-server': ('p2s', 'vpn-server'), 'site-to-site': ('s2s', 'site-to-site')}
|
|
|
|
#: The extended key usages of vpnrouter.EXTENDED_KEY_USAGES.
|
|
EKU_OIDS = {
|
|
'server_auth': ExtendedKeyUsageOID.SERVER_AUTH,
|
|
'client_auth': ExtendedKeyUsageOID.CLIENT_AUTH,
|
|
'code_signing': ExtendedKeyUsageOID.CODE_SIGNING,
|
|
'email_protection': ExtendedKeyUsageOID.EMAIL_PROTECTION,
|
|
'time_stamping': ExtendedKeyUsageOID.TIME_STAMPING,
|
|
'ocsp_signing': ExtendedKeyUsageOID.OCSP_SIGNING,
|
|
'ipsec_ike': ExtendedKeyUsageOID.IPSEC_IKE,
|
|
'ike_intermediate': x509.ObjectIdentifier('1.3.6.1.5.5.8.2.2'),
|
|
}
|
|
|
|
#: The revocation reasons revoke takes.
|
|
REASONS = ('unspecified', 'key_compromise', 'affiliation_changed', 'superseded', 'cessation_of_operation',
|
|
'privilege_withdrawn')
|
|
|
|
#: Days before an expiry or a CRL's next update that status warns.
|
|
WARN_DAYS = 30
|
|
|
|
#: The longest PFX password, in characters.
|
|
PASSWORD_MAX = 128
|
|
|
|
DNS_NAME = re.compile(r'^(\*\.)?[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?(\.[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?)*$')
|
|
EMAIL = re.compile(r'^[^@\s]+@[^@\s]+\.[^@\s]+$')
|
|
SERIAL = re.compile(r'^[0-9A-Fa-f]{1,40}$')
|
|
|
|
CA_USAGE = x509.KeyUsage(digital_signature=False, content_commitment=False, key_encipherment=False,
|
|
data_encipherment=False, key_agreement=False, key_cert_sign=True, crl_sign=True,
|
|
encipher_only=False, decipher_only=False)
|
|
|
|
|
|
class Failure(Exception):
|
|
"""A request that cannot be done; status is 2 invalid, 3 not found, 4 not allowed in this state, 1 other."""
|
|
|
|
def __init__(self, message, status=1):
|
|
super().__init__(message)
|
|
self.status = status
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def locked():
|
|
"""Hold the lock of the PKI, which the user module's sessions share."""
|
|
LOCK_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(LOCK_FILE, 'w') as lock:
|
|
fcntl.flock(lock, fcntl.LOCK_EX)
|
|
yield
|
|
|
|
|
|
def now():
|
|
return datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0)
|
|
|
|
|
|
def iso(moment):
|
|
return moment.astimezone(datetime.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
|
|
|
|
|
|
def parse_iso(text):
|
|
return datetime.datetime.strptime(text, '%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=datetime.timezone.utc)
|
|
|
|
|
|
def utc(item, name):
|
|
"""A time of a certificate or CRL in UTC: the *_utc attribute of newer cryptography, or the naive one of older."""
|
|
if hasattr(item, f'{name}_utc'):
|
|
return getattr(item, f'{name}_utc')
|
|
value = getattr(item, name)
|
|
return value.replace(tzinfo=datetime.timezone.utc) if value is not None else None
|
|
|
|
|
|
def directory(name):
|
|
"""The directory of the root (root) or of an issuing CA."""
|
|
return ROOT_DIR if name == 'root' else CA_DIR / name
|
|
|
|
|
|
def read_cert(path):
|
|
return x509.load_pem_x509_certificate(path.read_bytes()) if path.is_file() else None
|
|
|
|
|
|
def read_key(path):
|
|
return serialization.load_pem_private_key(path.read_bytes(), None) if path.is_file() else None
|
|
|
|
|
|
def read_crl(path):
|
|
return x509.load_pem_x509_crl(path.read_bytes()) if path.is_file() else None
|
|
|
|
|
|
def cert_pem(cert):
|
|
return cert.public_bytes(serialization.Encoding.PEM)
|
|
|
|
|
|
def key_pem(key, password=None):
|
|
encryption = serialization.BestAvailableEncryption(password) if password else serialization.NoEncryption()
|
|
return key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, encryption)
|
|
|
|
|
|
def write(path, data, mode):
|
|
"""Write the bytes to the file with the mode, replacing it in one step; True when its content changed."""
|
|
if path.is_file() and path.read_bytes() == data:
|
|
os.chmod(path, mode)
|
|
return False
|
|
path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
descriptor, name = tempfile.mkstemp(dir=path.parent, prefix=f'.{path.name}.')
|
|
try:
|
|
with os.fdopen(descriptor, 'wb') as handle:
|
|
handle.write(data)
|
|
os.chmod(name, mode)
|
|
os.replace(name, path)
|
|
except OSError:
|
|
os.unlink(name)
|
|
raise
|
|
return True
|
|
|
|
|
|
def remove(path):
|
|
"""Remove the file; True when it existed."""
|
|
if path.is_file():
|
|
path.unlink()
|
|
return True
|
|
return False
|
|
|
|
|
|
def read_state():
|
|
try:
|
|
state = json.loads(STATE.read_text(encoding='utf-8'))
|
|
except FileNotFoundError:
|
|
state = {}
|
|
state.setdefault('certificates', [])
|
|
state.setdefault('crl_numbers', {})
|
|
state.setdefault('retired', [])
|
|
state.setdefault('root_crl', None)
|
|
return state
|
|
|
|
|
|
def write_state(state):
|
|
write(STATE, (json.dumps(state, indent=2) + '\n').encode(), 0o600)
|
|
|
|
|
|
def new_key(kind):
|
|
"""A new private key of one of vpnrouter.PKI_KEYS."""
|
|
if kind.startswith('rsa-'):
|
|
return rsa.generate_private_key(public_exponent=65537, key_size=int(kind[4:]))
|
|
return ec.generate_private_key(ec.SECP384R1() if kind == 'ec-p384' else ec.SECP256R1())
|
|
|
|
|
|
def digest(key):
|
|
"""The hash a key signs with: SHA-384 for a P-384 key, SHA-256 otherwise."""
|
|
return hashes.SHA384() if isinstance(key, ec.EllipticCurvePrivateKey) and key.curve.name == 'secp384r1' \
|
|
else hashes.SHA256()
|
|
|
|
|
|
def common_name(item):
|
|
"""The common name of a certificate's or a CSR's subject, or the whole subject without one."""
|
|
names = item.subject.get_attributes_for_oid(NameOID.COMMON_NAME)
|
|
return names[0].value if names else item.subject.rfc4514_string()
|
|
|
|
|
|
def san_values(item):
|
|
"""The subject alternative names of a certificate or a CSR as strings: DNS:, IP: and email: values."""
|
|
try:
|
|
names = item.extensions.get_extension_for_class(x509.SubjectAlternativeName).value
|
|
except x509.ExtensionNotFound:
|
|
return []
|
|
return ([f'DNS:{value}' for value in names.get_values_for_type(x509.DNSName)] +
|
|
[f'IP:{value}' for value in names.get_values_for_type(x509.IPAddress)] +
|
|
[f'email:{value}' for value in names.get_values_for_type(x509.RFC822Name)])
|
|
|
|
|
|
def same_key(first, second):
|
|
der = serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo
|
|
return first.public_bytes(*der) == second.public_bytes(*der)
|
|
|
|
|
|
def issued_by(cert, issuer):
|
|
"""Whether the issuer's key signed the certificate."""
|
|
try:
|
|
cert.verify_directly_issued_by(issuer)
|
|
except (ValueError, TypeError, InvalidSignature):
|
|
return False
|
|
return True
|
|
|
|
|
|
def is_ca(cert):
|
|
try:
|
|
return cert.extensions.get_extension_for_class(x509.BasicConstraints).value.ca
|
|
except x509.ExtensionNotFound:
|
|
return False
|
|
|
|
|
|
def is_ca_certificate(text):
|
|
"""Whether the text is the PEM certificate of a CA."""
|
|
try:
|
|
return is_ca(x509.load_pem_x509_certificate(text.strip().encode()))
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def authority_key_id(issuer):
|
|
try:
|
|
return x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(
|
|
issuer.extensions.get_extension_for_class(x509.SubjectKeyIdentifier).value)
|
|
except x509.ExtensionNotFound:
|
|
return x509.AuthorityKeyIdentifier.from_issuer_public_key(issuer.public_key())
|
|
|
|
|
|
def sign(subject, public_key, issuer, issuer_key, days, extensions, urls=(), issuer_name=None):
|
|
"""A certificate with the common name subject for public_key, signed by issuer_key, valid for days but not past the
|
|
issuer, with the extensions [(extension, critical)].
|
|
|
|
issuer is the issuer's certificate, None for a self-signed one. The certificate points at the issuer's certificate
|
|
and CRL, issuer_name.crt and issuer_name.crl, under each of the base urls.
|
|
"""
|
|
moment = now()
|
|
end = moment + datetime.timedelta(days=days)
|
|
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, subject)])
|
|
builder = (x509.CertificateBuilder()
|
|
.subject_name(name)
|
|
.issuer_name(issuer.subject if issuer is not None else name)
|
|
.public_key(public_key)
|
|
.serial_number(x509.random_serial_number())
|
|
.not_valid_before(moment - datetime.timedelta(minutes=5))
|
|
.not_valid_after(min(end, utc(issuer, 'not_valid_after')) if issuer is not None else end)
|
|
.add_extension(x509.SubjectKeyIdentifier.from_public_key(public_key), False))
|
|
if issuer is not None:
|
|
builder = builder.add_extension(authority_key_id(issuer), False)
|
|
if issuer is not None and urls:
|
|
builder = builder.add_extension(x509.AuthorityInformationAccess([
|
|
x509.AccessDescription(AuthorityInformationAccessOID.CA_ISSUERS,
|
|
x509.UniformResourceIdentifier(f'{url}{issuer_name}.crt')) for url in urls]), False)
|
|
builder = builder.add_extension(x509.CRLDistributionPoints([x509.DistributionPoint(
|
|
[x509.UniformResourceIdentifier(f'{url}{issuer_name}.crl') for url in urls], None, None, None)]), False)
|
|
for extension, critical in extensions:
|
|
builder = builder.add_extension(extension, critical)
|
|
return builder.sign(issuer_key, digest(issuer_key))
|
|
|
|
|
|
def ca_extensions(path_length):
|
|
return [(x509.BasicConstraints(ca=True, path_length=path_length), True), (CA_USAGE, True)]
|
|
|
|
|
|
def general_names(values, allowed):
|
|
"""The subject alternative names of values, each a DNS name, an IP address or an e-mail address of the allowed
|
|
types (vpnrouter.SAN_TYPES)."""
|
|
names = []
|
|
for value in values:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise Failure('a subject alternative name must be a non-empty string', 2)
|
|
value = value.strip()
|
|
try:
|
|
kind, name = 'ip', x509.IPAddress(ipaddress.ip_address(value))
|
|
except ValueError:
|
|
if '@' in value:
|
|
if not EMAIL.match(value):
|
|
raise Failure(f'{value!r} is not an e-mail address', 2)
|
|
kind, name = 'email', x509.RFC822Name(value)
|
|
else:
|
|
if not DNS_NAME.match(value.lower()):
|
|
raise Failure(f'{value!r} is not a DNS name, an IP address or an e-mail address', 2)
|
|
kind, name = 'dns', x509.DNSName(value.lower())
|
|
if kind not in allowed:
|
|
raise Failure(f'{value}: the template allows no {kind} names', 2)
|
|
if name not in names:
|
|
names.append(name)
|
|
return names
|
|
|
|
|
|
def leaf_extensions(template, names):
|
|
"""The extensions of a certificate of the template with the subject alternative names."""
|
|
usage = x509.KeyUsage(**{option: option in template['key_usage'] for option in vpnrouter.KEY_USAGES},
|
|
key_cert_sign=False, crl_sign=False, encipher_only=False, decipher_only=False)
|
|
extensions = [(x509.BasicConstraints(ca=False, path_length=None), True), (usage, True)]
|
|
if template['extended_key_usage']:
|
|
extensions.append((x509.ExtendedKeyUsage([EKU_OIDS[item] for item in template['extended_key_usage']]), False))
|
|
if names:
|
|
extensions.append((x509.SubjectAlternativeName(names), False))
|
|
return extensions
|
|
|
|
|
|
def check_subject(kind, subject):
|
|
"""Check a common name against the subject of a template (vpnrouter.TEMPLATE_SUBJECTS)."""
|
|
if not isinstance(subject, str) or not subject.strip():
|
|
raise Failure('the subject is required', 2)
|
|
subject = subject.strip()
|
|
if kind == 'dns' and not DNS_NAME.match(subject.lower()):
|
|
raise Failure(f'{subject!r} is not a DNS name, which the template needs as the subject', 2)
|
|
if kind == 'email' and not EMAIL.match(subject):
|
|
raise Failure(f'{subject!r} is not an e-mail address, which the template needs as the subject', 2)
|
|
if len(subject) > 64 or not subject.isprintable():
|
|
raise Failure('the subject must have 1 to 64 printable characters', 2)
|
|
return subject.lower() if kind == 'dns' else subject
|
|
|
|
|
|
def record(state, cert, ca, kind, template=None):
|
|
"""Add a certificate to the state: kind is ca, router, user or external."""
|
|
state['certificates'].append({
|
|
'serial': f'{cert.serial_number:X}', 'ca': ca, 'kind': kind, 'template': template,
|
|
'subject': common_name(cert), 'sans': san_values(cert), 'created': iso(now()),
|
|
'expires': iso(utc(cert, 'not_valid_after')), 'revoked': None, 'reason': None,
|
|
})
|
|
|
|
|
|
def sign_crl(cfg, name, state):
|
|
"""Sign a new CRL of the CA with the certificates it revoked that have not expired, valid for pki.root_crl_days for
|
|
the root and pki.crl_days for an issuing CA."""
|
|
folder = directory(name)
|
|
cert, key = read_cert(folder / 'cert.pem'), read_key(folder / 'key.pem')
|
|
number = state['crl_numbers'].get(name, 0) + 1
|
|
moment = now()
|
|
days = cfg['pki']['root_crl_days'] if name == 'root' else cfg['pki']['crl_days']
|
|
builder = (x509.CertificateRevocationListBuilder()
|
|
.issuer_name(cert.subject)
|
|
.last_update(moment)
|
|
.next_update(moment + datetime.timedelta(days=days))
|
|
.add_extension(x509.CRLNumber(number), False)
|
|
.add_extension(authority_key_id(cert), False))
|
|
for entry in state['certificates']:
|
|
if entry['ca'] == name and entry['revoked'] and entry['expires'] > iso(moment):
|
|
revoked = (x509.RevokedCertificateBuilder()
|
|
.serial_number(int(entry['serial'], 16))
|
|
.revocation_date(parse_iso(entry['revoked'])))
|
|
if entry['reason'] and entry['reason'] != 'unspecified':
|
|
revoked = revoked.add_extension(x509.CRLReason(x509.ReasonFlags[entry['reason']]), False)
|
|
builder = builder.add_revoked_certificate(revoked.build())
|
|
crl = builder.sign(key, digest(key))
|
|
state['crl_numbers'][name] = number
|
|
if name == 'root':
|
|
state['root_crl'] = 'signed'
|
|
write(folder / 'crl.pem', crl.public_bytes(serialization.Encoding.PEM), 0o644)
|
|
|
|
|
|
def crl_stale(path):
|
|
"""Whether the CRL is missing or past half of its validity."""
|
|
crl = read_crl(path)
|
|
if crl is None:
|
|
return True
|
|
last, following = utc(crl, 'last_update'), utc(crl, 'next_update')
|
|
return following is None or now() >= last + (following - last) / 2
|
|
|
|
|
|
def active():
|
|
"""The names of the issuing CAs with a certificate, sorted."""
|
|
return sorted(folder.name for folder in CA_DIR.iterdir() if (folder / 'cert.pem').is_file()) \
|
|
if CA_DIR.is_dir() else []
|
|
|
|
|
|
def ready(name):
|
|
"""Whether the issuing CA has its certificate and key, and so issues certificates."""
|
|
return (CA_DIR / name / 'cert.pem').is_file() and (CA_DIR / name / 'key.pem').is_file()
|
|
|
|
|
|
def refresh_crls(cfg, state, force=()):
|
|
"""Sign a new CRL for each CA with its key on the router whose CRL is stale, or that force names."""
|
|
for name in ['root', *active()]:
|
|
folder = directory(name)
|
|
if (folder / 'key.pem').is_file() and (folder / 'cert.pem').is_file() and \
|
|
(name in force or crl_stale(folder / 'crl.pem')):
|
|
sign_crl(cfg, name, state)
|
|
|
|
|
|
def renewal_due(cert):
|
|
"""Whether less than a third of the certificate's lifetime is left."""
|
|
start, end = utc(cert, 'not_valid_before'), utc(cert, 'not_valid_after')
|
|
return now() >= end - (end - start) / 3
|
|
|
|
|
|
def router_identities(name, ctx):
|
|
"""The common name and the subject alternative names of one of the router's certificates (ROUTER_CERTS)."""
|
|
fqdn = ctx['local_fqdn']
|
|
if name == 'vpn-server':
|
|
return fqdn, [fqdn]
|
|
return fqdn, [value for value in (fqdn, ctx['public_ip'], ctx['int_addr']) if value]
|
|
|
|
|
|
def ensure_router_cert(cfg, ctx, state, name, urls):
|
|
"""Issue one of the router's certificates when it is missing, due for renewal, not issued by its CA's current key,
|
|
named differently or without one of the names it needs; True when it was issued."""
|
|
ca, template_name = ROUTER_CERTS[name]
|
|
issuer, issuer_key = read_cert(CA_DIR / ca / 'cert.pem'), read_key(CA_DIR / ca / 'key.pem')
|
|
subject, sans = router_identities(name, ctx)
|
|
if issuer is None or issuer_key is None or not subject:
|
|
return False
|
|
cert_path, key_path = ROUTER_DIR / f'{name}_cert.pem', ROUTER_DIR / f'{name}_key.pem'
|
|
cert = read_cert(cert_path)
|
|
wanted = {f'IP:{value}' if is_address(value) else f'DNS:{value.lower()}' for value in sans}
|
|
if cert is not None and key_path.is_file() and not renewal_due(cert) and issued_by(cert, issuer) and \
|
|
common_name(cert) == subject and wanted <= set(san_values(cert)):
|
|
return False
|
|
template = vpnrouter.pki_templates(cfg)[template_name]
|
|
key = new_key(template['key'])
|
|
cert = sign(subject, key.public_key(), issuer, issuer_key, template['days'],
|
|
leaf_extensions(template, general_names(sans, vpnrouter.SAN_TYPES)), urls, ca)
|
|
write(key_path, key_pem(key), 0o600)
|
|
write(cert_path, cert_pem(cert), 0o644)
|
|
record(state, cert, ca, 'router', template_name)
|
|
return True
|
|
|
|
|
|
def is_address(value):
|
|
try:
|
|
ipaddress.ip_address(value)
|
|
except ValueError:
|
|
return False
|
|
return True
|
|
|
|
|
|
def provision(cfg, ctx):
|
|
"""Create what the configuration needs and is missing, and keep it current; the notices to report.
|
|
|
|
The root is created when it has neither certificate nor key. Each issuing CA without a certificate is signed by
|
|
the root key, or, while the root key is not on the router, gets its key and a CSR and is pending. An issuing CA
|
|
that pki.issuing_cas no longer names is retired: its directory moves to RETIRED_DIR, and its certificate is revoked
|
|
on the root CRL. Stale CRLs are signed again, the router's certificates issued or renewed (ensure_router_cert), and
|
|
the copies of PUBLISH_DIR written.
|
|
"""
|
|
for folder in (PKI_DIR, CA_DIR, ROUTER_DIR, USER_DIR):
|
|
folder.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
os.chmod(folder, 0o700)
|
|
pki = cfg['pki']
|
|
state = read_state()
|
|
urls = vpnrouter.pki_base_urls(cfg)
|
|
notices = []
|
|
|
|
root_cert, root_key = read_cert(ROOT_DIR / 'cert.pem'), read_key(ROOT_DIR / 'key.pem')
|
|
if root_cert is None:
|
|
if root_key is not None:
|
|
raise Failure(f'{ROOT_DIR / "key.pem"} has no certificate; place root/cert.pem beside it')
|
|
root_key = new_key(pki['key'])
|
|
root_cert = sign(pki['root_name'], root_key.public_key(), None, root_key, pki['root_days'], ca_extensions(1))
|
|
write(ROOT_DIR / 'key.pem', key_pem(root_key), 0o600)
|
|
write(ROOT_DIR / 'cert.pem', cert_pem(root_cert), 0o644)
|
|
|
|
authorities = vpnrouter.issuing_cas(cfg)
|
|
force = set()
|
|
for name, authority in authorities.items():
|
|
folder = CA_DIR / name
|
|
if (folder / 'cert.pem').is_file():
|
|
continue
|
|
key = read_key(folder / 'key.pem')
|
|
if key is None:
|
|
key = new_key(pki['key'])
|
|
write(folder / 'key.pem', key_pem(key), 0o600)
|
|
if root_key is not None:
|
|
cert = sign(authority['name'].strip(), key.public_key(), root_cert, root_key, pki['ca_days'],
|
|
ca_extensions(0), urls, 'root')
|
|
write(folder / 'cert.pem', cert_pem(cert), 0o644)
|
|
remove(folder / 'csr.pem')
|
|
record(state, cert, 'root', 'ca')
|
|
force.add(name)
|
|
continue
|
|
if not (folder / 'csr.pem').is_file():
|
|
request = (x509.CertificateSigningRequestBuilder()
|
|
.subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, authority['name'].strip())]))
|
|
.add_extension(x509.BasicConstraints(ca=True, path_length=0), True)
|
|
.add_extension(CA_USAGE, True)
|
|
.sign(key, digest(key)))
|
|
write(folder / 'csr.pem', request.public_bytes(serialization.Encoding.PEM), 0o644)
|
|
notices.append(f'issuing CA {name} waits for its certificate: sign {folder / "csr.pem"} with the root CA and '
|
|
f'import it with vpn-router-ca ca import {name}')
|
|
|
|
for folder in sorted(CA_DIR.iterdir()):
|
|
if not folder.is_dir() or folder.name in authorities:
|
|
continue
|
|
cert = read_cert(folder / 'cert.pem')
|
|
RETIRED_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
shutil.move(str(folder), str(RETIRED_DIR / f'{folder.name}-{now().strftime("%Y%m%d%H%M%S")}'))
|
|
state['retired'].append({'name': folder.name, 'retired': iso(now()),
|
|
'serial': f'{cert.serial_number:X}' if cert is not None else None})
|
|
for entry in state['certificates']:
|
|
if cert is not None and entry['ca'] == 'root' and entry['serial'] == f'{cert.serial_number:X}' and \
|
|
not entry['revoked']:
|
|
entry['revoked'], entry['reason'] = iso(now()), 'cessation_of_operation'
|
|
force.add('root')
|
|
if root_key is None and cert is not None:
|
|
notices.append(f'issuing CA {folder.name} was retired: revoke its certificate on the root CRL offline and '
|
|
'upload the CRL with vpn-router-ca root import-crl')
|
|
|
|
refresh_crls(cfg, state, force)
|
|
for name in ROUTER_CERTS:
|
|
ensure_router_cert(cfg, ctx, state, name, urls)
|
|
write_state(state)
|
|
publish()
|
|
return notices
|
|
|
|
|
|
def publish():
|
|
"""Write the DER copies of the certificates and CRLs of the root and the issuing CAs with a certificate to
|
|
PUBLISH_DIR, and remove any other file there; True when anything changed."""
|
|
PUBLISH_DIR.mkdir(parents=True, exist_ok=True)
|
|
os.chmod(PUBLISH_DIR, 0o755)
|
|
wanted = {}
|
|
for name in ['root', *active()]:
|
|
cert, crl = read_cert(directory(name) / 'cert.pem'), read_crl(directory(name) / 'crl.pem')
|
|
if cert is not None:
|
|
wanted[f'{name}.crt'] = cert.public_bytes(serialization.Encoding.DER)
|
|
if crl is not None:
|
|
wanted[f'{name}.crl'] = crl.public_bytes(serialization.Encoding.DER)
|
|
changed = False
|
|
for path in PUBLISH_DIR.iterdir():
|
|
if path.name not in wanted:
|
|
path.unlink()
|
|
changed = True
|
|
for name, data in wanted.items():
|
|
changed |= write(PUBLISH_DIR / name, data, 0o644)
|
|
return changed
|
|
|
|
|
|
def remote_ca_file(connection):
|
|
"""The name in X509CA_DIR of the remote CA certificate of a site_to_site connection."""
|
|
return f's2s-{connection}.pem'
|
|
|
|
|
|
def install_crls():
|
|
"""Copy the CRLs of the root, p2s and s2s to strongSwan, and those of the root and p2s to OpenVPN; True on change."""
|
|
changed = False
|
|
for name, file in SWANCTL_CA_FILES.items():
|
|
source = directory(name) / 'crl.pem'
|
|
changed |= vpnrouter.install_file(source, X509CRL_DIR / file, 0o644) if source.is_file() \
|
|
else remove(X509CRL_DIR / file)
|
|
crls = [directory(name) / 'crl.pem' for name in ('root', 'p2s')]
|
|
if all(path.is_file() for path in crls):
|
|
changed |= write(OPENVPN_CRL, b''.join(path.read_bytes() for path in crls), 0o644)
|
|
return changed
|
|
|
|
|
|
def reload_crls():
|
|
"""Install the CRLs, and have strongSwan load them when they changed; OpenVPN reads its file when it changes."""
|
|
if install_crls():
|
|
subprocess.run(['swanctl', '--load-creds'], capture_output=True, text=True)
|
|
|
|
|
|
def distribute(cfg):
|
|
"""Copy the certificates, keys and CRLs strongSwan, OpenVPN and the system's trust store read; True on change.
|
|
|
|
strongSwan gets the root, p2s and s2s, the remote CA of each site_to_site connection that has one, the router's
|
|
certificates and keys, and the CRLs; OpenVPN the root and p2s in one file, and their CRLs (install_crls).
|
|
"""
|
|
changed = False
|
|
for name, file in SWANCTL_CA_FILES.items():
|
|
source = directory(name) / 'cert.pem'
|
|
changed |= vpnrouter.install_file(source, X509CA_DIR / file, 0o644) if source.is_file() \
|
|
else remove(X509CA_DIR / file)
|
|
remote = {remote_ca_file(name): connection['remote_ca'].strip().encode() + b'\n'
|
|
for name, connection in cfg['site_to_site'].items() if connection['remote_ca'].strip()}
|
|
for path in X509CA_DIR.glob(remote_ca_file('*')):
|
|
if path.name not in remote:
|
|
changed |= remove(path)
|
|
for file, data in remote.items():
|
|
changed |= write(X509CA_DIR / file, data, 0o644)
|
|
for name, cert, key in (('vpn-server', VPN_SERVER_CERT, VPN_SERVER_KEY), ('site-to-site', S2S_CERT, S2S_KEY)):
|
|
if (ROUTER_DIR / f'{name}_cert.pem').is_file():
|
|
changed |= vpnrouter.install_file(ROUTER_DIR / f'{name}_cert.pem', cert, 0o644)
|
|
changed |= vpnrouter.install_file(ROUTER_DIR / f'{name}_key.pem', key, 0o600)
|
|
else:
|
|
changed |= remove(cert) | remove(key)
|
|
chain = [directory(name) / 'cert.pem' for name in ('root', 'p2s')]
|
|
if all(path.is_file() for path in chain):
|
|
changed |= write(OPENVPN_CA, b''.join(path.read_bytes() for path in chain), 0o644)
|
|
changed |= install_crls()
|
|
if vpnrouter.install_file(ROOT_DIR / 'cert.pem', TRUST_ANCHOR, 0o644):
|
|
changed = True
|
|
if shutil.which('update-ca-certificates'):
|
|
subprocess.run(['update-ca-certificates'], capture_output=True, text=True)
|
|
return changed
|
|
|
|
|
|
def chain_of(ca):
|
|
"""The PEM certificates of the issuing CA and the root."""
|
|
return (cert_pem(read_cert(CA_DIR / ca / 'cert.pem')) + cert_pem(read_cert(ROOT_DIR / 'cert.pem'))).decode()
|
|
|
|
|
|
def issuer_of(cfg, ca, template):
|
|
"""The certificate and key of the issuing CA, and the template, checked to be one the CA issues."""
|
|
authorities = vpnrouter.issuing_cas(cfg)
|
|
if ca not in authorities:
|
|
raise Failure(f'no issuing CA {ca}', 3)
|
|
if template not in authorities[ca]['templates']:
|
|
raise Failure(f'issuing CA {ca} does not issue {template} certificates', 2)
|
|
if not ready(ca):
|
|
raise Failure(f'issuing CA {ca} has no certificate yet', 4)
|
|
return read_cert(CA_DIR / ca / 'cert.pem'), read_key(CA_DIR / ca / 'key.pem'), \
|
|
vpnrouter.pki_templates(cfg)[template]
|
|
|
|
|
|
def issue(cfg, ca, template, csr=None, subject=None, sans=None, password=None):
|
|
"""Issue a certificate of the template from the issuing CA for something other than the router.
|
|
|
|
With a CSR, the certificate is for the CSR's key, common name and subject alternative names, and the answer is
|
|
{serial, certificate, chain} in PEM. Without one, the router makes a key of the template for the subject and sans,
|
|
and the answer is {serial, pfx}: a PKCS#12 file in base64 with the key, the certificate and the chain, protected
|
|
by the password; the key is not kept. The templates of vpnrouter.ROUTER_TEMPLATES are not issued here.
|
|
"""
|
|
if template in vpnrouter.ROUTER_TEMPLATES:
|
|
raise Failure(f'{template} certificates are issued by the router only', 2)
|
|
issuer, issuer_key, values = issuer_of(cfg, ca, template)
|
|
key = None
|
|
if csr is not None:
|
|
try:
|
|
request = x509.load_pem_x509_csr(csr.strip().encode() if isinstance(csr, str) else b'')
|
|
except ValueError:
|
|
raise Failure('the CSR is not a PEM certificate signing request', 2)
|
|
if not request.is_signature_valid:
|
|
raise Failure('the signature of the CSR is not valid', 2)
|
|
subject = common_name(request)
|
|
sans = [value.split(':', 1)[1] for value in san_values(request)]
|
|
public_key = request.public_key()
|
|
else:
|
|
if not isinstance(password, str) or not 1 <= len(password) <= PASSWORD_MAX:
|
|
raise Failure(f'the password of the PFX file must have 1 to {PASSWORD_MAX} characters', 2)
|
|
if sans is None:
|
|
sans = []
|
|
if not isinstance(sans, list):
|
|
raise Failure('sans must be a list of names', 2)
|
|
key = new_key(values['key'])
|
|
public_key = key.public_key()
|
|
subject = check_subject(values['subject'], subject)
|
|
names = general_names(([subject] if values['subject'] == 'dns' else []) + list(sans),
|
|
set(values['sans']) | ({'dns'} if values['subject'] == 'dns' else set()))
|
|
cert = sign(subject, public_key, issuer, issuer_key, values['days'], leaf_extensions(values, names),
|
|
vpnrouter.pki_base_urls(cfg), ca)
|
|
state = read_state()
|
|
record(state, cert, ca, 'external', template)
|
|
write_state(state)
|
|
serial = f'{cert.serial_number:X}'
|
|
if key is None:
|
|
return {'serial': serial, 'certificate': cert_pem(cert).decode(), 'chain': chain_of(ca)}
|
|
encryption = (serialization.PrivateFormat.PKCS12.encryption_builder()
|
|
.kdf_rounds(50000)
|
|
.key_cert_algorithm(pkcs12.PBES.PBESv1SHA1And3KeyTripleDESCBC)
|
|
.hmac_hash(hashes.SHA1())
|
|
.build(password.encode()))
|
|
pfx = pkcs12.serialize_key_and_certificates(subject.encode(), key, cert,
|
|
[issuer, read_cert(ROOT_DIR / 'cert.pem')], encryption)
|
|
return {'serial': serial, 'pfx': base64.b64encode(pfx).decode()}
|
|
|
|
|
|
def user_files(name):
|
|
"""The certificate and the key of a user."""
|
|
return USER_DIR / f'{name}_cert.pem', USER_DIR / f'{name}_key.pem'
|
|
|
|
|
|
def issue_user(cfg, name, email):
|
|
"""Issue the user's certificate from the P2S CA with the template vpn-user: the user name as the common name and a
|
|
DNS name, and the e-mail address; the certificate and the key are kept in USER_DIR."""
|
|
issuer, issuer_key, values = issuer_of(cfg, 'p2s', 'vpn-user')
|
|
key = new_key(values['key'])
|
|
cert = sign(name, key.public_key(), issuer, issuer_key, values['days'],
|
|
leaf_extensions(values, general_names([name, email], vpnrouter.SAN_TYPES)),
|
|
vpnrouter.pki_base_urls(cfg), 'p2s')
|
|
cert_path, key_path = user_files(name)
|
|
write(key_path, key_pem(key), 0o600)
|
|
write(cert_path, cert_pem(cert), 0o644)
|
|
state = read_state()
|
|
record(state, cert, 'p2s', 'user', 'vpn-user')
|
|
write_state(state)
|
|
|
|
|
|
def revoke_entries(cfg, state, entries, reason):
|
|
"""Mark the certificates revoked, sign new CRLs of their CAs, and publish and install them."""
|
|
for entry in entries:
|
|
entry['revoked'], entry['reason'] = iso(now()), reason
|
|
names = {entry['ca'] for entry in entries}
|
|
for name in names:
|
|
if not (directory(name) / 'key.pem').is_file():
|
|
raise Failure(f'the key of {name} is not on the router, so it cannot sign its CRL', 4)
|
|
for name in names:
|
|
sign_crl(cfg, name, state)
|
|
write_state(state)
|
|
publish()
|
|
reload_crls()
|
|
|
|
|
|
def revoke_user(cfg, name):
|
|
"""Revoke the user's certificates and remove the certificate and the key."""
|
|
state = read_state()
|
|
entries = [entry for entry in state['certificates']
|
|
if entry['kind'] == 'user' and entry['subject'] == name and not entry['revoked']]
|
|
if entries:
|
|
revoke_entries(cfg, state, entries, 'cessation_of_operation')
|
|
for path in user_files(name):
|
|
path.unlink(missing_ok=True)
|
|
|
|
|
|
def revoke(cfg, serial, reason=None):
|
|
"""Revoke a certificate the router issued, by its serial number, for one of REASONS.
|
|
|
|
A user's certificate is revoked by removing the user, and an issuing CA by removing it from pki.issuing_cas. A
|
|
revoked router certificate is removed, so that vpn-router-setup issues a new one; the answer says so with
|
|
reissue.
|
|
"""
|
|
reason = reason or 'unspecified'
|
|
if reason not in REASONS:
|
|
raise Failure(f'the reason must be one of {", ".join(REASONS)}', 2)
|
|
if not isinstance(serial, str) or not SERIAL.match(serial):
|
|
raise Failure('the serial number must be hexadecimal', 2)
|
|
state = read_state()
|
|
entry = next((entry for entry in state['certificates'] if entry['serial'] == serial.upper().lstrip('0')), None)
|
|
if entry is None:
|
|
raise Failure(f'no certificate with the serial number {serial}', 3)
|
|
if entry['revoked']:
|
|
raise Failure(f'certificate {serial} is revoked', 4)
|
|
if entry['kind'] == 'user':
|
|
raise Failure(f'certificate {serial} is the certificate of user {entry["subject"]}; remove the user', 2)
|
|
if entry['kind'] == 'ca':
|
|
raise Failure(f'certificate {serial} is an issuing CA; remove it from pki.issuing_cas', 2)
|
|
revoke_entries(cfg, state, [entry], reason)
|
|
reissue = False
|
|
if entry['kind'] == 'router':
|
|
for name, (ca, _) in ROUTER_CERTS.items():
|
|
cert = read_cert(ROUTER_DIR / f'{name}_cert.pem')
|
|
if cert is not None and f'{cert.serial_number:X}' == entry['serial']:
|
|
remove(ROUTER_DIR / f'{name}_cert.pem')
|
|
reissue = True
|
|
return {**entry, 'reissue': reissue}
|
|
|
|
|
|
def certificates():
|
|
"""The certificates the router issued, newest first, each with state: valid, expired or revoked."""
|
|
moment = iso(now())
|
|
return [{**entry, 'state': 'revoked' if entry['revoked'] else 'expired' if entry['expires'] <= moment else 'valid'}
|
|
for entry in reversed(read_state()['certificates'])]
|
|
|
|
|
|
def csr(cfg, name):
|
|
"""The CSR of a pending issuing CA and the openssl extension section to sign it with: {name, csr, extensions}."""
|
|
if name not in vpnrouter.issuing_cas(cfg):
|
|
raise Failure(f'no issuing CA {name}', 3)
|
|
path = CA_DIR / name / 'csr.pem'
|
|
if not path.is_file():
|
|
raise Failure(f'issuing CA {name} is not waiting for a certificate', 4)
|
|
urls = vpnrouter.pki_base_urls(cfg)
|
|
lines = ['[ issuing_ca ]', 'basicConstraints = critical, CA:TRUE, pathlen:0', 'keyUsage = critical, keyCertSign, cRLSign',
|
|
'subjectKeyIdentifier = hash', 'authorityKeyIdentifier = keyid:always']
|
|
if urls:
|
|
lines.append('authorityInfoAccess = ' + ', '.join(f'caIssuers;URI:{url}root.crt' for url in urls))
|
|
lines.append('crlDistributionPoints = ' + ', '.join(f'URI:{url}root.crl' for url in urls))
|
|
return {'name': name, 'csr': path.read_text(encoding='utf-8'), 'extensions': '\n'.join(lines) + '\n'}
|
|
|
|
|
|
def import_ca(cfg, name, text):
|
|
"""Take the certificate of a pending issuing CA, signed by the root offline: it must be a CA certificate the root
|
|
signed for the key of the CA's CSR."""
|
|
if name not in vpnrouter.issuing_cas(cfg):
|
|
raise Failure(f'no issuing CA {name}', 3)
|
|
folder = CA_DIR / name
|
|
if (folder / 'cert.pem').is_file():
|
|
raise Failure(f'issuing CA {name} has its certificate', 4)
|
|
key = read_key(folder / 'key.pem')
|
|
if key is None:
|
|
raise Failure(f'issuing CA {name} has no key; vpn-router-setup makes it with the CSR', 3)
|
|
try:
|
|
cert = x509.load_pem_x509_certificate(text.strip().encode() if isinstance(text, str) else b'')
|
|
except ValueError:
|
|
raise Failure('the certificate is not a PEM certificate', 2)
|
|
if not issued_by(cert, read_cert(ROOT_DIR / 'cert.pem')):
|
|
raise Failure('the certificate is not signed by the root CA', 2)
|
|
if not same_key(cert.public_key(), key.public_key()):
|
|
raise Failure(f'the certificate is not for the key of the CSR of {name}', 2)
|
|
if not is_ca(cert):
|
|
raise Failure('the certificate is not a CA certificate', 2)
|
|
write(folder / 'cert.pem', cert_pem(cert), 0o644)
|
|
remove(folder / 'csr.pem')
|
|
state = read_state()
|
|
record(state, cert, 'root', 'ca')
|
|
sign_crl(cfg, name, state)
|
|
write_state(state)
|
|
publish()
|
|
return {'name': name, 'subject': common_name(cert), 'expires': iso(utc(cert, 'not_valid_after'))}
|
|
|
|
|
|
def import_root_crl(data):
|
|
"""Take a CRL of the root signed offline, in PEM or DER; it must be signed by the root and not older than the
|
|
current one."""
|
|
if isinstance(data, str):
|
|
data = data.strip().encode()
|
|
try:
|
|
crl = x509.load_pem_x509_crl(data) if data.startswith(b'-----') else x509.load_der_x509_crl(data)
|
|
except ValueError:
|
|
raise Failure('the CRL is not a PEM or DER certificate revocation list', 2)
|
|
root = read_cert(ROOT_DIR / 'cert.pem')
|
|
if root is None:
|
|
raise Failure('there is no root CA', 3)
|
|
if crl.issuer != root.subject or not crl.is_signature_valid(root.public_key()):
|
|
raise Failure('the CRL is not signed by the root CA', 2)
|
|
current = read_crl(ROOT_DIR / 'crl.pem')
|
|
if current is not None and utc(crl, 'last_update') < utc(current, 'last_update'):
|
|
raise Failure('the CRL is older than the current one', 2)
|
|
write(ROOT_DIR / 'crl.pem', crl.public_bytes(serialization.Encoding.PEM), 0o644)
|
|
state = read_state()
|
|
state['root_crl'] = 'uploaded'
|
|
write_state(state)
|
|
publish()
|
|
reload_crls()
|
|
following = utc(crl, 'next_update')
|
|
return {'next_update': iso(following) if following else None}
|
|
|
|
|
|
def import_root_cert(data):
|
|
"""Place an existing root CA's certificate, PEM, before the root exists: without its key the root is offline."""
|
|
if read_cert(ROOT_DIR / 'cert.pem') is not None or (ROOT_DIR / 'key.pem').exists():
|
|
raise Failure('the root CA exists', 4)
|
|
try:
|
|
cert = x509.load_pem_x509_certificate(data.strip() if isinstance(data, bytes) else data.strip().encode())
|
|
except ValueError:
|
|
raise Failure('the certificate is not a PEM certificate', 2)
|
|
if not is_ca(cert):
|
|
raise Failure('the certificate is not a CA certificate', 2)
|
|
write(ROOT_DIR / 'cert.pem', cert_pem(cert), 0o644)
|
|
return {'subject': common_name(cert), 'expires': iso(utc(cert, 'not_valid_after'))}
|
|
|
|
|
|
def export_root_key(passphrase):
|
|
"""The root key in PEM, encrypted with the passphrase."""
|
|
key = read_key(ROOT_DIR / 'key.pem')
|
|
if key is None:
|
|
raise Failure('the root key is not on the router', 3)
|
|
if not passphrase:
|
|
raise Failure('the passphrase is required', 2)
|
|
return key_pem(key, passphrase.encode())
|
|
|
|
|
|
def remove_root_key():
|
|
if not remove(ROOT_DIR / 'key.pem'):
|
|
raise Failure('the root key is not on the router', 3)
|
|
|
|
|
|
def import_root_key(data, passphrase=None):
|
|
"""Put the root key back, PEM, encrypted with the passphrase or not; it must be the key of the root certificate."""
|
|
root = read_cert(ROOT_DIR / 'cert.pem')
|
|
if root is None:
|
|
raise Failure('there is no root CA', 3)
|
|
try:
|
|
key = serialization.load_pem_private_key(data, passphrase.encode() if passphrase else None)
|
|
except (ValueError, TypeError):
|
|
raise Failure('the file is not a PEM private key, or the passphrase is wrong', 2)
|
|
if not same_key(key.public_key(), root.public_key()):
|
|
raise Failure('the key is not the key of the root CA', 2)
|
|
write(ROOT_DIR / 'key.pem', key_pem(key), 0o600)
|
|
|
|
|
|
def maintain(cfg):
|
|
"""Sign the stale CRLs again, publish and install them; True when one of the router's certificates is due for
|
|
renewal, which vpn-router-setup does."""
|
|
state = read_state()
|
|
refresh_crls(cfg, state)
|
|
write_state(state)
|
|
publish()
|
|
reload_crls()
|
|
for name in ROUTER_CERTS:
|
|
cert = read_cert(ROUTER_DIR / f'{name}_cert.pem')
|
|
if cert is not None and renewal_due(cert):
|
|
return True
|
|
return False
|
|
|
|
|
|
def acme_url(cfg, name):
|
|
"""The ACME directory of an issuing CA served by Caddy's ACME server."""
|
|
return f'https://{cfg["wan"]["local_fqdn"].strip()}:{cfg["https"]["web_port"]}/acme/{name}/directory'
|
|
|
|
|
|
def status(cfg):
|
|
"""The root, the issuing CAs, the retired CAs and the warnings, as /ca and vpn-router-ca status show them."""
|
|
moment = now()
|
|
soon = moment + datetime.timedelta(days=WARN_DAYS)
|
|
state = read_state()
|
|
warnings = []
|
|
|
|
def crl_of(name):
|
|
crl = read_crl(directory(name) / 'crl.pem')
|
|
following = utc(crl, 'next_update') if crl is not None else None
|
|
if crl is None:
|
|
warnings.append(f'{name} has no CRL')
|
|
elif following is not None and following <= soon:
|
|
warnings.append(f'the CRL of {name} {"expired" if following <= moment else "expires"} on {iso(following)}')
|
|
return iso(following) if following else None
|
|
|
|
root_cert = read_cert(ROOT_DIR / 'cert.pem')
|
|
root = None
|
|
if root_cert is not None:
|
|
key = (ROOT_DIR / 'key.pem').is_file()
|
|
root = {'subject': common_name(root_cert), 'expires': iso(utc(root_cert, 'not_valid_after')), 'key': key,
|
|
'crl_next_update': crl_of('root') if key or (ROOT_DIR / 'crl.pem').is_file() else None,
|
|
'crl_source': state['root_crl']}
|
|
if key:
|
|
warnings.append('the root key is on the router; export it with vpn-router-ca root export-key and remove it '
|
|
'with vpn-router-ca root remove-key to keep the root offline')
|
|
elif not (ROOT_DIR / 'crl.pem').is_file():
|
|
warnings.append('the root has no CRL; sign one offline and upload it with vpn-router-ca root import-crl')
|
|
if utc(root_cert, 'not_valid_after') <= soon:
|
|
warnings.append(f'the root certificate expires on {root["expires"]}')
|
|
|
|
authorities = []
|
|
for name, authority in vpnrouter.issuing_cas(cfg).items():
|
|
folder = CA_DIR / name
|
|
cert = read_cert(folder / 'cert.pem')
|
|
kind = 'active' if cert is not None else 'pending' if (folder / 'csr.pem').is_file() else 'missing'
|
|
entry = {'name': name, 'common_name': authority['name'], 'state': kind, 'builtin': authority['builtin'],
|
|
'templates': authority['templates'], 'acme': authority['acme'],
|
|
'subject': common_name(cert) if cert is not None else None,
|
|
'expires': iso(utc(cert, 'not_valid_after')) if cert is not None else None,
|
|
'key': (folder / 'key.pem').is_file(), 'crl_next_update': crl_of(name) if cert is not None else None,
|
|
'acme_url': acme_url(cfg, name) if authority['acme'] and cert is not None else None}
|
|
if kind == 'pending':
|
|
warnings.append(f'issuing CA {name} waits for its certificate signed by the root CA')
|
|
if cert is not None and utc(cert, 'not_valid_after') <= soon:
|
|
warnings.append(f'the certificate of issuing CA {name} expires on {entry["expires"]}')
|
|
authorities.append(entry)
|
|
|
|
root_crl = read_crl(ROOT_DIR / 'crl.pem')
|
|
for retired in state['retired']:
|
|
if retired['serial'] and root_crl is not None and \
|
|
root_crl.get_revoked_certificate_by_serial_number(int(retired['serial'], 16)) is None:
|
|
warnings.append(f'the root CRL does not revoke retired issuing CA {retired["name"]}')
|
|
for name in ROUTER_CERTS:
|
|
cert = read_cert(ROUTER_DIR / f'{name}_cert.pem')
|
|
if cert is not None and utc(cert, 'not_valid_after') <= soon:
|
|
warnings.append(f'the router certificate {name} expires on {iso(utc(cert, "not_valid_after"))}')
|
|
return {'root': root, 'authorities': authorities, 'retired': state['retired'], 'warnings': warnings}
|