91 lines
5.6 KiB
Python
91 lines
5.6 KiB
Python
"""The PKI: /ca, and the certificates and CRLs /pki/ serves without authentication."""
|
|
|
|
import ldap
|
|
|
|
import vpnrouter
|
|
import vpnrouter_pki
|
|
from vpnrouter_status import run
|
|
|
|
from .common import PKI_STATUS
|
|
|
|
#: A file /pki/ serves: the certificate or the CRL of a CA, and its content type.
|
|
PUBLISHED = {'.crt': 'application/pkix-cert', '.crl': 'application/pkix-crl'}
|
|
|
|
|
|
def ca(method, path, body):
|
|
"""Answer a /ca request with vpnrouter_pki: the HTTP status and the JSON document.
|
|
|
|
GET /ca {"settings", "templates", "extended_key_usages", "status",
|
|
"certificates"}: the pki section, the templates with the built-in
|
|
ones, the extended key usages, each {oid, builtin}, the root, the
|
|
issuing CAs, the retired CAs and the warnings (vpnrouter_pki.status),
|
|
and the certificates issued, newest first
|
|
PUT /ca change any of the settings of the pki section
|
|
POST /ca/certificates issue a certificate: {"ca", "template", "csr"} answers {"serial",
|
|
"certificate", "chain"}; {"ca", "template", "subject", "sans",
|
|
"password"} answers {"serial", "pfx"}, a PKCS#12 file in base64
|
|
POST /ca/certificates/<serial>/revoke revoke a certificate: {"reason"?}
|
|
GET /ca/bundle/<name> the PEM certificates a client trusts the CA with: {"name", "pem"}; for
|
|
root the root alone, for an issuing CA the issuing CA and the root
|
|
GET /ca/issuing/<name>/csr the CSR of an issuing CA that waits for the offline root, with the
|
|
openssl extensions to sign it with: {"name", "csr", "extensions"}
|
|
PUT /ca/issuing/<name>/certificate take its certificate signed by the root: {"certificate"}
|
|
PUT /ca/root/crl take a root CRL signed offline: {"crl"}
|
|
|
|
A change of the settings is made as /site-to-site makes one; it, a revoked router certificate and an imported
|
|
issuing CA certificate are applied by restarting vpn-router-setup without waiting for it.
|
|
"""
|
|
if body == 'invalid' or (body is not None and not isinstance(body, dict)):
|
|
return 400, {'error': 'the body must be a JSON object'}
|
|
body = body or {}
|
|
parts = path.strip('/').split('/')
|
|
route = (method, *parts[1:])
|
|
try:
|
|
cfg = vpnrouter.load_config()
|
|
if len(route) == 3 and route[:2] == ('GET', 'bundle'):
|
|
return 200, vpnrouter_pki.bundle(route[2])
|
|
if route == ('GET',):
|
|
return 200, {'settings': cfg['pki'], 'templates': vpnrouter.pki_templates(cfg),
|
|
'extended_key_usages': {name: {'oid': oid, 'builtin': name in vpnrouter.EXTENDED_KEY_USAGES}
|
|
for name, oid in vpnrouter.extended_key_usages(cfg).items()},
|
|
'status': vpnrouter_pki.status(cfg), 'certificates': vpnrouter_pki.certificates()}
|
|
if route == ('PUT',):
|
|
settings = {option: body[option] for option in vpnrouter.SCHEMA['pki'] if option in body}
|
|
with vpnrouter.config_lock():
|
|
data = vpnrouter.read_settings()
|
|
current = data.get('pki')
|
|
data['pki'] = {**(current if isinstance(current, dict) else {}), **settings}
|
|
cfg = vpnrouter.with_defaults(data)
|
|
vpnrouter.validate(cfg)
|
|
vpnrouter.save_config(data)
|
|
run('systemctl', 'restart', '--no-block', 'vpn-router-setup.service')
|
|
return 200, {'settings': cfg['pki'], 'templates': vpnrouter.pki_templates(cfg)}
|
|
with vpnrouter_pki.locked():
|
|
if route == ('POST', 'certificates'):
|
|
return 201, vpnrouter_pki.issue(cfg, body.get('ca'), body.get('template'), csr=body.get('csr'),
|
|
subject=body.get('subject'), sans=body.get('sans'),
|
|
password=body.get('password'))
|
|
if len(route) == 4 and route[:2] == ('POST', 'certificates') and route[3] == 'revoke':
|
|
document = vpnrouter_pki.revoke(cfg, route[2], body.get('reason'))
|
|
elif len(route) == 4 and route[:2] == ('GET', 'issuing') and route[3] == 'csr':
|
|
return 200, vpnrouter_pki.csr(cfg, route[2])
|
|
elif len(route) == 4 and route[:2] == ('PUT', 'issuing') and route[3] == 'certificate':
|
|
document = vpnrouter_pki.import_ca(cfg, route[2], body.get('certificate'))
|
|
elif route == ('PUT', 'root', 'crl'):
|
|
return 200, vpnrouter_pki.import_root_crl(body.get('crl'))
|
|
else:
|
|
return 404, {'error': f'no route {method} {path}'}
|
|
except vpnrouter.ConfigError as exc:
|
|
return 400, {'error': str(exc)}
|
|
except vpnrouter_pki.Failure as failure:
|
|
return PKI_STATUS.get(failure.status, 500), {'error': str(failure)}
|
|
except ldap.LDAPError as exc:
|
|
return 500, {'error': f'the directory, which holds the PKI, failed: {exc}'}
|
|
except OSError as exc:
|
|
return 500, {'error': f'the PKI cannot be changed: {exc}'}
|
|
except SystemExit as exc:
|
|
return 500, {'error': str(exc.code)}
|
|
if route[1] == 'issuing' or document.get('reissue'):
|
|
run('systemctl', 'restart', '--no-block', 'vpn-router-setup.service')
|
|
return 200, document
|