Skip to content

C
caucase
Commit 3aefb18a authored by Vincent Pelletier's avatar Vincent Pelletier
1 Browse files
Options

caucase: Fix CRL support. 

Emit Certificate Revocation Lists signed by all valid CAs.
Apparently openssl (or at least how it is used in stunnel4) fails to
validate a certificate when CRL validation is enabled and the key which
signed the CRL differs from the key which signed the certificate.
Also, add Authority Key Identifier CRL extension, required to be standard-
compliant.
Also, fix revocation entry expiration: the RFC requires them to be kept
at least one renewal cycle after the certificate's expiration.
As a consequence of this whole change:
- the protocol for retrieving the curren CRL changes to return the
  concatenated list of CRLs, which breaks the CRL distribution (...but
  the distributed CRLs were invalid anyway)
- stop storing the CRL PEM in caucased's database so that it gets
  re-generated with fresh code. As caucased is not expected to be
  restarted very often, the extra CRL generation on every start should
  not make a difference.
parent 58c51150
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 1120 additions and 245 deletions
CHANGES.txt
View file @ 3aefb18a
......@@ -3,6 +3,7 @@
* Add AuthorityKeyIdentifier extension in CRLs.
* Accept user certificates signed by non-current CA.
* Name CA certificates after their AuthorityKeyIdentifier keyid extension instead of their serial.
* Produce one CRL per CA certificate, as some ssl-using services fail when there is no CRL signed by the same CA as the certificate being validated.
0.9.8 (2020-06-29)
==================
......
README.rst
View file @ 3aefb18a
......@@ -117,6 +117,44 @@ caucase, the CRL is re-generated whenever it is requested and:
- previous CRL expired
- any revocation happened since previous CRL was created
Here is an illustration of the certificate and CA certificate renewal process::
Time from first caucased start:
+--------+--------+--------+--------+--------+--------+--------+-->
Certificate 1 validity: | | |
|[cert 1v1] [cert 1v3] [cert 1v5] [cert 1v7] [cert 1v9] [ce...
| [cert 1v2] [cert 1v4] [cert 1v6] [cert 1v8] [cert 1vA]
Certificate 2 validity: | | |
| [cert 2v1] [cert 2v3]| [cert 2v5] [cert 2v7] [cert 2v9]|
| [cert 2v2] [cert 2v4] [cert 2v6]| [cert 2v8] [cert...
CA certificates validity: | | |
[ca v1 | ] | |
| [ca v2 | | ] |
| | [ca v3 | |...
| | | [ca v4 |...
CRL validity for CA1: | | |
[ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ] | |
CRL validity for CA2: | | |
| [ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ] |
CRL validity for CA3: | | |
| | [ ][ ][ ][ ][ ][ ][ ][ ][ ][...
CA renewal phase: | | |
|none |passive |active |passive |active |passive |...
Active CA: | | |
[ca v1 ][ca v2 ][ca v3 |...
Legend::
+--------+ : One certificate validity period (default: 93 days)
Points of interest:
- this illustration assumes no revocation happen
- there usually are 2 simultaneously-valid CA certificates
- there usually are 2 simultaneously-valid CRLs overall, one per CA certificate
- the first ``cert 1`` signed by CA v2 is ``cert 1v6``
- the first ``cert 2`` signed by CA v2 is ``cert 1v5``
Commands
========
......
caucase/ca.py
View file @ 3aefb18a
......@@ -151,6 +151,7 @@ class CertificateAuthority(object):
When given with a true value, auto_sign_csr_amount is stored and the
value given on later instanciation will be ignored.
"""
self._current_crl_dict = {}
self._storage = storage
self._ca_renewal_lock = threading.Lock()
if lock_auto_sign_csr_amount:
......@@ -209,10 +210,12 @@ class CertificateAuthority(object):
pem_key_pair['key_pem'],
)
crt_pem = pem_key_pair['crt_pem']
crt = utils.load_ca_certificate(pem_key_pair['crt_pem'])
key = utils.load_privatekey(pem_key_pair['key_pem'])
ca_key_pair_list.append({
'crt': utils.load_ca_certificate(pem_key_pair['crt_pem']),
'crt': crt,
'key': key,
'authority_key_identifier': utils.getAuthorityKeyIdentifier(crt),
})
if previous_key is not None:
ca_certificate_chain.append(utils.wrap(
......@@ -225,6 +228,7 @@ class CertificateAuthority(object):
))
previous_crt_pem = crt_pem
previous_key = key
self._current_crl_dict.clear()
self._ca_key_pairs_list = ca_key_pair_list
self._ca_certificate_chain = tuple(
ca_certificate_chain
......@@ -341,11 +345,9 @@ class CertificateAuthority(object):
critical=False, # "MUST mark this extension as non-critical"
),
Extension(
x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(
ca_crt.extensions.get_extension_for_class(
x509.SubjectKeyIdentifier,
).value,
),
ca_crt.extensions.get_extension_for_class(
x509.AuthorityKeyIdentifier,
).value,
critical=False, # "MUST mark this extension as non-critical"
),
],
......@@ -355,7 +357,13 @@ class CertificateAuthority(object):
x509.CRLDistributionPoints([
x509.DistributionPoint(
full_name=[
x509.UniformResourceIdentifier(self._crl_base_url),
x509.UniformResourceIdentifier(
self._crl_base_url + (
'/%i' % (
utils.getAuthorityKeyIdentifier(ca_crt),
)
),
),
],
relative_name=None,
crl_issuer=None,
......@@ -662,15 +670,22 @@ class CertificateAuthority(object):
crt = utils.load_certificate(
crt_pem,
self.getCACertificateList(),
x509.load_pem_x509_crl(
self.getCertificateRevocationList(),
_cryptography_backend,
),
crl_list=[
x509.load_pem_x509_crl(x, _cryptography_backend)
for x in self.getCertificateRevocationListDict().itervalues()
],
)
self._storage.revoke(
serial=crt.serial_number,
expiration_date=utils.datetime2timestamp(crt.not_valid_after),
expiration_date=utils.datetime2timestamp(
# https://tools.ietf.org/html/rfc5280#section-3.3
# An entry MUST NOT be removed
# from the CRL until it appears on one regularly scheduled CRL issued
# beyond the revoked certificate's validity period.
crt.not_valid_after + self._crl_life_time,
),
)
self._current_crl_dict.clear()
def revokeSerial(self, serial):
"""
......@@ -687,10 +702,13 @@ class CertificateAuthority(object):
"""
self._storage.revoke(
serial=serial,
expiration_date=utils.datetime2timestamp(max(
x.not_valid_after for x in self.getCACertificateList()
)),
expiration_date=utils.datetime2timestamp(
max(
x.not_valid_after for x in self.getCACertificateList()
) + self._crl_life_time,
),
)
self._current_crl_dict.clear()
def renew(self, crt_pem, csr_pem):
"""
......@@ -704,10 +722,10 @@ class CertificateAuthority(object):
crt = utils.load_certificate(
crt_pem,
self.getCACertificateList(),
x509.load_pem_x509_crl(
self.getCertificateRevocationList(),
_cryptography_backend,
),
crl_list=[
x509.load_pem_x509_crl(x, _cryptography_backend)
for x in self.getCertificateRevocationListDict().itervalues()
],
)
return self._createCertificate(
csr_id=self.appendCertificateSigningRequest(
......@@ -728,53 +746,82 @@ class CertificateAuthority(object):
),
)
def getCertificateRevocationList(self):
"""
Return PEM-encoded certificate revocation list.
"""
crl_pem = self._storage.getCertificateRevocationList()
if crl_pem is None:
ca_key_pair = self._getCurrentCAKeypair()
ca_crt = ca_key_pair['crt']
now = datetime.datetime.utcnow()
crl = x509.CertificateRevocationListBuilder(
issuer_name=ca_crt.issuer,
last_update=now,
next_update=now + self._crl_life_time,
extensions=[
Extension(
x509.CRLNumber(
self._storage.getNextCertificateRevocationListNumber(),
def getCertificateRevocationListDict(self):
"""
Return PEM-encoded certificate revocation lists for all CAs.
"""
now = datetime.datetime.utcnow()
result = {}
crl_pem_dict = self._current_crl_dict
self._renewCAIfNeeded()
storage = self._storage
crl_number, last_update = storage.getCurrentCRLNumberAndLastUpdate()
has_renewed = last_update is None
last_update = (
None
if last_update is None else
utils.timestamp2datetime(last_update)
)
revoked_certificate_list = None
for ca_key_pair in self._ca_key_pairs_list:
authority_key_identifier = ca_key_pair['authority_key_identifier']
try:
crl_pem, crl_expiration_date = crl_pem_dict[authority_key_identifier]
except KeyError:
crl_pem = None
if crl_pem is None or crl_expiration_date < now:
if not has_renewed and crl_pem is not None:
# CRL expired, generate a new serial
last_update = None
has_renewed = True
if (
last_update is None or
last_update + self._crl_renew_time < now
):
# We cannot use the existing CRL (or maybe none exist), generate a
# new one.
last_update = now
crl_number = storage.getNextCertificateRevocationListNumber()
storage.storeCRLLastUpdate(
last_update=utils.datetime2timestamp(last_update),
)
if revoked_certificate_list is None:
revoked_certificate_list = [
x509.RevokedCertificateBuilder(
serial_number=x['serial'],
revocation_date=utils.timestamp2datetime(x['revocation_date']),
).build(_cryptography_backend)
for x in storage.getRevocationList()
]
ca_crt = ca_key_pair['crt']
crl_pem = x509.CertificateRevocationListBuilder(
issuer_name=ca_crt.issuer,
last_update=last_update,
next_update=last_update + self._crl_life_time,
extensions=[
Extension(
x509.CRLNumber(crl_number),
critical=False, # "MUST mark this extension as non-critical"
),
critical=False, # "MUST mark this extension as non-critical"
),
Extension(
x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(
Extension(
ca_crt.extensions.get_extension_for_class(
x509.SubjectKeyIdentifier,
x509.AuthorityKeyIdentifier,
).value,
critical=False, # No mention in RFC5280 5.2.1
),
critical=False, # No mention in RFC5280 5.2.1
),
],
revoked_certificates=[
x509.RevokedCertificateBuilder(
serial_number=x['serial'],
revocation_date=utils.timestamp2datetime(x['revocation_date']),
).build(_cryptography_backend)
for x in self._storage.getRevocationList()
],
).sign(
private_key=ca_key_pair['key'],
algorithm=self._default_digest_class(),
backend=_cryptography_backend,
)
crl_pem = crl.public_bytes(serialization.Encoding.PEM)
self._storage.storeCertificateRevocationList(
crl_pem,
expiration_date=utils.datetime2timestamp(now + self._crl_renew_time),
)
return crl_pem
],
revoked_certificates=revoked_certificate_list,
).sign(
private_key=ca_key_pair['key'],
algorithm=self._default_digest_class(),
backend=_cryptography_backend,
).public_bytes(serialization.Encoding.PEM)
crl_pem_dict[authority_key_identifier] = (
crl_pem,
last_update + self._crl_renew_time,
)
result[authority_key_identifier] = crl_pem
return result
class UserCertificateAuthority(CertificateAuthority):
"""
......@@ -795,10 +842,10 @@ class UserCertificateAuthority(CertificateAuthority):
certificates.
"""
ca_cert_list = self.getCACertificateList()
crl = x509.load_pem_x509_crl(
self.getCertificateRevocationList(),
_cryptography_backend,
)
crl_list = [
x509.load_pem_x509_crl(x, _cryptography_backend)
for x in self.getCertificateRevocationListDict().itervalues()
]
signing_key = os.urandom(32)
symetric_key = os.urandom(32)
iv = os.urandom(16)
......@@ -817,7 +864,7 @@ class UserCertificateAuthority(CertificateAuthority):
key_list = []
for crt_pem in self._storage.iterCertificates():
try:
crt = utils.load_certificate(crt_pem, ca_cert_list, crl)
crt = utils.load_certificate(crt_pem, ca_cert_list, crl_list)
except CertificateVerificationError:
continue
public_key = crt.public_key()
......
caucase/cli.py
View file @ 3aefb18a
......@@ -403,13 +403,27 @@ def main(argv=None, stdout=sys.stdout, stderr=sys.stderr):
'--crl',
default='cas.crl.pem',
metavar='CRL_PATH',
help='Services certificate revocation list location. default: %(default)s',
help='Services certificate revocation list location. '
'May be an existing directory or file, or non-existing. '
'If non-existing and given path has an extension, a file will be created, '
'otherwise a directory will be. '
'When it is a file, it may contain multiple PEM-encoded concatenated '
'CRLs. When it is a directory, it may contain multiple files, each '
'containing a single PEM-encoded CRL. '
'default: %(default)s',
)
parser.add_argument(
'--user-crl',
default='cau.crl.pem',
metavar='CRL_PATH',
help='Users certificate revocation list location. default: %(default)s',
help='Users certificate revocation list location. '
'May be an existing directory or file, or non-existing. '
'If non-existing and given path has an extension, a file will be created, '
'otherwise a directory will be. '
'When it is a file, it may contain multiple PEM-encoded concatenated '
'CRLs. When it is a directory, it may contain multiple files, each '
'containing a single PEM-encoded CRL. '
'default: %(default)s',
)
parser.add_argument(
'--threshold',
......@@ -814,6 +828,12 @@ def updater(argv=None, until=utils.until):
required=True,
metavar='CRT_PATH',
help='Path of your certificate revocation list for MODE. '
'May be an existing directory or file, or non-existing. '
'If non-existing and given path has an extension, a file will be created, '
'otherwise a directory will be. '
'When it is a file, it may contain multiple PEM-encoded concatenated '
'CRLs. When it is a directory, it may contain multiple files, each '
'containing a single PEM-encoded CRL. '
'Will be maintained up-to-date.'
)
args = parser.parse_args(argv)
......@@ -889,11 +909,11 @@ def updater(argv=None, until=utils.until):
if RetryingCaucaseClient.updateCRLFile(ca_url, args.crl, ca_crt_list):
print('Got new CRL')
updated = True
with open(args.crl, 'rb') as crl_file:
for crl_pem in utils.getCRLList(args.crl):
next_deadline = min(
next_deadline,
utils.load_crl(
crl_file.read(),
crl_pem,
ca_crt_list,
).next_update - crl_threshold,
)
......
caucase/client.py
View file @ 3aefb18a
......@@ -25,12 +25,12 @@ from __future__ import absolute_import
import datetime
import httplib
import json
import os
import ssl
from urlparse import urlparse
from cryptography import x509
from cryptography.hazmat.backends import default_backend
import cryptography.exceptions
import pem
from . import utils
from . import version
......@@ -118,25 +118,38 @@ class CaucaseClient(object):
url (str)
URL to caucase, ending in eithr /cas or /cau.
crl_path (str)
Path to the CRL file, which may not exist.
Path to the CRL file or directory, which may not exist.
If it does not exist, it is created. If there is an extension, a file is
created, otherwise a directory is.
ca_list (list of cryptography.x509.Certificate instances)
One of these CA certificates must have signed the CRL for it to be
accepted.
Return whether an update happened.
"""
if os.path.exists(crl_path):
with open(crl_path, 'rb') as crl_file:
my_crl = utils.load_crl(crl_file.read(), ca_list)
def _asCRLDict(crl_list):
return {
utils.getAuthorityKeyIdentifier(utils.load_crl(x, ca_list)): x
for x in crl_list
}
local_crl_list = utils.getCRLList(crl_path)
try:
local_crl_dict = _asCRLDict(crl_list=local_crl_list)
except x509.extensions.ExtensionNotFound:
# BBB: caucased used to issue CRLs without the AuthorityKeyIdentifier
# extension. In such case, local CRLs need to be replaced.
local_crl_list = []
local_crl_dict = {}
updated = True
else:
my_crl = None
latest_crl_pem = cls(ca_url=url).getCertificateRevocationList()
latest_crl = utils.load_crl(latest_crl_pem, ca_list)
if my_crl is None or latest_crl.signature != my_crl.signature:
with open(crl_path, 'wb') as crl_file:
crl_file.write(latest_crl_pem)
return True
return False
updated = len(local_crl_list) != len(local_crl_dict)
server_crl_list = cls(ca_url=url).getCertificateRevocationListList()
for ca_key_id, crl_pem in _asCRLDict(crl_list=server_crl_list).iteritems():
updated |= local_crl_dict.pop(ca_key_id, None) != crl_pem
updated |= bool(local_crl_dict)
if updated:
utils.saveCRLList(crl_path, server_crl_list)
return updated
def __init__(
self,
......@@ -208,11 +221,25 @@ class CaucaseClient(object):
def _https(self, method, url, body=None, headers=None):
return self._request(self._https_connection, method, url, body, headers)
def getCertificateRevocationList(self):
def getCertificateRevocationList(self, authority_key_identifier):
"""
[ANONYMOUS] Retrieve latest CRL.
[ANONYMOUS] Retrieve latest CRL for given integer authority key
identifier.
"""
return self._http('GET', '/crl')
return self._http(
'GET',
'/crl/%i' % (authority_key_identifier, ),
)
def getCertificateRevocationListList(self):
"""
[ANONYMOUS] Retrieve the latest CRLs for each CA certificate.
"""
return [
x.as_bytes()
for x in pem.parse(self._http('GET', '/crl'))
if isinstance(x, pem.CertificateRevocationList)
]
def getCertificateSigningRequest(self, csr_id):
"""
......
caucase/http.py
View file @ 3aefb18a
......@@ -1121,10 +1121,20 @@ def manage(argv=None, stdout=sys.stdout):
for x in trusted_ca_crt_set
)
already_revoked_count = revoked_count = 0
crl_number = crl_last_update = None
for import_crl in args.import_crl:
with open(import_crl, 'rb') as crl_file:
crl_data = crl_file.read()
for revoked in utils.load_crl(crl_data, trusted_ca_crt_set):
crl = utils.load_crl(crl_file.read(), trusted_ca_crt_set)
current_crl_number = crl.extensions.get_extension_for_class(
x509.CRLNumber,
).value.crl_number
if crl_number is None:
crl_number = current_crl_number
crl_last_update = crl.last_update
else:
crl_number = max(crl_number, current_crl_number)
crl_last_update = max(crl_last_update, crl.last_update)
for revoked in crl:
try:
db.revoke(
revoked.serial_number,
......@@ -1134,6 +1144,15 @@ def manage(argv=None, stdout=sys.stdout):
already_revoked_count += 1
else:
revoked_count += 1
db.storeCRLLastUpdate(utils.datetime2timestamp(crl_last_update))
db.storeCRLNumber(crl_number)
print(
'Set CRL number to %i and last update to %s' % (
crl_number,
crl_last_update.isoformat(' '),
),
file=stdout,
)
print(
'Revoked %i certificates (%i were already revoked)' % (
revoked_count,
......
caucase/storage.py
View file @ 3aefb18a
......@@ -27,11 +27,14 @@ import os
import sqlite3
from threading import local
from time import time
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from .exceptions import NoStorage, NotFound, Found
from .utils import toBytes, toUnicode
from .utils import toBytes, toUnicode, datetime2timestamp
__all__ = ('SQLite3Storage', )
_cryptography_backend = default_backend()
DAY_IN_SECONDS = 60 * 60 * 24
class NoReentryConnection(sqlite3.Connection):
......@@ -113,7 +116,8 @@ class SQLite3Storage(local):
# sqlite can accept as integers, so store these as text. Use a trivial
# string serialisation: not very space efficient, but this should not be
# a limiting issue for our use-cases anyway.
db.cursor().executescript('''
c = db.cursor()
c.executescript('''
CREATE TABLE IF NOT EXISTS %(prefix)sca (
expiration_date INTEGER,
key TEXT,
......@@ -131,10 +135,6 @@ class SQLite3Storage(local):
revocation_date INTEGER,
expiration_date INTEGER
);
CREATE TABLE IF NOT EXISTS %(prefix)scrl (
expiration_date INTEGER,
crl TEXT
);
CREATE TABLE IF NOT EXISTS %(prefix)scounter (
name TEXT PRIMARY KEY,
value INTEGER
......@@ -143,10 +143,33 @@ class SQLite3Storage(local):
name TEXT PRIMARY KEY,
value TEXT
);
CREATE TABLE IF NOT EXISTS %(prefix)sconfig (
name TEXT PRIMARY KEY,
value TEXT
);
''' % {
'prefix': table_prefix,
'key_id_constraint': 'UNIQUE' if enforce_unique_key_id else '',
})
try:
crl_row = self._executeSingleRow(
'SELECT crl FROM %(prefix)scrl '
'ORDER BY expiration_date DESC LIMIT 1',
)
except sqlite3.OperationalError as exc:
# XXX: no error codes in sqlite and only a generic error class ?
if not exc.args[0].startswith('no such table: '): # pragma: no cover
raise
crl_row = None
if crl_row is not None:
self._setConfig(
'crl_last_update',
str(datetime2timestamp(x509.load_pem_x509_crl(
toBytes(crl_row['crl']),
_cryptography_backend,
).last_update)),
)
self._execute(c, 'DROP TABLE IF EXISTS %(prefix)scrl')
def _execute(self, cursor, sql, parameters=()):
return cursor.execute(
......@@ -217,6 +240,30 @@ class SQLite3Storage(local):
except sqlite3.IntegrityError:
pass
def _getConfig(self, name, default):
"""
Retrieve the value of <name> from config list, or <default> if not
stored.
"""
result = self._executeSingleRow(
'SELECT value FROM %(prefix)sconfig WHERE name = ?',
(name, ),
)
if result is None:
return default
return result['value']
def _setConfig(self, name, value):
"""
Store <value> as <name> in config list, possibly overwriting an existing
entry.
"""
self._execute(
self._db.cursor(),
'INSERT OR REPLACE INTO %(prefix)sconfig (name, value) VALUES (?, ?)',
(name, value),
)
def getCAKeyPairList(self, prune=True):
"""
Return the chronologically sorted (oldest in [0], newest in [-1])
......@@ -462,7 +509,6 @@ class SQLite3Storage(local):
"""
with self._db as db:
c = db.cursor()
self._execute(c, 'DELETE FROM %(prefix)scrl')
try:
self._execute(
c,
......@@ -477,44 +523,51 @@ class SQLite3Storage(local):
)
except sqlite3.IntegrityError:
raise Found
self._incrementCounter('crl_number')
def getCertificateRevocationList(self):
def getNextCertificateRevocationListNumber(self):
"""
Get PEM-encoded current Certificate Revocation List.
Returns None if there is no CRL.
Get next CRL sequence number.
"""
with self._db:
row = self._executeSingleRow(
'SELECT crl FROM %scrl '
'WHERE expiration_date > ? ORDER BY expiration_date DESC LIMIT 1' % (
self._table_prefix,
),
(time(), )
)
if row is not None:
return toBytes(row['crl'])
return None
return self._incrementCounter('crl_number')
def getNextCertificateRevocationListNumber(self):
def storeCRLLastUpdate(self, last_update):
"""
Get next CRL sequence number.
"""
return self._incrementCounter('crl_number')
with self._db:
self._setConfig('crl_last_update', str(last_update))
def storeCertificateRevocationList(self, crl, expiration_date):
def storeCRLNumber(self, crl_number):
"""
Store Certificate Revocation List.
Set the current CRL sequence number.
Use only when importing an existing CA.
"""
with self._db as db:
c = db.cursor()
self._execute(c, 'DELETE FROM %(prefix)scrl')
self._execute(
c,
'INSERT INTO %(prefix)scrl (expiration_date, crl) VALUES (?, ?)',
db.cursor(),
'INSERT OR REPLACE INTO %(prefix)scounter (name, value) VALUES (?, ?)',
(
'crl_number',
crl_number,
),
)
def getCurrentCRLNumberAndLastUpdate(self):
"""
Get the current CRL sequence number.
"""
with self._db:
last_update = self._getConfig('crl_last_update', None)
return (
# Note: does not increment the counter, but may set it to the default
# value.
self._incrementCounter('crl_number', increment=0),
(
int(expiration_date),
crl,
last_update
if last_update is None else
int(last_update, 10)
),
)
......
caucase/test.py
View file @ 3aefb18a
This diff is collapsed.
caucase/utils.py
View file @ 3aefb18a
......@@ -32,6 +32,7 @@ import datetime
import email
import json
import os
import sys
import threading
import traceback
import time
......@@ -86,7 +87,6 @@ _CAUCASE_LEGACY_OID_AUTO_SIGNED = x509.oid.ObjectIdentifier(
CAUCASE_LEGACY_OID_AUTO_SIGNED,
)
def isCertificateAutoSigned(crt):
"""
Checks whether given certificate was automatically signed by caucase.
......@@ -127,6 +127,12 @@ def getCertList(crt_path):
"""
return _getPEMListFromPath(crt_path, pem.Certificate)
def getCRLList(crl_path):
"""
Return a list of Certificate Revocation Lists.
"""
return _getPEMListFromPath(crl_path, pem.CertificateRevocationList)
def _getPEMListFromPath(path, pem_type):
if not os.path.exists(path):
return []
......@@ -155,6 +161,26 @@ def saveCertList(crt_path, cert_pem_list):
"""
_savePEMList(crt_path, cert_pem_list, load_ca_certificate, '.ca.pem')
def saveCRLList(crl_path, crl_pem_list):
"""
Store given list of PEM-encoded Certificate Revocation Lists in given path.
crl_path (str)
May point to a directory a file, or nothing.
If it does not exist, and this value contains an extension, a file is
created, otherwise a directory is.
If it is a file, all CRLs are written in it.
If it is a folder, each CRL is stored in a separate file.
crl_pem_list (list of bytes)
"""
_savePEMList(
crl_path,
crl_pem_list,
lambda x: x509.load_pem_x509_crl(x, _cryptography_backend),
'.crl.pem',
)