1155 lines
57 KiB
Python
1155 lines
57 KiB
Python
"""The router's PKI, for vpn-router-setup, the web service, vpn-router-ca and the user module.
|
|
|
|
The root CA 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. The PKI is kept in ou=pki of the router's directory (vpnrouter_ldap), which root reaches through
|
|
the local socket. Each CA is an entry cn=<name> of class vpnRouterAuthority, with its certificate, key, CRL and CRL
|
|
number; while the root key is not on the router, a new issuing CA has its key and a CSR and is pending until import_ca
|
|
takes its certificate, signed offline. Each certificate a CA issued is an entry serialNumber=<serial> of class
|
|
vpnRouterCertificate under the CA, with its kind, template, names, expiry and revocation; the router's own certificates
|
|
and the users' keep their keys there, as the router and the OpenVPN profile need them, and a certificate issued for
|
|
anything else is kept without its key. A user's certificate is also the userCertificate of the user's entry. A retired
|
|
CA keeps its entry, renamed cn=<name>-<time> and marked with vpnRouterRetired.
|
|
|
|
An existing root CA is supplied by placing its certificate, and its key if the router may use it, in SEED_DIR before
|
|
vpn-router-setup runs; provision imports them into the directory and removes the files.
|
|
|
|
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 writes the files strongSwan, OpenVPN and the system's trust
|
|
store 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 os
|
|
import pathlib
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
|
|
import ldap
|
|
import ldap.dn
|
|
import ldap.filter
|
|
import ldap.modlist
|
|
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, NameOID
|
|
|
|
import vpnrouter
|
|
import vpnrouter_ldap
|
|
|
|
#: The certificate (cert.pem) and the key (key.pem) of an existing root CA, which provision imports and removes.
|
|
SEED_DIR = pathlib.Path('/etc/vpn-router/pki/root')
|
|
|
|
#: 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, and the files strongSwan reads.
|
|
ROUTER_CERTS = {'vpn-server': ('p2s', 'vpn-server'), 'site-to-site': ('s2s', 'site-to-site')}
|
|
ROUTER_FILES = {'vpn-server': (VPN_SERVER_CERT, VPN_SERVER_KEY), 'site-to-site': (S2S_CERT, S2S_KEY)}
|
|
|
|
#: The attributes of a CA's entry and of a certificate's entry.
|
|
AUTHORITY_ATTRIBUTES = ['cACertificate;binary', 'certificateRevocationList;binary', 'vpnRouterPrivateKey',
|
|
'vpnRouterCertificateRequest', 'vpnRouterCRLNumber', 'vpnRouterCRLSource', 'vpnRouterRetired']
|
|
CERTIFICATE_ATTRIBUTES = ['serialNumber', 'cn', 'vpnRouterKind', 'vpnRouterTemplate', 'vpnRouterSubjectAltName',
|
|
'vpnRouterNotAfter', 'vpnRouterRevoked', 'vpnRouterRevocationReason', 'createTimestamp']
|
|
|
|
#: The suffix of a retired CA's name: -<time of the retirement>.
|
|
RETIRED_SUFFIX = re.compile(r'-\d{14}$')
|
|
|
|
#: 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
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def directory():
|
|
"""A connection to the router's directory as root, closed at the end."""
|
|
connection = vpnrouter_ldap.root()
|
|
try:
|
|
yield connection
|
|
finally:
|
|
connection.unbind_s()
|
|
|
|
|
|
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 generalized(moment):
|
|
"""A time as LDAP's GeneralizedTime, in UTC."""
|
|
return moment.astimezone(datetime.timezone.utc).strftime('%Y%m%d%H%M%SZ').encode()
|
|
|
|
|
|
def from_generalized(value):
|
|
"""A GeneralizedTime value as an ISO 8601 UTC string, None for none."""
|
|
if not value:
|
|
return None
|
|
return iso(datetime.datetime.strptime(value[:14].decode(), '%Y%m%d%H%M%S').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 cert_pem(cert):
|
|
return cert.public_bytes(serialization.Encoding.PEM)
|
|
|
|
|
|
def cert_der(cert):
|
|
return cert.public_bytes(serialization.Encoding.DER)
|
|
|
|
|
|
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=0o755, 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 first(attributes, name):
|
|
values = attributes.get(name)
|
|
return values[0] if values else None
|
|
|
|
|
|
def authority_dn(name):
|
|
return f'cn={ldap.dn.escape_dn_chars(name)},{vpnrouter_ldap.PKI}'
|
|
|
|
|
|
def authority(connection, name):
|
|
"""The entry of a CA: {name, cert, key, csr (PEM), crl, crl_number, crl_source, retired}, None when there is none."""
|
|
try:
|
|
attributes = connection.search_s(authority_dn(name), ldap.SCOPE_BASE, attrlist=AUTHORITY_ATTRIBUTES)[0][1]
|
|
except ldap.NO_SUCH_OBJECT:
|
|
return None
|
|
cert, crl, key = (first(attributes, item) for item in AUTHORITY_ATTRIBUTES[:3])
|
|
return {
|
|
'name': name,
|
|
'cert': x509.load_der_x509_certificate(cert) if cert else None,
|
|
'key': serialization.load_pem_private_key(key, None) if key else None,
|
|
'csr': first(attributes, 'vpnRouterCertificateRequest'),
|
|
'crl': x509.load_der_x509_crl(crl) if crl else None,
|
|
'crl_number': int(first(attributes, 'vpnRouterCRLNumber') or 0),
|
|
'crl_source': (first(attributes, 'vpnRouterCRLSource') or b'').decode() or None,
|
|
'retired': from_generalized(first(attributes, 'vpnRouterRetired')),
|
|
}
|
|
|
|
|
|
def update(connection, name, values):
|
|
"""Set attributes of a CA's entry, adding the entry when it is missing; a value None removes its attribute."""
|
|
dn = authority_dn(name)
|
|
try:
|
|
connection.search_s(dn, ldap.SCOPE_BASE, attrlist=['1.1'])
|
|
except ldap.NO_SUCH_OBJECT:
|
|
connection.add_s(dn, ldap.modlist.addModlist({
|
|
'objectClass': [b'vpnRouterAuthority'], 'cn': [name.encode()],
|
|
**{attribute: [value] for attribute, value in values.items() if value is not None}}))
|
|
return
|
|
connection.modify_s(dn, [(ldap.MOD_REPLACE, attribute, [value] if value is not None else [])
|
|
for attribute, value in values.items()])
|
|
|
|
|
|
def issuing(connection):
|
|
"""The names of the issuing CAs in the directory that are not retired, sorted."""
|
|
found = connection.search_s(vpnrouter_ldap.PKI, ldap.SCOPE_ONELEVEL,
|
|
'(&(objectClass=vpnRouterAuthority)(!(vpnRouterRetired=*))(!(cn=root)))', ['cn'])
|
|
return sorted(attributes['cn'][0].decode() for _, attributes in found)
|
|
|
|
|
|
def certificate_entry(dn, attributes):
|
|
"""A certificate's entry as the answers show it: serial, ca, kind, template, subject, sans, created, expires,
|
|
revoked and reason."""
|
|
def value(name):
|
|
return (first(attributes, name) or b'').decode() or None
|
|
|
|
return {
|
|
'serial': value('serialNumber'), 'ca': ldap.dn.str2dn(dn)[1][0][1], 'kind': value('vpnRouterKind'),
|
|
'template': value('vpnRouterTemplate'), 'subject': value('cn') or '',
|
|
'sans': [item.decode() for item in attributes.get('vpnRouterSubjectAltName', [])],
|
|
'created': from_generalized(first(attributes, 'createTimestamp')),
|
|
'expires': from_generalized(first(attributes, 'vpnRouterNotAfter')),
|
|
'revoked': from_generalized(first(attributes, 'vpnRouterRevoked')), 'reason': value('vpnRouterRevocationReason'),
|
|
'dn': dn,
|
|
}
|
|
|
|
|
|
def certificate_entries(connection, base, condition=''):
|
|
"""The certificates' entries under base that match the LDAP condition, newest first."""
|
|
found = connection.search_s(base, ldap.SCOPE_SUBTREE, f'(&(objectClass=vpnRouterCertificate){condition})',
|
|
CERTIFICATE_ATTRIBUTES)
|
|
return sorted((certificate_entry(dn, attributes) for dn, attributes in found),
|
|
key=lambda entry: entry['created'] or '', reverse=True)
|
|
|
|
|
|
def record(connection, cert, ca, kind, template=None, key=None):
|
|
"""Add a certificate the CA issued: kind is ca, router, user or external; key is kept for router and user."""
|
|
serial = f'{cert.serial_number:X}'
|
|
attributes = {
|
|
'objectClass': [b'vpnRouterCertificate'], 'serialNumber': [serial.encode()], 'cn': [common_name(cert).encode()],
|
|
'userCertificate;binary': [cert_der(cert)], 'vpnRouterKind': [kind.encode()],
|
|
'vpnRouterNotAfter': [generalized(utc(cert, 'not_valid_after'))],
|
|
}
|
|
if template:
|
|
attributes['vpnRouterTemplate'] = [template.encode()]
|
|
if san_values(cert):
|
|
attributes['vpnRouterSubjectAltName'] = [value.encode() for value in san_values(cert)]
|
|
if key is not None:
|
|
attributes['vpnRouterPrivateKey'] = [key_pem(key)]
|
|
connection.add_s(f'serialNumber={serial},{authority_dn(ca)}', ldap.modlist.addModlist(attributes))
|
|
|
|
|
|
def holders(connection, ca, condition):
|
|
"""The certificates of the CA that match the LDAP condition and have their key, not revoked, newest first: [(dn,
|
|
cert, key)]."""
|
|
found = connection.search_s(
|
|
authority_dn(ca), ldap.SCOPE_ONELEVEL,
|
|
f'(&(objectClass=vpnRouterCertificate)(vpnRouterPrivateKey=*)(!(vpnRouterRevoked=*)){condition})',
|
|
['userCertificate;binary', 'vpnRouterPrivateKey'])
|
|
pairs = [(dn, x509.load_der_x509_certificate(attributes['userCertificate;binary'][0]),
|
|
serialization.load_pem_private_key(attributes['vpnRouterPrivateKey'][0], None)) for dn, attributes in found]
|
|
return sorted(pairs, key=lambda pair: utc(pair[1], 'not_valid_after'), reverse=True)
|
|
|
|
|
|
def router_condition(name):
|
|
return f'(vpnRouterKind=router)(vpnRouterTemplate={ROUTER_CERTS[name][1]})'
|
|
|
|
|
|
def user_condition(name):
|
|
return f'(vpnRouterKind=user)(cn={ldap.filter.escape_filter_chars(name)})'
|
|
|
|
|
|
def drop_key(connection, dn):
|
|
connection.modify_s(dn, [(ldap.MOD_REPLACE, 'vpnRouterPrivateKey', [])])
|
|
|
|
|
|
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(one, other):
|
|
der = serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo
|
|
return one.public_bytes(*der) == other.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(cfg, template, names):
|
|
"""The extensions of a certificate of the template with the subject alternative names: the extended key usages by
|
|
their OIDs (vpnrouter.extended_key_usages), and the template's policies as certificate policies."""
|
|
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']:
|
|
oids = vpnrouter.extended_key_usages(cfg)
|
|
extensions.append((x509.ExtendedKeyUsage([x509.ObjectIdentifier(oids[item])
|
|
for item in template['extended_key_usage']]), False))
|
|
if template['policies']:
|
|
extensions.append((x509.CertificatePolicies([x509.PolicyInformation(x509.ObjectIdentifier(oid), None)
|
|
for oid in template['policies']]), 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 sign_crl(connection, cfg, name):
|
|
"""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."""
|
|
entry = authority(connection, name)
|
|
cert, key = entry['cert'], entry['key']
|
|
number = entry['crl_number'] + 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 revoked_entry in certificate_entries(connection, authority_dn(name), '(vpnRouterRevoked=*)'):
|
|
if revoked_entry['expires'] and revoked_entry['expires'] <= iso(moment):
|
|
continue
|
|
revoked = (x509.RevokedCertificateBuilder()
|
|
.serial_number(int(revoked_entry['serial'], 16))
|
|
.revocation_date(datetime.datetime.strptime(revoked_entry['revoked'], '%Y-%m-%dT%H:%M:%SZ')
|
|
.replace(tzinfo=datetime.timezone.utc)))
|
|
if revoked_entry['reason'] and revoked_entry['reason'] != 'unspecified':
|
|
revoked = revoked.add_extension(x509.CRLReason(x509.ReasonFlags[revoked_entry['reason']]), False)
|
|
builder = builder.add_revoked_certificate(revoked.build())
|
|
crl = builder.sign(key, digest(key))
|
|
update(connection, name, {'certificateRevocationList;binary': crl.public_bytes(serialization.Encoding.DER),
|
|
'vpnRouterCRLNumber': str(number).encode(),
|
|
**({'vpnRouterCRLSource': b'signed'} if name == 'root' else {})})
|
|
|
|
|
|
def crl_stale(crl):
|
|
"""Whether the CRL is missing or past half of its validity."""
|
|
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 ready(name):
|
|
"""Whether the issuing CA has its certificate and key, and so issues certificates."""
|
|
with directory() as connection:
|
|
entry = authority(connection, name)
|
|
return bool(entry and entry['cert'] and entry['key'] and not entry['retired'])
|
|
|
|
|
|
def refresh_crls(connection, cfg, 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', *issuing(connection)]:
|
|
entry = authority(connection, name)
|
|
if entry and entry['key'] and entry['cert'] and (name in force or crl_stale(entry['crl'])):
|
|
sign_crl(connection, cfg, name)
|
|
|
|
|
|
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 is_address(value):
|
|
try:
|
|
ipaddress.ip_address(value)
|
|
except ValueError:
|
|
return False
|
|
return True
|
|
|
|
|
|
def router_identities(name, ctx):
|
|
"""The common name and the subject alternative names of one of the router's certificates (ROUTER_CERTS): the VPN
|
|
server certificate names wan.local_fqdn, and the site-to-site certificate also the local identity of each
|
|
site_to_site connection with certificate authentication, the public or the internal address of its
|
|
local_id_mode; None while such an identity is not known yet."""
|
|
fqdn = ctx['local_fqdn']
|
|
if name == 'vpn-server':
|
|
return fqdn, [fqdn]
|
|
names = [fqdn]
|
|
for connection in ctx['connections']:
|
|
if connection['auth'] != 'certificate' or connection['local_id_mode'] == 'fqdn':
|
|
continue
|
|
if not connection['local_id']:
|
|
return None
|
|
if connection['local_id'] not in names:
|
|
names.append(connection['local_id'])
|
|
return fqdn, names
|
|
|
|
|
|
def ensure_router_cert(connection, cfg, ctx, name, urls):
|
|
"""Issue one of the router's certificates when it has none, or its current one is due for renewal, not issued by
|
|
its CA's current key, or named otherwise than router_identities says; the older ones lose their keys. Nothing is
|
|
issued while an identity it needs is not known yet."""
|
|
ca, template_name = ROUTER_CERTS[name]
|
|
issuer = authority(connection, ca)
|
|
identities = router_identities(name, ctx)
|
|
if not issuer or issuer['cert'] is None or issuer['key'] is None or identities is None or not identities[0]:
|
|
return
|
|
subject, sans = identities
|
|
current = holders(connection, ca, router_condition(name))
|
|
wanted = {f'IP:{value}' if is_address(value) else f'DNS:{value.lower()}' for value in sans}
|
|
if current:
|
|
_, cert, _ = current[0]
|
|
if not renewal_due(cert) and issued_by(cert, issuer['cert']) and common_name(cert) == subject and \
|
|
wanted == set(san_values(cert)):
|
|
return
|
|
template = vpnrouter.pki_templates(cfg)[template_name]
|
|
key = new_key(template['key'])
|
|
cert = sign(subject, key.public_key(), issuer['cert'], issuer['key'], template['days'],
|
|
leaf_extensions(cfg, template, general_names(sans, vpnrouter.SAN_TYPES)), urls, ca)
|
|
record(connection, cert, ca, 'router', template_name, key)
|
|
for dn, _, _ in current:
|
|
drop_key(connection, dn)
|
|
|
|
|
|
def import_seed(connection):
|
|
"""Import the root CA of SEED_DIR into the directory, and remove the files; False when there is none."""
|
|
cert_file, key_file = SEED_DIR / 'cert.pem', SEED_DIR / 'key.pem'
|
|
if not cert_file.is_file():
|
|
if key_file.is_file():
|
|
raise Failure(f'{key_file} has no certificate; place cert.pem beside it')
|
|
return False
|
|
cert = x509.load_pem_x509_certificate(cert_file.read_bytes())
|
|
key = serialization.load_pem_private_key(key_file.read_bytes(), None) if key_file.is_file() else None
|
|
if not is_ca(cert):
|
|
raise Failure(f'{cert_file} is not a CA certificate')
|
|
if key is not None and not same_key(key.public_key(), cert.public_key()):
|
|
raise Failure(f'{key_file} is not the key of {cert_file}')
|
|
update(connection, 'root', {'cACertificate;binary': cert_der(cert), 'vpnRouterPrivateKey': key_pem(key) if key else None})
|
|
for path in (key_file, cert_file):
|
|
remove(path)
|
|
for folder in (SEED_DIR, SEED_DIR.parent):
|
|
with contextlib.suppress(OSError):
|
|
folder.rmdir()
|
|
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 imported from SEED_DIR, or created when it has neither certificate nor seed. 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, and its certificate 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.
|
|
"""
|
|
pki = cfg['pki']
|
|
urls = vpnrouter.pki_base_urls(cfg)
|
|
notices = []
|
|
with directory() as connection:
|
|
root = authority(connection, 'root')
|
|
if root is None or root['cert'] is None:
|
|
if root is not None and root['key'] is not None:
|
|
raise Failure('the root has a key but no certificate')
|
|
if not import_seed(connection):
|
|
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))
|
|
update(connection, 'root', {'cACertificate;binary': cert_der(root_cert), 'vpnRouterPrivateKey': key_pem(root_key)})
|
|
root = authority(connection, 'root')
|
|
root_cert, root_key = root['cert'], root['key']
|
|
|
|
authorities = vpnrouter.issuing_cas(cfg)
|
|
force = set()
|
|
for name, settings in authorities.items():
|
|
entry = authority(connection, name)
|
|
if entry is not None and entry['cert'] is not None:
|
|
continue
|
|
key = entry['key'] if entry is not None and entry['key'] is not None else None
|
|
if key is None:
|
|
key = new_key(pki['key'])
|
|
update(connection, name, {'vpnRouterPrivateKey': key_pem(key)})
|
|
if root_key is not None:
|
|
cert = sign(settings['name'].strip(), key.public_key(), root_cert, root_key, pki['ca_days'],
|
|
ca_extensions(0), urls, 'root')
|
|
update(connection, name, {'cACertificate;binary': cert_der(cert), 'vpnRouterCertificateRequest': None})
|
|
record(connection, cert, 'root', 'ca')
|
|
force.add(name)
|
|
continue
|
|
if entry is None or entry['csr'] is None:
|
|
request = (x509.CertificateSigningRequestBuilder()
|
|
.subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, settings['name'].strip())]))
|
|
.add_extension(x509.BasicConstraints(ca=True, path_length=0), True)
|
|
.add_extension(CA_USAGE, True)
|
|
.sign(key, digest(key)))
|
|
update(connection, name, {'vpnRouterCertificateRequest': request.public_bytes(serialization.Encoding.PEM)})
|
|
notices.append(f'issuing CA {name} waits for its certificate: sign its CSR (vpn-router-ca ca csr {name}) with '
|
|
f'the root CA and import it with vpn-router-ca ca import {name}')
|
|
|
|
for name in issuing(connection):
|
|
if name in authorities:
|
|
continue
|
|
entry = authority(connection, name)
|
|
moment = now()
|
|
if entry['cert'] is not None:
|
|
serial = f'{entry["cert"].serial_number:X}'
|
|
for revoked in certificate_entries(connection, authority_dn('root'),
|
|
f'(serialNumber={serial})(!(vpnRouterRevoked=*))'):
|
|
connection.modify_s(revoked['dn'], [(ldap.MOD_REPLACE, 'vpnRouterRevoked', [generalized(moment)]),
|
|
(ldap.MOD_REPLACE, 'vpnRouterRevocationReason', [b'cessation_of_operation'])])
|
|
force.add('root')
|
|
if root_key is None:
|
|
notices.append(f'issuing CA {name} was retired: revoke its certificate on the root CRL offline and '
|
|
'upload the CRL with vpn-router-ca root import-crl')
|
|
retired = f'{name}-{moment.strftime("%Y%m%d%H%M%S")}'
|
|
connection.rename_s(authority_dn(name), f'cn={retired}')
|
|
update(connection, retired, {'vpnRouterRetired': generalized(moment), 'vpnRouterPrivateKey': None})
|
|
|
|
refresh_crls(connection, cfg, force)
|
|
for name in ROUTER_CERTS:
|
|
ensure_router_cert(connection, cfg, ctx, name, urls)
|
|
publish(connection)
|
|
return notices
|
|
|
|
|
|
def publish(connection):
|
|
"""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', *issuing(connection)]:
|
|
entry = authority(connection, name)
|
|
if entry is not None and entry['cert'] is not None:
|
|
wanted[f'{name}.crt'] = cert_der(entry['cert'])
|
|
if entry is not None and entry['crl'] is not None:
|
|
wanted[f'{name}.crl'] = entry['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(connection):
|
|
"""Write the CRLs of the root, p2s and s2s for strongSwan, and those of the root and p2s for OpenVPN; True on
|
|
change."""
|
|
changed = False
|
|
crls = {}
|
|
for name, file in SWANCTL_CA_FILES.items():
|
|
entry = authority(connection, name)
|
|
crls[name] = entry['crl'].public_bytes(serialization.Encoding.PEM) if entry and entry['crl'] is not None else None
|
|
changed |= write(X509CRL_DIR / file, crls[name], 0o644) if crls[name] else remove(X509CRL_DIR / file)
|
|
if crls['root'] and crls['p2s']:
|
|
changed |= write(OPENVPN_CRL, crls['root'] + crls['p2s'], 0o644)
|
|
return changed
|
|
|
|
|
|
def reload_crls(connection):
|
|
"""Install the CRLs, and have strongSwan load them when they changed; OpenVPN reads its file when it changes."""
|
|
if install_crls(connection):
|
|
subprocess.run(['swanctl', '--load-creds'], capture_output=True, text=True)
|
|
|
|
|
|
def distribute(cfg):
|
|
"""Write 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
|
|
with directory() as connection:
|
|
certs = {}
|
|
for name, file in SWANCTL_CA_FILES.items():
|
|
entry = authority(connection, name)
|
|
certs[name] = cert_pem(entry['cert']) if entry and entry['cert'] else None
|
|
changed |= write(X509CA_DIR / file, certs[name], 0o644) if certs[name] else remove(X509CA_DIR / file)
|
|
remote = {remote_ca_file(name): connection_settings['remote_ca'].strip().encode() + b'\n'
|
|
for name, connection_settings in cfg['site_to_site'].items() if connection_settings['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_file, key_file) in ROUTER_FILES.items():
|
|
current = holders(connection, ROUTER_CERTS[name][0], router_condition(name))
|
|
if current:
|
|
_, cert, key = current[0]
|
|
changed |= write(cert_file, cert_pem(cert), 0o644)
|
|
changed |= write(key_file, key_pem(key), 0o600)
|
|
else:
|
|
changed |= remove(cert_file) | remove(key_file)
|
|
if certs['root'] and certs['p2s']:
|
|
changed |= write(OPENVPN_CA, certs['root'] + certs['p2s'], 0o644)
|
|
changed |= install_crls(connection)
|
|
if certs['root'] and write(TRUST_ANCHOR, certs['root'], 0o644):
|
|
changed = True
|
|
if shutil.which('update-ca-certificates'):
|
|
subprocess.run(['update-ca-certificates'], capture_output=True, text=True)
|
|
return changed
|
|
|
|
|
|
def bundle(name):
|
|
"""The PEM certificates a client trusts a CA with: {name, pem}; for root the root alone, for an issuing CA the
|
|
issuing CA followed by the root."""
|
|
with directory() as connection:
|
|
root = authority(connection, 'root')
|
|
entry = root if name == 'root' else authority(connection, name) if isinstance(name, str) else None
|
|
if root is None or root['cert'] is None:
|
|
raise Failure('there is no root CA', 3)
|
|
if entry is None or entry['retired']:
|
|
raise Failure(f'no CA {name}', 3)
|
|
if entry['cert'] is None:
|
|
raise Failure(f'issuing CA {name} has no certificate yet', 4)
|
|
pem = cert_pem(root['cert']) if name == 'root' else cert_pem(entry['cert']) + cert_pem(root['cert'])
|
|
return {'name': name, 'pem': pem.decode()}
|
|
|
|
|
|
def authority_files(name):
|
|
"""The PEM certificate and key of an issuing CA, for Caddy's ACME server; None when it has not both."""
|
|
with directory() as connection:
|
|
entry = authority(connection, name)
|
|
if not entry or entry['cert'] is None or entry['key'] is None:
|
|
return None
|
|
return cert_pem(entry['cert']), key_pem(entry['key'])
|
|
|
|
|
|
def issuer_of(connection, cfg, ca, template):
|
|
"""The entry 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)
|
|
entry = authority(connection, ca)
|
|
if not entry or entry['cert'] is None or entry['key'] is None:
|
|
raise Failure(f'issuing CA {ca} has no certificate yet', 4)
|
|
return entry, 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)
|
|
with directory() as connection:
|
|
issuer, values = issuer_of(connection, 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['cert'], issuer['key'], values['days'], leaf_extensions(cfg, values, names),
|
|
vpnrouter.pki_base_urls(cfg), ca)
|
|
record(connection, cert, ca, 'external', template)
|
|
root = authority(connection, 'root')['cert']
|
|
serial = f'{cert.serial_number:X}'
|
|
if key is None:
|
|
return {'serial': serial, 'certificate': cert_pem(cert).decode(),
|
|
'chain': (cert_pem(issuer['cert']) + cert_pem(root)).decode()}
|
|
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['cert'], root], encryption)
|
|
return {'serial': serial, 'pfx': base64.b64encode(pfx).decode()}
|
|
|
|
|
|
def user_dn(name):
|
|
return f'uid={ldap.dn.escape_dn_chars(name)},{vpnrouter_ldap.USERS}'
|
|
|
|
|
|
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; it is kept with its key, and is the userCertificate of the user's entry."""
|
|
with directory() as connection:
|
|
issuer, values = issuer_of(connection, cfg, 'p2s', 'vpn-user')
|
|
key = new_key(values['key'])
|
|
cert = sign(name, key.public_key(), issuer['cert'], issuer['key'], values['days'],
|
|
leaf_extensions(cfg, values, general_names([name, email], vpnrouter.SAN_TYPES)),
|
|
vpnrouter.pki_base_urls(cfg), 'p2s')
|
|
for dn, _, _ in holders(connection, 'p2s', user_condition(name)):
|
|
drop_key(connection, dn)
|
|
record(connection, cert, 'p2s', 'user', 'vpn-user', key)
|
|
connection.modify_s(user_dn(name), [(ldap.MOD_REPLACE, 'userCertificate;binary', [cert_der(cert)])])
|
|
|
|
|
|
def user_credentials(name):
|
|
"""The PEM certificate and key of the user's current certificate, None when the user has none."""
|
|
with directory() as connection:
|
|
current = holders(connection, 'p2s', user_condition(name))
|
|
if not current:
|
|
return None
|
|
_, cert, key = current[0]
|
|
return cert_pem(cert), key_pem(key)
|
|
|
|
|
|
def revoke_entries(connection, cfg, entries, reason):
|
|
"""Mark the certificates revoked, sign new CRLs of their CAs, and publish and install them."""
|
|
names = {entry['ca'] for entry in entries}
|
|
for name in names:
|
|
current = authority(connection, name)
|
|
if not current or current['key'] is None:
|
|
raise Failure(f'the key of {name} is not on the router, so it cannot sign its CRL', 4)
|
|
moment = generalized(now())
|
|
for entry in entries:
|
|
connection.modify_s(entry['dn'], [(ldap.MOD_REPLACE, 'vpnRouterRevoked', [moment]),
|
|
(ldap.MOD_REPLACE, 'vpnRouterRevocationReason', [reason.encode()]),
|
|
(ldap.MOD_REPLACE, 'vpnRouterPrivateKey', [])])
|
|
for name in names:
|
|
sign_crl(connection, cfg, name)
|
|
publish(connection)
|
|
reload_crls(connection)
|
|
|
|
|
|
def revoke_user(cfg, name):
|
|
"""Revoke the user's certificates, and remove the userCertificate of the user's entry."""
|
|
with directory() as connection:
|
|
entries = certificate_entries(connection, authority_dn('p2s'), f'{user_condition(name)}(!(vpnRouterRevoked=*))')
|
|
if entries:
|
|
revoke_entries(connection, cfg, entries, 'cessation_of_operation')
|
|
with contextlib.suppress(ldap.NO_SUCH_OBJECT):
|
|
connection.modify_s(user_dn(name), [(ldap.MOD_REPLACE, 'userCertificate;binary', [])])
|
|
|
|
|
|
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 loses its key, 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)
|
|
with directory() as connection:
|
|
found = certificate_entries(connection, vpnrouter_ldap.PKI, f'(serialNumber={serial.upper().lstrip("0")})')
|
|
if not found:
|
|
raise Failure(f'no certificate with the serial number {serial}', 3)
|
|
entry = found[0]
|
|
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(connection, cfg, [entry], reason)
|
|
return {**{key: value for key, value in entry.items() if key != 'dn'}, 'revoked': iso(now()), 'reason': reason,
|
|
'reissue': entry['kind'] == 'router'}
|
|
|
|
|
|
def certificates():
|
|
"""The certificates the router issued, newest first, each with state: valid, expired or revoked."""
|
|
moment = iso(now())
|
|
with directory() as connection:
|
|
entries = certificate_entries(connection, vpnrouter_ldap.PKI)
|
|
return [{**{key: value for key, value in entry.items() if key != 'dn'},
|
|
'ca': RETIRED_SUFFIX.sub('', entry['ca']),
|
|
'state': 'revoked' if entry['revoked'] else 'expired' if (entry['expires'] or '') <= moment else 'valid'}
|
|
for entry in entries]
|
|
|
|
|
|
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)
|
|
with directory() as connection:
|
|
entry = authority(connection, name)
|
|
if not entry or entry['cert'] is not None or entry['csr'] is None:
|
|
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': entry['csr'].decode(), '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)
|
|
with directory() as connection:
|
|
entry = authority(connection, name)
|
|
if entry is not None and entry['cert'] is not None:
|
|
raise Failure(f'issuing CA {name} has its certificate', 4)
|
|
if entry is None or entry['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, authority(connection, 'root')['cert']):
|
|
raise Failure('the certificate is not signed by the root CA', 2)
|
|
if not same_key(cert.public_key(), entry['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)
|
|
update(connection, name, {'cACertificate;binary': cert_der(cert), 'vpnRouterCertificateRequest': None})
|
|
record(connection, cert, 'root', 'ca')
|
|
sign_crl(connection, cfg, name)
|
|
publish(connection)
|
|
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)
|
|
with directory() as connection:
|
|
root = authority(connection, 'root')
|
|
if root is None or root['cert'] is None:
|
|
raise Failure('there is no root CA', 3)
|
|
if crl.issuer != root['cert'].subject or not crl.is_signature_valid(root['cert'].public_key()):
|
|
raise Failure('the CRL is not signed by the root CA', 2)
|
|
if root['crl'] is not None and utc(crl, 'last_update') < utc(root['crl'], 'last_update'):
|
|
raise Failure('the CRL is older than the current one', 2)
|
|
update(connection, 'root', {'certificateRevocationList;binary': crl.public_bytes(serialization.Encoding.DER),
|
|
'vpnRouterCRLSource': b'uploaded'})
|
|
publish(connection)
|
|
reload_crls(connection)
|
|
following = utc(crl, 'next_update')
|
|
return {'next_update': iso(following) if following else None}
|
|
|
|
|
|
def export_root_key(passphrase):
|
|
"""The root key in PEM, encrypted with the passphrase."""
|
|
if not passphrase:
|
|
raise Failure('the passphrase is required', 2)
|
|
with directory() as connection:
|
|
root = authority(connection, 'root')
|
|
if root is None or root['key'] is None:
|
|
raise Failure('the root key is not on the router', 3)
|
|
return key_pem(root['key'], passphrase.encode())
|
|
|
|
|
|
def remove_root_key():
|
|
with directory() as connection:
|
|
root = authority(connection, 'root')
|
|
if root is None or root['key'] is None:
|
|
raise Failure('the root key is not on the router', 3)
|
|
update(connection, 'root', {'vpnRouterPrivateKey': None})
|
|
|
|
|
|
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."""
|
|
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)
|
|
with directory() as connection:
|
|
root = authority(connection, 'root')
|
|
if root is None or root['cert'] is None:
|
|
raise Failure('there is no root CA', 3)
|
|
if not same_key(key.public_key(), root['cert'].public_key()):
|
|
raise Failure('the key is not the key of the root CA', 2)
|
|
update(connection, 'root', {'vpnRouterPrivateKey': key_pem(key)})
|
|
|
|
|
|
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."""
|
|
with directory() as connection:
|
|
refresh_crls(connection, cfg)
|
|
publish(connection)
|
|
reload_crls(connection)
|
|
due = [current[0][1] for current in (holders(connection, ROUTER_CERTS[name][0], router_condition(name))
|
|
for name in ROUTER_CERTS) if current]
|
|
return any(renewal_due(cert) for cert in due)
|
|
|
|
|
|
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; the root
|
|
key on the router is a warning while pki.root_key_warning is true."""
|
|
moment = now()
|
|
soon = moment + datetime.timedelta(days=WARN_DAYS)
|
|
warnings = []
|
|
|
|
def crl_of(entry):
|
|
"""The next update of the CA's CRL; a CRL the router signs warns once past half of its validity, which the timer
|
|
renews before, and one signed offline WARN_DAYS before it expires."""
|
|
crl = entry['crl']
|
|
following = utc(crl, 'next_update') if crl is not None else None
|
|
if crl is None:
|
|
warnings.append(f'{entry["name"]} has no CRL')
|
|
elif following is not None and (crl_stale(crl) if entry['key'] is not None else following <= soon):
|
|
warnings.append(f'the CRL of {entry["name"]} {"expired" if following <= moment else "expires"} on {iso(following)}')
|
|
return iso(following) if following else None
|
|
|
|
with directory() as connection:
|
|
root_entry = authority(connection, 'root')
|
|
entries = {name: authority(connection, name) for name in vpnrouter.issuing_cas(cfg)}
|
|
retired_found = connection.search_s(vpnrouter_ldap.PKI, ldap.SCOPE_ONELEVEL,
|
|
'(&(objectClass=vpnRouterAuthority)(vpnRouterRetired=*))',
|
|
['cn', 'vpnRouterRetired', 'cACertificate;binary'])
|
|
router_certs = [current[0][1] for current in (holders(connection, ROUTER_CERTS[name][0], router_condition(name))
|
|
for name in ROUTER_CERTS) if current]
|
|
|
|
root = None
|
|
if root_entry is not None and root_entry['cert'] is not None:
|
|
cert, key = root_entry['cert'], root_entry['key'] is not None
|
|
root = {'subject': common_name(cert), 'expires': iso(utc(cert, 'not_valid_after')), 'key': key,
|
|
'crl_next_update': crl_of(root_entry) if key or root_entry['crl'] is not None else None,
|
|
'crl_source': root_entry['crl_source']}
|
|
if key and cfg['pki']['root_key_warning']:
|
|
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 key and root_entry['crl'] is None:
|
|
warnings.append('the root has no CRL; sign one offline and upload it with vpn-router-ca root import-crl')
|
|
if utc(cert, 'not_valid_after') <= soon:
|
|
warnings.append(f'the root certificate expires on {root["expires"]}')
|
|
|
|
authorities = []
|
|
for name, settings in vpnrouter.issuing_cas(cfg).items():
|
|
entry = entries[name]
|
|
cert = entry['cert'] if entry else None
|
|
kind = 'active' if cert is not None else 'pending' if entry and entry['csr'] is not None else 'missing'
|
|
authorities.append({
|
|
'name': name, 'common_name': settings['name'], 'state': kind, 'builtin': settings['builtin'],
|
|
'templates': settings['templates'], 'acme': settings['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': bool(entry and entry['key'] is not None), 'crl_next_update': crl_of(entry) if cert is not None else None,
|
|
'acme_url': acme_url(cfg, name) if settings['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 {authorities[-1]["expires"]}')
|
|
|
|
retired = []
|
|
for _, attributes in retired_found:
|
|
der = first(attributes, 'cACertificate;binary')
|
|
serial = f'{x509.load_der_x509_certificate(der).serial_number:X}' if der else None
|
|
name = RETIRED_SUFFIX.sub('', attributes['cn'][0].decode())
|
|
retired.append({'name': name, 'retired': from_generalized(first(attributes, 'vpnRouterRetired')), 'serial': serial})
|
|
root_crl = root_entry['crl'] if root_entry else None
|
|
if serial and root_crl is not None and root_crl.get_revoked_certificate_by_serial_number(int(serial, 16)) is None:
|
|
warnings.append(f'the root CRL does not revoke retired issuing CA {name}')
|
|
for cert in router_certs:
|
|
if utc(cert, 'not_valid_after') <= soon:
|
|
warnings.append(f'the router certificate {common_name(cert)} expires on {iso(utc(cert, "not_valid_after"))}')
|
|
return {'root': root, 'authorities': authorities, 'retired': retired, 'warnings': warnings}
|