Skip to content

C
caucase
Commit 5fe1e86b authored by Vincent Pelletier's avatar Vincent Pelletier
Browse files
Options

all: Keep track of certificate issuances. 

And use this tracking to to warn about surviving certificates which are
related to the one just revoked - they may need some attention too.

NOTE: While this should be correctly implemented, I think this is not
usable, and hence probably not worth the extra complexity: what can one
do when given a list of serials ? This version discards old tracking
entries, but even if it did not how is one supposed to browse these ?
parent 228e01d7
Hide whitespace changes
Inline Side-by-side
Showing with 641 additions and 96 deletions
caucase/ca.py
View file @ 5fe1e86b
......@@ -38,7 +38,9 @@ from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from . import utils
from .exceptions import (
CertificateVerificationError,
CertificateRevokedError,
NotACertificateSigningRequest,
Found,
)
__all__ = ('CertificateAuthority', 'UserCertificateAuthority', 'Extension')
......@@ -245,7 +247,12 @@ class CertificateAuthority(object):
if requested_amount is not None and \
requested_amount <= self._auto_sign_csr_amount:
# if allowed to sign this certificate automaticaly
self._createCertificate(csr_id, auto_signed=_AUTO_SIGNED_YES)
self._createCertificate(
csr_id,
auto_signed=_AUTO_SIGNED_YES,
is_renewal=False,
authorisation_serial=None,
)
return csr_id
def deletePendingCertificateSigningRequest(self, csr_id):
......@@ -264,7 +271,7 @@ class CertificateAuthority(object):
"""
return self._storage.getCertificateSigningRequestList()
def createCertificate(self, csr_id, template_csr=None):
def createCertificate(self, csr_id, template_csr=None, authorisation_serial=None):
"""
Sign a pending certificate signing request, storing produced certificate.
......@@ -274,39 +281,97 @@ class CertificateAuthority(object):
Copy extensions and subject from this CSR instead of stored one.
Useful to renew a certificate.
Public key is always copied from stored CSR.
authorisation_serial (int, None)
Serial of the certificate which authorised certificate issuance.
"""
self._createCertificate(
csr_id=csr_id,
auto_signed=_AUTO_SIGNED_NO,
template_csr=template_csr,
is_renewal=False,
authorisation_serial=authorisation_serial,
)
def _createCertificate(self, csr_id, auto_signed, template_csr=None):
def _createCertificate(
self,
csr_id,
auto_signed,
is_renewal,
authorisation_serial,
template_csr=None,
):
"""
auto_signed (bool)
When True, mark certificate as having been auto-signed.
When False, prevent such mark from being set.
When None, do not filter (useful when renewing).
is_renewal (bool)
True to signal a renewal, False otherwise.
authorisation_serial (int, None)
If non-None, the serial of the certificate which authorised certificate
issuance:
- If is_renewal is True, this is the serial of the certificate being
renewed.
- Otherwise, this is the serial of user certificate who triggered the
renewal.
If None, this is an auto-issued certificate.
tempate_csr (None or X509Req)
Copy extensions and subject from this CSR instead of stored one.
Useful to renew a certificate.
Public key is always copied from stored CSR.
"""
csr_pem = self._storage.getCertificateSigningRequest(csr_id)
csr = utils.load_certificate_request(csr_pem)
not_valid_before = datetime.datetime.utcnow()
not_valid_after = not_valid_before + self._crt_life_time
csr = utils.load_certificate_request(
self._storage.getCertificateSigningRequest(csr_id),
)
# Note: this is quite unlikely to loop even once, as
# x509.random_serial_number produces a 160-bits random number.
while True:
serial_number = x509.random_serial_number()
try:
with self._storage.trackIssuance(
is_renewal=is_renewal,
authorisation_serial=authorisation_serial,
serial=serial_number,
not_valid_after=utils.datetime2timestamp(not_valid_after),
) as storeCertificate:
return self.__createCertificate(
csr_id=csr_id,
csr=csr,
auto_signed=auto_signed,
template_csr=template_csr,
serial_number=serial_number,
not_valid_before=not_valid_before,
not_valid_after=not_valid_after,
storeCertificate=storeCertificate,
)
except Found: # pragma: no cover
pass
def __createCertificate(
self,
csr_id,
csr,
auto_signed,
template_csr,
serial_number,
not_valid_before,
not_valid_after,
storeCertificate,
):
if template_csr is None:
template_csr = csr
ca_key_pair = self._getCurrentCAKeypair()
ca_crt = ca_key_pair['crt']
public_key = csr.public_key()
now = datetime.datetime.utcnow()
builder = x509.CertificateBuilder(
subject_name=template_csr.subject,
issuer_name=ca_crt.subject,
not_valid_before=now,
not_valid_after=now + self._crt_life_time,
serial_number=x509.random_serial_number(),
not_valid_before=not_valid_before,
not_valid_after=not_valid_after,
serial_number=serial_number,
public_key=public_key,
extensions=[
Extension(
......@@ -459,7 +524,7 @@ class CertificateAuthority(object):
algorithm=self._default_digest_class(),
backend=_cryptography_backend,
))
self._storage.storeCertificate(csr_id, cert_pem)
storeCertificate(csr_id, cert_pem)
return cert_pem
def getCertificate(self, csr_id):
......@@ -662,14 +727,17 @@ class CertificateAuthority(object):
crt_pem (str)
PEM-encoded certificat to revoke.
"""
crt = utils.load_certificate(
crt_pem,
self.getCACertificateList(),
x509.load_pem_x509_crl(
self.getCertificateRevocationList(),
_cryptography_backend,
),
)
try:
crt = utils.load_certificate(
crt_pem,
self.getCACertificateList(),
x509.load_pem_x509_crl(
self.getCertificateRevocationList(),
_cryptography_backend,
),
)
except CertificateRevokedError:
raise Found
self._storage.revoke(
serial=crt.serial_number,
expiration_date=utils.datetime2timestamp(crt.not_valid_after),
......@@ -695,6 +763,18 @@ class CertificateAuthority(object):
)),
)
def getIssuedBy(self, serial_list, renewal):
"""
(see .storage.getIssuedBy)
"""
return self._storage.getIssuedBy(serial_list, renewal)
def getNonRevokedCertificateSerialList(self, serial_list):
"""
(see .storage.getNonRevokedCertificateSerialList)
"""
return self._storage.getNonRevokedCertificateSerialList(serial_list)
def renew(self, crt_pem, csr_pem):
"""
Renew certificate.
......@@ -718,6 +798,8 @@ class CertificateAuthority(object):
override_limits=True,
),
auto_signed=_AUTO_SIGNED_PASSTHROUGH,
is_renewal=True,
authorisation_serial=crt.serial_number,
# Do a dummy signature, just so we get a usable
# x509.CertificateSigningRequest instance. Use latest CA private key just
# because it is available for free (unlike generating a new one).
......
caucase/cli.py
View file @ 5fe1e86b
......@@ -193,6 +193,18 @@ class CLICaucaseClient(object):
crt_file.write(crt_pem)
return warning, error
def _printRevocationResult(self, result):
if result:
self._print(
'WARNING: The certificate just revoked has been used to issue '
'certificates with the following serials:',
)
for mode, serial_list in result.iteritems():
self._print('mode:', mode)
for serial in serial_list:
self._print(' ', serial)
self._print('You may want to check these.')
def revokeCRT(self, error, crt_key_list):
"""
--revoke-crt
......@@ -209,7 +221,14 @@ class CLICaucaseClient(object):
)
error = True
continue
self._client.revokeCertificate(crt, key)
try:
result = self._client.revokeCertificate(crt, key)
except CaucaseError as e:
if e.args[0] != httplib.CONFLICT:
raise
self._print('Certificate', crt_path, 'was already revoked')
result = e.args[2]
self._printRevocationResult(result)
return error
def renewCRT(
......@@ -343,14 +362,29 @@ class CLICaucaseClient(object):
),
file=self._stderr,
)
self._client.revokeCertificate(crt_pem)
try:
result = self._client.revokeCertificate(crt_pem)
except CaucaseError as e:
if e.args[0] != httplib.CONFLICT:
raise
self._print('Certificate', crt_path, 'was already revoked')
result = e.args[2]
self._printRevocationResult(result)
return error
def revokeSerial(self, serial_list):
"""
--revoke-serial
"""
for serial in serial_list:
self._client.revokeSerial(serial)
try:
result = self._client.revokeSerial(serial)
except CaucaseError as e:
if e.args[0] != httplib.CONFLICT:
raise
self._print('Certificate', serial, 'was already revoked')
result = e.args[2]
self._printRevocationResult(result)
def main(argv=None, stdout=sys.stdout, stderr=sys.stderr):
"""
......
caucase/client.py
View file @ 5fe1e86b
......@@ -330,12 +330,19 @@ class CaucaseClient(object):
data = utils.nullWrap({
'revoke_crt_pem': crt,
})
method(
'PUT',
'/crt/revoke',
json.dumps(data).encode('utf-8'),
{'Content-Type': 'application/json'},
)
try:
return json.loads(method(
'PUT',
'/crt/revoke',
json.dumps(data).encode('utf-8'),
{'Content-Type': 'application/json'},
).decode('utf-8'))
except CaucaseError as e:
if e.args[0] != httplib.CONFLICT: # pragma: no cover
raise
args = list(e.args)
args[2] = json.loads(e.args[2].decode('utf-8'))
raise CaucaseError(*args)
def revokeSerial(self, serial):
"""
......@@ -345,12 +352,19 @@ class CaucaseClient(object):
[AUTHENTICATED]
"""
self._https(
'PUT',
'/crt/revoke',
json.dumps(utils.nullWrap({'revoke_serial': serial})).encode('utf-8'),
{'Content-Type': 'application/json'},
)
try:
return json.loads(self._https(
'PUT',
'/crt/revoke',
json.dumps(utils.nullWrap({'revoke_serial': serial})).encode('utf-8'),
{'Content-Type': 'application/json'},
).decode('utf-8'))
except CaucaseError as e:
if e.args[0] != httplib.CONFLICT: # pragma: no cover
raise
args = list(e.args)
args[2] = json.loads(e.args[2].decode('utf-8'))
raise CaucaseError(*args)
def createCertificate(self, csr_id, template_csr=''):
"""
......
caucase/exceptions.py
View file @ 5fe1e86b
......@@ -42,6 +42,10 @@ class CertificateVerificationError(CertificateAuthorityException):
"""Certificate is not valid, it was not signed by CA"""
pass
class CertificateRevokedError(CertificateVerificationError):
"""Certificate is revoked"""
pass
class NotACertificateSigningRequest(CertificateAuthorityException):
"""Provided value is not a certificate signing request"""
pass
......
caucase/storage.py
View file @ 5fe1e86b
......@@ -22,6 +22,7 @@
Caucase - Certificate Authority for Users, Certificate Authority for SErvices
"""
from __future__ import absolute_import
import contextlib
from random import getrandbits
import os
import sqlite3
......@@ -53,7 +54,7 @@ class NoReentryConnection(sqlite3.Connection):
self.__entered = False
return super(NoReentryConnection, self).__exit__(exc_type, exc_value, traceback)
class SQLite3Storage(local):
class SQLite3Storage(local): # pylint: disable=too-many-public-methods
"""
CA data storage.
......@@ -106,10 +107,10 @@ class SQLite3Storage(local):
self._table_prefix = table_prefix
db.row_factory = sqlite3.Row
self._max_csr_amount = max_csr_amount
self._crt_keep_time = crt_keep_time * DAY_IN_SECONDS
self._crt_keep_time = int(crt_keep_time * DAY_IN_SECONDS)
self._crt_read_keep_time = crt_read_keep_time * DAY_IN_SECONDS
with db:
# Note about revoked.serial: certificate serials exceed the 63 bits
# Note about serials: certificate serials exceed the 63 bits
# 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.
......@@ -142,7 +143,13 @@ class SQLite3Storage(local):
CREATE TABLE IF NOT EXISTS %(prefix)sconfig_once (
name TEXT PRIMARY KEY,
value TEXT
)
);
CREATE TABLE IF NOT EXISTS %(prefix)sissuance (
serial TEXT PRIMARY KEY,
is_renewal INTEGER,
authorisation_serial TEXT,
not_valid_after INTEGER
);
''' % {
'prefix': table_prefix,
'key_id_constraint': 'UNIQUE' if enforce_unique_key_id else '',
......@@ -374,29 +381,6 @@ class SQLite3Storage(local):
).fetchall()
]
def storeCertificate(self, csr_id, crt):
"""
Store certificate for pre-existing CSR.
Raises NotFound if there is no matching CSR, or if a certificate was
already stored.
"""
with self._db as db:
c = db.cursor()
c.execute(
'UPDATE %scrt SET crt=?, expiration_date = ? '
'WHERE id = ? AND crt IS NULL' % (
self._table_prefix,
),
(
crt,
int(time() + self._crt_keep_time),
csr_id,
),
)
if c.rowcount == 0:
raise NotFound
def getCertificate(self, crt_id):
"""
Retrieve a PEM-encoded certificate.
......@@ -463,6 +447,82 @@ class SQLite3Storage(local):
break
yield toBytes(row['crt'])
@contextlib.contextmanager
def trackIssuance(self, is_renewal, authorisation_serial, serial, not_valid_after):
"""
Track certificate issuance.
is_renewal (bool)
If true, this is a renewal.
Otherwise, it is an original issuance.
authorisation_serial (int)
Serial of the certificate which authorised issuance:
- renewed certificate serial if is_renewal is true
- user certificate serial otherwise
- None if certificate was auto-signed
serial (int)
Serial of issued certificate.
not_valid_after (int)
Expiration timestamp of issued certificate.
Returns a context manager which, on entry raises Found if serial has
already been issued. Otherwise, returns a callable taking 2 parameters
storing certificate for a pre-existing CSR:
csr_id (int)
CSR identifier.
crt (str)
PEM-encoded certificate.
Raises NotFound if there is no matching CSR, or if a certificate was
already stored.
"""
with self._db as db:
now = int(time())
c = db.cursor()
c.execute(
'DELETE FROM %sissuance '
'WHERE not_valid_after < ?' % (
self._table_prefix,
),
(
now,
),
)
try:
c.execute(
'INSERT INTO %sissuance '
' (is_renewal, authorisation_serial, serial, not_valid_after) '
'VALUES (?, ?, ?, ?)' % (
self._table_prefix,
),
(
int(is_renewal),
str(authorisation_serial),
str(serial),
not_valid_after,
),
)
except sqlite3.IntegrityError: # pragma: no cover
raise Found
# Just to have a mutable
store_argument_list = []
yield lambda csr_id, crt: store_argument_list.append((csr_id, crt))
# pylint: disable=unbalanced-tuple-unpacking
(csr_id, crt), = store_argument_list
# pylint: enable=unbalanced-tuple-unpacking
c.execute(
'UPDATE %scrt SET crt=?, expiration_date = ? '
'WHERE id = ? AND crt IS NULL' % (
self._table_prefix,
),
(
crt,
now + self._crt_keep_time,
csr_id,
),
)
if c.rowcount == 0:
raise NotFound
def revoke(self, serial, expiration_date):
"""
Add given certificate serial to the list of revoked certificates.
......@@ -495,6 +555,74 @@ class SQLite3Storage(local):
except sqlite3.IntegrityError:
raise Found
def getIssuedBy(self, serial_list, renewal):
"""
Returns the list of serials of certificates which were issued, directly or
not, using the certificates with given serials.
serial_list (list of int)
Serial of the certificates to query the descendants of.
renewal (bool)
If true, only renewals will be followed.
Otherwise, initial issuances by given serials are fetched, and then
the renewals of these are followed. Renewals of given serials are not
followed.
"""
# do not mutate parameter
serial_list = list(serial_list)
result = set()
with self._db as db:
c = db.cursor()
while serial_list:
for next_serial, in c.execute(
'SELECT serial FROM %sissuance '
'WHERE authorisation_serial = ?%s' % (
self._table_prefix,
' and is_renewal = 1' if renewal else '',
),
(
str(serial_list.pop()),
),
):
next_serial = int(next_serial)
# Unlikely to be false: serials are server-enforced and random
# enough to be very unlikely to be reused. But still prevents a
# possible infinite loop.
if next_serial not in result:
serial_list.append(next_serial)
result.add(next_serial)
else: # pragma: no cover
pass
return list(result)
def getNonRevokedCertificateSerialList(self, serial_list):
"""
Return the list of serials whose certificates are not revoked, out of
those provided.
serial_list (list of int)
List of certificate serials to check for non-revocation.
"""
if serial_list:
with self._db as db:
revoked_serial_set = {
int(x)
for x, in db.cursor().execute(
'SELECT serial FROM %srevoked WHERE serial IN (%s)' % (
self._table_prefix,
','.join('?' * len(serial_list)),
),
[str(x) for x in serial_list],
)
}
else:
revoked_serial_set = ()
return [
x
for x in serial_list
if x not in revoked_serial_set
]
def getCertificateRevocationList(self):
"""
Get PEM-encoded current Certificate Revocation List.
......
caucase/test.py
View file @ 5fe1e86b
......@@ -24,8 +24,9 @@ Caucase - Certificate Authority for Users, Certificate Authority for SErvices
Test suite
"""
# pylint: disable=too-many-lines, too-many-public-methods
from __future__ import absolute_import
from __future__ import absolute_import, print_function
from Cookie import SimpleCookie
from collections import defaultdict
import datetime
# pylint: disable=no-name-in-module, import-error
from distutils.spawn import find_executable
......@@ -331,7 +332,7 @@ class CaucaseTest(unittest.TestCase):
Prepare test data directory and file paths, and start caucased as most
tests will need to interact with it.
"""
global _clean_caucased_snapshot
global _clean_caucased_snapshot # pylint: disable=global-statement
self._data_dir = data_dir = tempfile.mkdtemp(prefix='caucase_test_')
self._client_dir = client_dir = os.path.join(data_dir, 'client')
......@@ -1067,23 +1068,17 @@ class CaucaseTest(unittest.TestCase):
# and crt & key did not change
self.assertEqual(service2_crt_after, service2_crt_after2)
self.assertEqual(service2_key_after, service2_key_after2)
# revoking again one's own certificate fails
self.assertRaises(
CaucaseError,
self._runClient,
# revoking again one's own certificate does not fail
self._runClient(
'--revoke-crt', service2_key_path, '',
)
# as does revoking with an authenticated user
self.assertRaises(
CaucaseError,
self._runClient,
self._runClient(
'--user-key', user_key_path,
'--revoke-other-crt', service2_key_path,
)
# and revoking by serial
self.assertRaises(
CaucaseError,
self._runClient,
self._runClient(
'--user-key', user_key_path,
'--revoke-serial', str(
utils.load_certificate(
......@@ -2290,6 +2285,16 @@ class CaucaseTest(unittest.TestCase):
user_key_path,
'user',
)
user2_key_renewal_query_fragment = (
",1,'%i'" % utils.load_certificate(
utils.getCert(user2_key_path),
[
utils.load_ca_certificate(x)
for x in utils.getCertList(self._client_user_ca_crt)
],
None,
).serial_number
).encode('utf-8')
# user2 sacrifice their private key, and prepare its replacement
basename = self._getBaseName()
user2_new_key_path = self._createPrivateKey(basename)
......@@ -2377,12 +2382,14 @@ class CaucaseTest(unittest.TestCase):
CRL_INSERT = b'INSERT INTO "caucrl" '
CRT_INSERT = b'INSERT INTO "caucrt" '
REV_INSERT = b'INSERT INTO "caurevoked" '
ISSUED_INSERT = b'INSERT INTO "cauissuance" '
def filterBackup(backup, expect_rev):
"""
Remove all lines which are know to differ between original batabase and
post-restoration database, so the rest (which must be the majority of the
database) can be tested to be equal.
"""
renew_found = None
rev_found = not expect_rev
new_backup = []
crt_list = []
......@@ -2395,6 +2402,13 @@ class CaucaseTest(unittest.TestCase):
if row.startswith(REV_INSERT): # pragma: no cover
assert not rev_found, 'Unexpected revocation found'
continue
if (
row.startswith(ISSUED_INSERT) and
user2_key_renewal_query_fragment in row
):
assert renew_found is None, '\n%r\n%r' % (renew_found, row)
renew_found = row
continue
new_backup.append(row)
return new_backup, crt_list
......@@ -3127,6 +3141,220 @@ class CaucaseTest(unittest.TestCase):
)
self.assertFalse(user_certificate_policies.critical)
def testIssuanceTracking(self):
"""
Issue a few certificates, revoke a few and check that all their
descendents are listed.
"""
self._runClient()
cas_crt_list = [
utils.load_ca_certificate(x)
for x in utils.getCertList(self._client_ca_crt)
]
self._runClient('--mode', 'user')
cau_crt_list = [
utils.load_ca_certificate(x)
for x in utils.getCertList(self._client_user_ca_crt)
]
def getSerial(key_path, ca_crt_list):
'get serial as a string'
return str(utils.load_certificate(
utils.getCert(key_path),
ca_crt_list,
None,
).serial_number)
def approveUser(user_key_path):
'issue a new user certificate'
new_key_path = self._createAndApproveCertificate(
user_key_path,
'user',
)
return new_key_path, getSerial(new_key_path, cau_crt_list)
def approveService(user_key_path):
'issue a new service certificate'
new_key_path = self._createAndApproveCertificate(
user_key_path,
'service',
)
return new_key_path, getSerial(new_key_path, cas_crt_list)
def renew(mode, key_path):
'renew a <mode> certificate'
self._runClient(
'--mode', mode,
'--threshold', '100',
'--renew-crt', key_path, '',
)
def renewUser(key_path):
'renew a user certificate'
renew('user', key_path)
return getSerial(key_path, cau_crt_list)
def renewService(key_path):
'renew a service certificate'
renew('service', key_path)
return getSerial(key_path, cas_crt_list)
def assertRevokeResult(expected_dict, args, already_revoked=False):
'revoke a certificate and check displayed descendant certificates'
mode = None
output_dict = defaultdict(list)
output = self._runClient(*args)
already_revoked_found = False
for line in output.splitlines():
if line.endswith('was already revoked'):
already_revoked_found = True
elif line.startswith('mode: '):
_, mode = line.split()
elif mode is not None and line.startswith(' '):
output_dict[mode].append(line.strip())
self.assertEqual(already_revoked, already_revoked_found)
self.assertItemsEqual(expected_dict.keys(), output_dict.keys())
for mode, expected_value_list in expected_dict.iteritems():
self.assertItemsEqual(expected_value_list, output_dict[mode])
# Get first user certificate
user0_key_path = self._createFirstUser()
user0_serial0 = getSerial(user0_key_path, cau_crt_list)
# renew user0 certificate
user0_serial1 = renewUser(user0_key_path)
# Sanity check renewal
self.assertNotEqual(user0_serial0, user0_serial1)
# Get two more user certificates approved by user0
user1_key_path, user1_serial = approveUser(user0_key_path)
user2_key_path, user2_serial0 = approveUser(user0_key_path)
# renew user2 certificate
user2_serial1 = renewUser(user2_key_path)
# Get a service certificate approved by each user
service0_key_path, service0_serial0 = approveService(user0_key_path)
service1_key_path, service1_serial0 = approveService(user1_key_path)
service2_key_path, service2_serial0 = approveService(user2_key_path)
# Renew each
service0_serial1 = renewService(service0_key_path)
service1_serial1 = renewService(service1_key_path)
service2_serial1 = renewService(service2_key_path)
# because serials are unreadable, print a mapping to display if test fails
test_output = utils.toUnicodeWritableStream(self.caucase_test_output)
for serial, name in (
(user0_serial0, 'user0_serial0'),
(user0_serial1, 'user0_serial1'),
(user1_serial, 'user1_serial'),
(user2_serial0, 'user2_serial0'),
(user2_serial1, 'user2_serial1'),
(service0_serial0, 'service0_serial0'),
(service1_serial0, 'service1_serial0'),
(service2_serial0, 'service2_serial0'),
(service0_serial1, 'service0_serial1'),
(service1_serial1, 'service1_serial1'),
(service2_serial1, 'service2_serial1'),
):
print(serial, 'is', name, file=test_output)
assertRevokeResult(
{ # One renewal, which issued one service, which was renewed
'user': [user2_serial1],
'service': [service2_serial0, service2_serial1],
}, (
'--mode', 'user',
'--user-key', user0_key_path,
'--revoke-serial', user2_serial0,
),
)
assertRevokeResult(
{ # One renewal
'service': [service2_serial1],
}, (
'--mode', 'service',
'--user-key', user0_key_path,
'--revoke-serial', service2_serial0,
),
)
assertRevokeResult(
{}, ( # No renewals
'--mode', 'service',
'--revoke-crt', service1_key_path, '',
),
)
assertRevokeResult(
{}, ( # Renewal is already revoked
'--mode', 'service',
'--user-key', user0_key_path,
'--revoke-serial', service1_serial0,
),
)
assertRevokeResult(
{ # renewal, 2 issued user certificates (one being the renewal of an
# already-revoked certificate), one issued service and its renewal,
# and the renewal of and already-revoked issued service certificate
'user': [
user0_serial1,
user1_serial,
user2_serial1,
],
'service': [
service0_serial0, service0_serial1,
service2_serial1,
],
}, (
'--mode', 'user',
'--user-key', user1_key_path,
'--revoke-serial', user0_serial0,
),
)
assertRevokeResult(
{ # 2 issued user certificates (one being the renewal of an
# already-revoked certificate), one issued service and its renewal,
# and the renewal of and already-revoked issued service certificate
'user': [
user1_serial,
user2_serial1,
],
'service': [
service0_serial0, service0_serial1,
service2_serial1,
],
}, (
'--mode', 'user',
'--revoke-crt', user0_key_path, '',
),
)
# Duplicate revocations
# Must mention they are duplicate, but still display an updated view of
# descendant certificates.
assertRevokeResult(
{ # 2 issued user certificates (one being the renewal of an
# already-revoked certificate), one issued service and its renewal,
# and the renewal of and already-revoked issued service certificate
'user': [
user1_serial,
user2_serial1,
],
'service': [
service0_serial0, service0_serial1,
service2_serial1,
],
}, (
'--mode', 'user',
'--user-key', user1_key_path,
'--revoke-serial', user0_serial0,
),
already_revoked=True,
)
assertRevokeResult(
{ # 2 issued user certificates (one being the renewal of an
# already-revoked certificate), one issued service and its renewal,
# and the renewal of and already-revoked issued service certificate
'user': [
user1_serial,
user2_serial1,
],
'service': [
service0_serial0, service0_serial1,
service2_serial1,
],
}, (
'--mode', 'user',
'--revoke-crt', user0_key_path, '',
),
already_revoked=True,
)
for property_id, property_value in CaucaseTest.__dict__.iteritems():
if property_id.startswith('test') and callable(property_value):
setattr(CaucaseTest, property_id, print_buffer_on_error(property_value))
......
caucase/utils.py
View file @ 5fe1e86b
......@@ -44,6 +44,7 @@ import cryptography.exceptions
import pem
from .exceptions import (
CertificateVerificationError,
CertificateRevokedError,
NotJSON,
)
......@@ -342,10 +343,12 @@ def _verifyCertificateChain(cert, trusted_cert_list, crl):
Verifies whether certificate has been signed by any of the trusted
certificates, is not revoked and is whithin its validity period.
Raises CertificateVerificationError if validation fails.
Raises CertificateVerificationError if validation fails, or its
CertificateRevokedError subclass if it is because this certificate was
revoked.
"""
# Note: this function (validating a certificate without an SSL connection)
# does not seem to have many equivalents at all in python. OpenSSL module
# does not seem to have any equivalents at all in python. OpenSSL module
# seems to be a rare implementation of it, so we keep using this module.
# BUT it MUST NOT be used anywhere outside this function (hence the
# bad-style local import). Use "cryptography".
......@@ -362,13 +365,14 @@ def _verifyCertificateChain(cert, trusted_cert_list, crl):
store,
crypto.X509.from_cryptography(cert),
).verify_certificate()
except (
crypto.X509StoreContextError,
crypto.Error,
) as e:
raise CertificateVerificationError(
'Certificate verification error: %s' % str(e),
)
except crypto.X509StoreContextError as e:
error, depth, _ = e.args[0]
# 23 is X509_V_ERR_CERT_REVOKED (include/openssl/x509_vfy.h)
if error == 23 and depth == 0:
raise CertificateRevokedError(repr(e))
raise CertificateVerificationError(repr(e))
except crypto.Error as e:
raise CertificateVerificationError(repr(e))
def wrap(payload, key, digest):
"""
......
caucase/wsgi.py
View file @ 5fe1e86b
......@@ -23,6 +23,7 @@ Caucase - Certificate Authority for Users, Certificate Authority for SErvices
"""
from __future__ import absolute_import
from Cookie import SimpleCookie, CookieError
from functools import partial
import httplib
import json
import os
......@@ -180,6 +181,7 @@ STATUS_OK = _getStatus(httplib.OK)
STATUS_CREATED = _getStatus(httplib.CREATED)
STATUS_NO_CONTENT = _getStatus(httplib.NO_CONTENT)
STATUS_FOUND = _getStatus(httplib.FOUND)
STATUS_CONFLICT = Conflict.status
MAX_BODY_LENGTH = 10 * 1024 * 1024 # 10 MB
class CORSTokenManager(object):
......@@ -297,6 +299,7 @@ class Application(object):
List of Origin values to always trust.
"""
self._cau = cau
self._cas = cas
self._http_url = http_url.rstrip('/')
self._https_url = https_url.rstrip('/')
self._cors_cookie_id = cors_cookie_id
......@@ -577,8 +580,6 @@ class Application(object):
raise
except exceptions.NotFound:
raise NotFound
except exceptions.Found:
raise Conflict
except exceptions.NoStorage:
raise InsufficientStorage
except exceptions.NotJSON:
......@@ -604,12 +605,12 @@ class Application(object):
return result
@staticmethod
def _returnFile(data, content_type, header_list=None):
def _returnFile(data, content_type, header_list=None, status=STATUS_OK):
if header_list is None:
header_list = []
header_list.append(('Content-Type', content_type))
header_list.append(('Content-Length', str(len(data))))
return (STATUS_OK, header_list, [data])
return (status, header_list, [data])
@staticmethod
def _getCSRID(subpath):
......@@ -644,11 +645,12 @@ class Application(object):
Verify user authentication.
Raises SSLUnauthorized if authentication does not pass checks.
On success, appends a "Cache-Control" header.
On success, appends a "Cache-Control" header and returns certificate
object.
"""
try:
ca_list = self._cau.getCACertificateList()
utils.load_certificate(
result = utils.load_certificate(
environ.get('SSL_CLIENT_CERT', b''),
trusted_cert_list=ca_list,
crl=utils.load_crl(
......@@ -659,6 +661,7 @@ class Application(object):
except (exceptions.CertificateVerificationError, ValueError):
raise SSLUnauthorized
header_list.append(('Cache-Control', 'private'))
return result
def _readJSON(self, environ):
"""
......@@ -1049,6 +1052,34 @@ class Application(object):
'application/pkix-cert',
)
def _getIssuedBy(self, context, serial):
"""
Return a dict of non-revoked certificates descending from given serial.
Keys are modes, values are list of serials.
"""
result = {}
if context is self._cau:
# Revoking a user certificate, list:
# - other user certificates it issued and their renewals
# - service certificates they issued and their renewals
user_serial_list = context.getIssuedBy([serial], False)
service_serial_list = self._cas.getNonRevokedCertificateSerialList(
self._cas.getIssuedBy([serial] + user_serial_list, False),
)
user_serial_list = context.getNonRevokedCertificateSerialList(
user_serial_list,
)
if user_serial_list:
result['user'] = user_serial_list
else:
# Revoking a service certificate, list its renewals.
service_serial_list = context.getNonRevokedCertificateSerialList(
context.getIssuedBy([serial], True),
)
if service_serial_list:
result['service'] = service_serial_list
return result
def revokeCertificate(self, context, environ):
"""
Handle PUT /{context}/crt/revoke .
......@@ -1058,19 +1089,38 @@ class Application(object):
if data['digest'] is None:
self._authenticate(environ, header_list)
payload = utils.nullUnwrap(data)
if 'revoke_crt_pem' not in payload:
context.revokeSerial(payload['revoke_serial'])
return (STATUS_NO_CONTENT, header_list, [])
if 'revoke_crt_pem' in payload:
crt_pem = utils.toBytes(payload['revoke_crt_pem'])
revoke = partial(context.revoke, crt_pem=crt_pem)
else:
crt_pem = None
serial = payload['revoke_serial']
revoke = partial(context.revokeSerial, serial)
else:
payload = utils.unwrap(
crt_pem = utils.toBytes(utils.unwrap(
data,
lambda x: x['revoke_crt_pem'],
context.digest_list,
)
context.revoke(
crt_pem=utils.toBytes(payload['revoke_crt_pem']),
)['revoke_crt_pem'])
revoke = partial(context.revoke, crt_pem=crt_pem)
if crt_pem is not None:
serial = utils.load_certificate(
crt_pem,
context.getCACertificateList(),
None, # context will check its revocation status
).serial_number
try:
revoke()
except exceptions.Found:
status = STATUS_CONFLICT
else:
status = STATUS_OK
return self._returnFile(
json.dumps(self._getIssuedBy(context, serial)).encode('utf-8'),
'application/json',
header_list,
status,
)
return (STATUS_NO_CONTENT, header_list, [])
def renewCertificate(self, context, environ):
"""
......@@ -1104,9 +1154,10 @@ class Application(object):
else:
raise BadRequest(b'Bad Content-Type')
header_list = []
self._authenticate(environ, header_list)
user_crt = self._authenticate(environ, header_list)
context.createCertificate(
csr_id=crt_id,
template_csr=template_csr,
authorisation_serial=user_crt.serial_number,
)
return (STATUS_NO_CONTENT, header_list, [])
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7