## This file is part of Scapy
|
## Copyright (C) 2008 Arnaud Ebalard <arnaud.ebalard@eads.net>
|
## <arno@natisbad.org>
|
## 2015, 2016, 2017 Maxence Tury <maxence.tury@ssi.gouv.fr>
|
## This program is published under a GPLv2 license
|
|
"""
|
High-level methods for PKI objects (X.509 certificates, CRLs, asymmetric keys).
|
Supports both RSA and ECDSA objects.
|
|
The classes below are wrappers for the ASN.1 objects defined in x509.py.
|
By collecting their attributes, we bypass the ASN.1 structure, hence
|
there is no direct method for exporting a new full DER-encoded version
|
of a Cert instance after its serial has been modified (for example).
|
If you need to modify an import, just use the corresponding ASN1_Packet.
|
|
For instance, here is what you could do in order to modify the serial of
|
'cert' and then resign it with whatever 'key':
|
f = open('cert.der')
|
c = X509_Cert(f.read())
|
c.tbsCertificate.serialNumber = 0x4B1D
|
k = PrivKey('key.pem')
|
new_x509_cert = k.resignCert(c)
|
No need for obnoxious openssl tweaking anymore. :)
|
"""
|
|
from __future__ import absolute_import
|
from __future__ import print_function
|
import base64
|
import os
|
import time
|
|
from scapy.config import conf, crypto_validator
|
import scapy.modules.six as six
|
from scapy.modules.six.moves import range
|
if conf.crypto_valid:
|
from cryptography.hazmat.backends import default_backend
|
from cryptography.hazmat.primitives import serialization
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from scapy.error import warning
|
from scapy.utils import binrepr
|
from scapy.asn1.asn1 import ASN1_BIT_STRING
|
from scapy.asn1.mib import hash_by_oid
|
from scapy.layers.x509 import (X509_SubjectPublicKeyInfo,
|
RSAPublicKey, RSAPrivateKey,
|
ECDSAPublicKey, ECDSAPrivateKey,
|
RSAPrivateKey_OpenSSL, ECDSAPrivateKey_OpenSSL,
|
X509_Cert, X509_CRL)
|
from scapy.layers.tls.crypto.pkcs1 import (pkcs_os2ip, pkcs_i2osp, _get_hash,
|
_EncryptAndVerifyRSA,
|
_DecryptAndSignRSA)
|
|
from scapy.compat import *
|
|
# Maximum allowed size in bytes for a certificate file, to avoid
|
# loading huge file when importing a cert
|
_MAX_KEY_SIZE = 50*1024
|
_MAX_CERT_SIZE = 50*1024
|
_MAX_CRL_SIZE = 10*1024*1024 # some are that big
|
|
|
#####################################################################
|
# Some helpers
|
#####################################################################
|
|
@conf.commands.register
|
def der2pem(der_string, obj="UNKNOWN"):
|
"""Convert DER octet string to PEM format (with optional header)"""
|
# Encode a byte string in PEM format. Header advertizes <obj> type.
|
pem_string = ("-----BEGIN %s-----\n" % obj).encode()
|
base64_string = base64.b64encode(der_string)
|
chunks = [base64_string[i:i+64] for i in range(0, len(base64_string), 64)]
|
pem_string += b'\n'.join(chunks)
|
pem_string += ("\n-----END %s-----\n" % obj).encode()
|
return pem_string
|
|
@conf.commands.register
|
def pem2der(pem_string):
|
"""Convert PEM string to DER format"""
|
# Encode all lines between the first '-----\n' and the 2nd-to-last '-----'.
|
pem_string = pem_string.replace(b"\r", b"")
|
first_idx = pem_string.find(b"-----\n") + 6
|
if pem_string.find(b"-----BEGIN", first_idx) != -1:
|
raise Exception("pem2der() expects only one PEM-encoded object")
|
last_idx = pem_string.rfind(b"-----", 0, pem_string.rfind(b"-----"))
|
base64_string = pem_string[first_idx:last_idx]
|
base64_string.replace(b"\n", b"")
|
der_string = base64.b64decode(base64_string)
|
return der_string
|
|
def split_pem(s):
|
"""
|
Split PEM objects. Useful to process concatenated certificates.
|
"""
|
pem_strings = []
|
while s != b"":
|
start_idx = s.find(b"-----BEGIN")
|
if start_idx == -1:
|
break
|
end_idx = s.find(b"-----END")
|
end_idx = s.find(b"\n", end_idx) + 1
|
pem_strings.append(s[start_idx:end_idx])
|
s = s[end_idx:]
|
return pem_strings
|
|
|
class _PKIObj(object):
|
def __init__(self, frmt, der, pem):
|
# Note that changing attributes of the _PKIObj does not update these
|
# values (e.g. modifying k.modulus does not change k.der).
|
#XXX use __setattr__ for this
|
self.frmt = frmt
|
self.der = der
|
self.pem = pem
|
|
def __str__(self):
|
return self.der
|
|
|
class _PKIObjMaker(type):
|
def __call__(cls, obj_path, obj_max_size, pem_marker=None):
|
# This enables transparent DER and PEM-encoded data imports.
|
# Note that when importing a PEM file with multiple objects (like ECDSA
|
# private keys output by openssl), it will concatenate every object in
|
# order to create a 'der' attribute. When converting a 'multi' DER file
|
# into a PEM file, though, the PEM attribute will not be valid,
|
# because we do not try to identify the class of each object.
|
error_msg = "Unable to import data"
|
|
if obj_path is None:
|
raise Exception(error_msg)
|
obj_path = raw(obj_path)
|
|
if (not b'\x00' in obj_path) and os.path.isfile(obj_path):
|
_size = os.path.getsize(obj_path)
|
if _size > obj_max_size:
|
raise Exception(error_msg)
|
try:
|
f = open(obj_path, "rb")
|
_raw = f.read()
|
f.close()
|
except:
|
raise Exception(error_msg)
|
else:
|
_raw = obj_path
|
|
try:
|
if b"-----BEGIN" in _raw:
|
frmt = "PEM"
|
pem = _raw
|
der_list = split_pem(_raw)
|
der = b''.join(map(pem2der, der_list))
|
else:
|
frmt = "DER"
|
der = _raw
|
pem = ""
|
if pem_marker is not None:
|
pem = der2pem(_raw, pem_marker)
|
# type identification may be needed for pem_marker
|
# in such case, the pem attribute has to be updated
|
except:
|
raise Exception(error_msg)
|
|
p = _PKIObj(frmt, der, pem)
|
return p
|
|
|
#####################################################################
|
# PKI objects wrappers
|
#####################################################################
|
|
###############
|
# Public Keys #
|
###############
|
|
class _PubKeyFactory(_PKIObjMaker):
|
"""
|
Metaclass for PubKey creation.
|
It casts the appropriate class on the fly, then fills in
|
the appropriate attributes with import_from_asn1pkt() submethod.
|
"""
|
def __call__(cls, key_path=None):
|
|
if key_path is None:
|
obj = type.__call__(cls)
|
if cls is PubKey:
|
cls = PubKeyRSA
|
obj.__class__ = cls
|
obj.frmt = "original"
|
obj.fill_and_store()
|
return obj
|
|
# This deals with the rare RSA 'kx export' call.
|
if isinstance(key_path, tuple):
|
obj = type.__call__(cls)
|
obj.__class__ = PubKeyRSA
|
obj.frmt = "tuple"
|
obj.import_from_tuple(key_path)
|
return obj
|
|
# Now for the usual calls, key_path may be the path to either:
|
# _an X509_SubjectPublicKeyInfo, as processed by openssl;
|
# _an RSAPublicKey;
|
# _an ECDSAPublicKey.
|
obj = _PKIObjMaker.__call__(cls, key_path, _MAX_KEY_SIZE)
|
try:
|
spki = X509_SubjectPublicKeyInfo(obj.der)
|
pubkey = spki.subjectPublicKey
|
if isinstance(pubkey, RSAPublicKey):
|
obj.__class__ = PubKeyRSA
|
obj.import_from_asn1pkt(pubkey)
|
elif isinstance(pubkey, ECDSAPublicKey):
|
obj.__class__ = PubKeyECDSA
|
try:
|
obj.import_from_der(obj.der)
|
except ImportError:
|
pass
|
else:
|
raise
|
marker = b"PUBLIC KEY"
|
except:
|
try:
|
pubkey = RSAPublicKey(obj.der)
|
obj.__class__ = PubKeyRSA
|
obj.import_from_asn1pkt(pubkey)
|
marker = b"RSA PUBLIC KEY"
|
except:
|
# We cannot import an ECDSA public key without curve knowledge
|
raise Exception("Unable to import public key")
|
|
if obj.frmt == "DER":
|
obj.pem = der2pem(obj.der, marker)
|
return obj
|
|
|
class PubKey(six.with_metaclass(_PubKeyFactory, object)):
|
"""
|
Parent class for both PubKeyRSA and PubKeyECDSA.
|
Provides a common verifyCert() method.
|
"""
|
|
def verifyCert(self, cert):
|
""" Verifies either a Cert or an X509_Cert. """
|
tbsCert = cert.tbsCertificate
|
sigAlg = tbsCert.signature
|
h = hash_by_oid[sigAlg.algorithm.val]
|
sigVal = raw(cert.signatureValue)
|
return self.verify(raw(tbsCert), sigVal, h=h, t='pkcs')
|
|
|
class PubKeyRSA(PubKey, _EncryptAndVerifyRSA):
|
"""
|
Wrapper for RSA keys based on _EncryptAndVerifyRSA from crypto/pkcs1.py
|
Use the 'key' attribute to access original object.
|
"""
|
@crypto_validator
|
def fill_and_store(self, modulus=None, modulusLen=None, pubExp=None):
|
pubExp = pubExp or 65537
|
if not modulus:
|
real_modulusLen = modulusLen or 2048
|
private_key = rsa.generate_private_key(public_exponent=pubExp,
|
key_size=real_modulusLen,
|
backend=default_backend())
|
self.pubkey = private_key.public_key()
|
else:
|
real_modulusLen = len(binrepr(modulus))
|
if modulusLen and real_modulusLen != modulusLen:
|
warning("modulus and modulusLen do not match!")
|
pubNum = rsa.RSAPublicNumbers(n=modulus, e=pubExp)
|
self.pubkey = pubNum.public_key(default_backend())
|
# Lines below are only useful for the legacy part of pkcs1.py
|
pubNum = self.pubkey.public_numbers()
|
self._modulusLen = real_modulusLen
|
self._modulus = pubNum.n
|
self._pubExp = pubNum.e
|
|
@crypto_validator
|
def import_from_tuple(self, tup):
|
# this is rarely used
|
e, m, mLen = tup
|
if isinstance(m, bytes):
|
m = pkcs_os2ip(m)
|
if isinstance(e, bytes):
|
e = pkcs_os2ip(e)
|
self.fill_and_store(modulus=m, pubExp=e)
|
self.pem = self.pubkey.public_bytes(
|
encoding=serialization.Encoding.PEM,
|
format=serialization.PublicFormat.SubjectPublicKeyInfo)
|
self.der = pem2der(self.pem)
|
|
def import_from_asn1pkt(self, pubkey):
|
modulus = pubkey.modulus.val
|
pubExp = pubkey.publicExponent.val
|
self.fill_and_store(modulus=modulus, pubExp=pubExp)
|
|
def encrypt(self, msg, t="pkcs", h="sha256", mgf=None, L=None):
|
# no ECDSA encryption support, hence no ECDSA specific keywords here
|
return _EncryptAndVerifyRSA.encrypt(self, msg, t, h, mgf, L)
|
|
def verify(self, msg, sig, t="pkcs", h="sha256", mgf=None, L=None):
|
return _EncryptAndVerifyRSA.verify(self, msg, sig, t, h, mgf, L)
|
|
class PubKeyECDSA(PubKey):
|
"""
|
Wrapper for ECDSA keys based on the cryptography library.
|
Use the 'key' attribute to access original object.
|
"""
|
@crypto_validator
|
def fill_and_store(self, curve=None):
|
curve = curve or ec.SECP256R1
|
private_key = ec.generate_private_key(curve(), default_backend())
|
self.pubkey = private_key.public_key()
|
|
@crypto_validator
|
def import_from_der(self, pubkey):
|
# No lib support for explicit curves nor compressed points.
|
self.pubkey = serialization.load_der_public_key(pubkey,
|
backend=default_backend())
|
|
def encrypt(self, msg, h="sha256", **kwargs):
|
# cryptography lib does not support ECDSA encryption
|
raise Exception("No ECDSA encryption support")
|
|
@crypto_validator
|
def verify(self, msg, sig, h="sha256", **kwargs):
|
# 'sig' should be a DER-encoded signature, as per RFC 3279
|
verifier = self.pubkey.verifier(sig, ec.ECDSA(_get_hash(h)))
|
verifier.update(msg)
|
return verifier.verify()
|
|
|
################
|
# Private Keys #
|
################
|
|
class _PrivKeyFactory(_PKIObjMaker):
|
"""
|
Metaclass for PrivKey creation.
|
It casts the appropriate class on the fly, then fills in
|
the appropriate attributes with import_from_asn1pkt() submethod.
|
"""
|
def __call__(cls, key_path=None):
|
"""
|
key_path may be the path to either:
|
_an RSAPrivateKey_OpenSSL (as generated by openssl);
|
_an ECDSAPrivateKey_OpenSSL (as generated by openssl);
|
_an RSAPrivateKey;
|
_an ECDSAPrivateKey.
|
"""
|
if key_path is None:
|
obj = type.__call__(cls)
|
if cls is PrivKey:
|
cls = PrivKeyECDSA
|
obj.__class__ = cls
|
obj.frmt = "original"
|
obj.fill_and_store()
|
return obj
|
|
obj = _PKIObjMaker.__call__(cls, key_path, _MAX_KEY_SIZE)
|
multiPEM = False
|
try:
|
privkey = RSAPrivateKey_OpenSSL(obj.der)
|
privkey = privkey.privateKey
|
obj.__class__ = PrivKeyRSA
|
marker = b"PRIVATE KEY"
|
except:
|
try:
|
privkey = ECDSAPrivateKey_OpenSSL(obj.der)
|
privkey = privkey.privateKey
|
obj.__class__ = PrivKeyECDSA
|
marker = b"EC PRIVATE KEY"
|
multiPEM = True
|
except:
|
try:
|
privkey = RSAPrivateKey(obj.der)
|
obj.__class__ = PrivKeyRSA
|
marker = b"RSA PRIVATE KEY"
|
except:
|
try:
|
privkey = ECDSAPrivateKey(obj.der)
|
obj.__class__ = PrivKeyECDSA
|
marker = b"EC PRIVATE KEY"
|
except:
|
raise Exception("Unable to import private key")
|
try:
|
obj.import_from_asn1pkt(privkey)
|
except ImportError:
|
pass
|
|
if obj.frmt == "DER":
|
if multiPEM:
|
# this does not restore the EC PARAMETERS header
|
obj.pem = der2pem(raw(privkey), marker)
|
else:
|
obj.pem = der2pem(obj.der, marker)
|
return obj
|
|
|
class PrivKey(six.with_metaclass(_PrivKeyFactory, object)):
|
"""
|
Parent class for both PrivKeyRSA and PrivKeyECDSA.
|
Provides common signTBSCert() and resignCert() methods.
|
"""
|
|
def signTBSCert(self, tbsCert, h="sha256"):
|
"""
|
Note that this will always copy the signature field from the
|
tbsCertificate into the signatureAlgorithm field of the result,
|
regardless of the coherence between its contents (which might
|
indicate ecdsa-with-SHA512) and the result (e.g. RSA signing MD2).
|
|
There is a small inheritance trick for the computation of sigVal
|
below: in order to use a sign() method which would apply
|
to both PrivKeyRSA and PrivKeyECDSA, the sign() methods of the
|
subclasses accept any argument, be it from the RSA or ECDSA world,
|
and then they keep the ones they're interested in.
|
Here, t will be passed eventually to pkcs1._DecryptAndSignRSA.sign().
|
"""
|
sigAlg = tbsCert.signature
|
h = h or hash_by_oid[sigAlg.algorithm.val]
|
sigVal = self.sign(raw(tbsCert), h=h, t='pkcs')
|
c = X509_Cert()
|
c.tbsCertificate = tbsCert
|
c.signatureAlgorithm = sigAlg
|
c.signatureValue = ASN1_BIT_STRING(sigVal, readable=True)
|
return c
|
|
def resignCert(self, cert):
|
""" Rewrite the signature of either a Cert or an X509_Cert. """
|
return self.signTBSCert(cert.tbsCertificate)
|
|
def verifyCert(self, cert):
|
""" Verifies either a Cert or an X509_Cert. """
|
tbsCert = cert.tbsCertificate
|
sigAlg = tbsCert.signature
|
h = hash_by_oid[sigAlg.algorithm.val]
|
sigVal = raw(cert.signatureValue)
|
return self.verify(raw(tbsCert), sigVal, h=h, t='pkcs')
|
|
|
class PrivKeyRSA(PrivKey, _EncryptAndVerifyRSA, _DecryptAndSignRSA):
|
"""
|
Wrapper for RSA keys based on _DecryptAndSignRSA from crypto/pkcs1.py
|
Use the 'key' attribute to access original object.
|
"""
|
@crypto_validator
|
def fill_and_store(self, modulus=None, modulusLen=None, pubExp=None,
|
prime1=None, prime2=None, coefficient=None,
|
exponent1=None, exponent2=None, privExp=None):
|
pubExp = pubExp or 65537
|
if None in [modulus, prime1, prime2, coefficient, privExp,
|
exponent1, exponent2]:
|
# note that the library requires every parameter
|
# in order to call RSAPrivateNumbers(...)
|
# if one of these is missing, we generate a whole new key
|
real_modulusLen = modulusLen or 2048
|
self.key = rsa.generate_private_key(public_exponent=pubExp,
|
key_size=real_modulusLen,
|
backend=default_backend())
|
self.pubkey = self.key.public_key()
|
else:
|
real_modulusLen = len(binrepr(modulus))
|
if modulusLen and real_modulusLen != modulusLen:
|
warning("modulus and modulusLen do not match!")
|
pubNum = rsa.RSAPublicNumbers(n=modulus, e=pubExp)
|
privNum = rsa.RSAPrivateNumbers(p=prime1, q=prime2,
|
dmp1=exponent1, dmq1=exponent2,
|
iqmp=coefficient, d=privExp,
|
public_numbers=pubNum)
|
self.key = privNum.private_key(default_backend())
|
self.pubkey = self.key.public_key()
|
|
# Lines below are only useful for the legacy part of pkcs1.py
|
pubNum = self.pubkey.public_numbers()
|
self._modulusLen = real_modulusLen
|
self._modulus = pubNum.n
|
self._pubExp = pubNum.e
|
|
def import_from_asn1pkt(self, privkey):
|
modulus = privkey.modulus.val
|
pubExp = privkey.publicExponent.val
|
privExp = privkey.privateExponent.val
|
prime1 = privkey.prime1.val
|
prime2 = privkey.prime2.val
|
exponent1 = privkey.exponent1.val
|
exponent2 = privkey.exponent2.val
|
coefficient = privkey.coefficient.val
|
self.fill_and_store(modulus=modulus, pubExp=pubExp,
|
privExp=privExp, prime1=prime1, prime2=prime2,
|
exponent1=exponent1, exponent2=exponent2,
|
coefficient=coefficient)
|
|
def verify(self, msg, sig, t="pkcs", h="sha256", mgf=None, L=None):
|
# Let's copy this from PubKeyRSA instead of adding another baseclass :)
|
return _EncryptAndVerifyRSA.verify(self, msg, sig, t, h, mgf, L)
|
|
def sign(self, data, t="pkcs", h="sha256", mgf=None, L=None):
|
return _DecryptAndSignRSA.sign(self, data, t, h, mgf, L)
|
|
|
class PrivKeyECDSA(PrivKey):
|
"""
|
Wrapper for ECDSA keys based on SigningKey from ecdsa library.
|
Use the 'key' attribute to access original object.
|
"""
|
@crypto_validator
|
def fill_and_store(self, curve=None):
|
curve = curve or ec.SECP256R1
|
self.key = ec.generate_private_key(curve(), default_backend())
|
self.pubkey = self.key.public_key()
|
|
@crypto_validator
|
def import_from_asn1pkt(self, privkey):
|
self.key = serialization.load_der_private_key(raw(privkey), None,
|
backend=default_backend())
|
self.pubkey = self.key.public_key()
|
|
@crypto_validator
|
def verify(self, msg, sig, h="sha256", **kwargs):
|
# 'sig' should be a DER-encoded signature, as per RFC 3279
|
verifier = self.pubkey.verifier(sig, ec.ECDSA(_get_hash(h)))
|
verifier.update(msg)
|
return verifier.verify()
|
|
@crypto_validator
|
def sign(self, data, h="sha256", **kwargs):
|
signer = self.key.signer(ec.ECDSA(_get_hash(h)))
|
signer.update(data)
|
return signer.finalize()
|
|
|
################
|
# Certificates #
|
################
|
|
class _CertMaker(_PKIObjMaker):
|
"""
|
Metaclass for Cert creation. It is not necessary as it was for the keys,
|
but we reuse the model instead of creating redundant constructors.
|
"""
|
def __call__(cls, cert_path):
|
obj = _PKIObjMaker.__call__(cls, cert_path,
|
_MAX_CERT_SIZE, "CERTIFICATE")
|
obj.__class__ = Cert
|
try:
|
cert = X509_Cert(obj.der)
|
except:
|
raise Exception("Unable to import certificate")
|
obj.import_from_asn1pkt(cert)
|
return obj
|
|
|
class Cert(six.with_metaclass(_CertMaker, object)):
|
"""
|
Wrapper for the X509_Cert from layers/x509.py.
|
Use the 'x509Cert' attribute to access original object.
|
"""
|
|
def import_from_asn1pkt(self, cert):
|
error_msg = "Unable to import certificate"
|
|
self.x509Cert = cert
|
|
tbsCert = cert.tbsCertificate
|
self.tbsCertificate = tbsCert
|
|
if tbsCert.version:
|
self.version = tbsCert.version.val + 1
|
else:
|
self.version = 1
|
self.serial = tbsCert.serialNumber.val
|
self.sigAlg = tbsCert.signature.algorithm.oidname
|
self.issuer = tbsCert.get_issuer()
|
self.issuer_str = tbsCert.get_issuer_str()
|
self.issuer_hash = hash(self.issuer_str)
|
self.subject = tbsCert.get_subject()
|
self.subject_str = tbsCert.get_subject_str()
|
self.subject_hash = hash(self.subject_str)
|
|
self.notBefore_str = tbsCert.validity.not_before.pretty_time
|
notBefore = tbsCert.validity.not_before.val
|
if notBefore[-1] == "Z":
|
notBefore = notBefore[:-1]
|
try:
|
self.notBefore = time.strptime(notBefore, "%y%m%d%H%M%S")
|
except:
|
raise Exception(error_msg)
|
self.notBefore_str_simple = time.strftime("%x", self.notBefore)
|
|
self.notAfter_str = tbsCert.validity.not_after.pretty_time
|
notAfter = tbsCert.validity.not_after.val
|
if notAfter[-1] == "Z":
|
notAfter = notAfter[:-1]
|
try:
|
self.notAfter = time.strptime(notAfter, "%y%m%d%H%M%S")
|
except:
|
raise Exception(error_msg)
|
self.notAfter_str_simple = time.strftime("%x", self.notAfter)
|
|
self.pubKey = PubKey(raw(tbsCert.subjectPublicKeyInfo))
|
|
if tbsCert.extensions:
|
for extn in tbsCert.extensions:
|
if extn.extnID.oidname == "basicConstraints":
|
self.cA = False
|
if extn.extnValue.cA:
|
self.cA = not (extn.extnValue.cA.val == 0)
|
elif extn.extnID.oidname == "keyUsage":
|
self.keyUsage = extn.extnValue.get_keyUsage()
|
elif extn.extnID.oidname == "extKeyUsage":
|
self.extKeyUsage = extn.extnValue.get_extendedKeyUsage()
|
elif extn.extnID.oidname == "authorityKeyIdentifier":
|
self.authorityKeyID = extn.extnValue.keyIdentifier.val
|
|
self.signatureValue = raw(cert.signatureValue)
|
self.signatureLen = len(self.signatureValue)
|
|
def isIssuerCert(self, other):
|
"""
|
True if 'other' issued 'self', i.e.:
|
- self.issuer == other.subject
|
- self is signed by other
|
"""
|
if self.issuer_hash != other.subject_hash:
|
return False
|
return other.pubKey.verifyCert(self)
|
|
def isSelfSigned(self):
|
"""
|
Return True if the certificate is self-signed:
|
- issuer and subject are the same
|
- the signature of the certificate is valid.
|
"""
|
if self.issuer_hash == self.subject_hash:
|
return self.isIssuerCert(self)
|
return False
|
|
def encrypt(self, msg, t="pkcs", h="sha256", mgf=None, L=None):
|
# no ECDSA *encryption* support, hence only RSA specific keywords here
|
return self.pubKey.encrypt(msg, t, h, mgf, L)
|
|
def verify(self, msg, sig, t="pkcs", h="sha256", mgf=None, L=None):
|
return self.pubKey.verify(msg, sig, t, h, mgf, L)
|
|
def remainingDays(self, now=None):
|
"""
|
Based on the value of notAfter field, returns the number of
|
days the certificate will still be valid. The date used for the
|
comparison is the current and local date, as returned by
|
time.localtime(), except if 'now' argument is provided another
|
one. 'now' argument can be given as either a time tuple or a string
|
representing the date. Accepted format for the string version
|
are:
|
|
- '%b %d %H:%M:%S %Y %Z' e.g. 'Jan 30 07:38:59 2008 GMT'
|
- '%m/%d/%y' e.g. '01/30/08' (less precise)
|
|
If the certificate is no more valid at the date considered, then
|
a negative value is returned representing the number of days
|
since it has expired.
|
|
The number of days is returned as a float to deal with the unlikely
|
case of certificates that are still just valid.
|
"""
|
if now is None:
|
now = time.localtime()
|
elif isinstance(now, str):
|
try:
|
if '/' in now:
|
now = time.strptime(now, '%m/%d/%y')
|
else:
|
now = time.strptime(now, '%b %d %H:%M:%S %Y %Z')
|
except:
|
warning("Bad time string provided, will use localtime() instead.")
|
now = time.localtime()
|
|
now = time.mktime(now)
|
nft = time.mktime(self.notAfter)
|
diff = (nft - now)/(24.*3600)
|
return diff
|
|
def isRevoked(self, crl_list):
|
"""
|
Given a list of trusted CRL (their signature has already been
|
verified with trusted anchors), this function returns True if
|
the certificate is marked as revoked by one of those CRL.
|
|
Note that if the Certificate was on hold in a previous CRL and
|
is now valid again in a new CRL and bot are in the list, it
|
will be considered revoked: this is because _all_ CRLs are
|
checked (not only the freshest) and revocation status is not
|
handled.
|
|
Also note that the check on the issuer is performed on the
|
Authority Key Identifier if available in _both_ the CRL and the
|
Cert. Otherwise, the issuers are simply compared.
|
"""
|
for c in crl_list:
|
if (self.authorityKeyID is not None and
|
c.authorityKeyID is not None and
|
self.authorityKeyID == c.authorityKeyID):
|
return self.serial in (x[0] for x in c.revoked_cert_serials)
|
elif self.issuer == c.issuer:
|
return self.serial in (x[0] for x in c.revoked_cert_serials)
|
return False
|
|
def export(self, filename, fmt="DER"):
|
"""
|
Export certificate in 'fmt' format (DER or PEM) to file 'filename'
|
"""
|
f = open(filename, "wb")
|
if fmt == "DER":
|
f.write(self.der)
|
elif fmt == "PEM":
|
f.write(self.pem)
|
f.close()
|
|
def show(self):
|
print("Serial: %s" % self.serial)
|
print("Issuer: " + self.issuer_str)
|
print("Subject: " + self.subject_str)
|
print("Validity: %s to %s" % (self.notBefore_str, self.notAfter_str))
|
|
def __repr__(self):
|
return "[X.509 Cert. Subject:%s, Issuer:%s]" % (self.subject_str, self.issuer_str)
|
|
|
################################
|
# Certificate Revocation Lists #
|
################################
|
|
class _CRLMaker(_PKIObjMaker):
|
"""
|
Metaclass for CRL creation. It is not necessary as it was for the keys,
|
but we reuse the model instead of creating redundant constructors.
|
"""
|
def __call__(cls, cert_path):
|
obj = _PKIObjMaker.__call__(cls, cert_path, _MAX_CRL_SIZE, "X509 CRL")
|
obj.__class__ = CRL
|
try:
|
crl = X509_CRL(obj.der)
|
except:
|
raise Exception("Unable to import CRL")
|
obj.import_from_asn1pkt(crl)
|
return obj
|
|
|
class CRL(six.with_metaclass(_CRLMaker, object)):
|
"""
|
Wrapper for the X509_CRL from layers/x509.py.
|
Use the 'x509CRL' attribute to access original object.
|
"""
|
|
def import_from_asn1pkt(self, crl):
|
error_msg = "Unable to import CRL"
|
|
self.x509CRL = crl
|
|
tbsCertList = crl.tbsCertList
|
self.tbsCertList = raw(tbsCertList)
|
|
if tbsCertList.version:
|
self.version = tbsCertList.version.val + 1
|
else:
|
self.version = 1
|
self.sigAlg = tbsCertList.signature.algorithm.oidname
|
self.issuer = tbsCertList.get_issuer()
|
self.issuer_str = tbsCertList.get_issuer_str()
|
self.issuer_hash = hash(self.issuer_str)
|
|
self.lastUpdate_str = tbsCertList.this_update.pretty_time
|
lastUpdate = tbsCertList.this_update.val
|
if lastUpdate[-1] == "Z":
|
lastUpdate = lastUpdate[:-1]
|
try:
|
self.lastUpdate = time.strptime(lastUpdate, "%y%m%d%H%M%S")
|
except:
|
raise Exception(error_msg)
|
self.lastUpdate_str_simple = time.strftime("%x", self.lastUpdate)
|
|
self.nextUpdate = None
|
self.nextUpdate_str_simple = None
|
if tbsCertList.next_update:
|
self.nextUpdate_str = tbsCertList.next_update.pretty_time
|
nextUpdate = tbsCertList.next_update.val
|
if nextUpdate[-1] == "Z":
|
nextUpdate = nextUpdate[:-1]
|
try:
|
self.nextUpdate = time.strptime(nextUpdate, "%y%m%d%H%M%S")
|
except:
|
raise Exception(error_msg)
|
self.nextUpdate_str_simple = time.strftime("%x", self.nextUpdate)
|
|
if tbsCertList.crlExtensions:
|
for extension in tbsCertList.crlExtensions:
|
if extension.extnID.oidname == "cRLNumber":
|
self.number = extension.extnValue.cRLNumber.val
|
|
revoked = []
|
if tbsCertList.revokedCertificates:
|
for cert in tbsCertList.revokedCertificates:
|
serial = cert.serialNumber.val
|
date = cert.revocationDate.val
|
if date[-1] == "Z":
|
date = date[:-1]
|
try:
|
revocationDate = time.strptime(date, "%y%m%d%H%M%S")
|
except:
|
raise Exception(error_msg)
|
revoked.append((serial, date))
|
self.revoked_cert_serials = revoked
|
|
self.signatureValue = raw(crl.signatureValue)
|
self.signatureLen = len(self.signatureValue)
|
|
def isIssuerCert(self, other):
|
# This is exactly the same thing as in Cert method.
|
if self.issuer_hash != other.subject_hash:
|
return False
|
return other.pubKey.verifyCert(self)
|
|
def verify(self, anchors):
|
# Return True iff the CRL is signed by one of the provided anchors.
|
for a in anchors:
|
if self.isIssuerCert(a):
|
return True
|
return False
|
|
def show(self):
|
print("Version: %d" % self.version)
|
print("sigAlg: " + self.sigAlg)
|
print("Issuer: " + self.issuer_str)
|
print("lastUpdate: %s" % self.lastUpdate_str)
|
print("nextUpdate: %s" % self.nextUpdate_str)
|
|
|
######################
|
# Certificate chains #
|
######################
|
|
class Chain(list):
|
"""
|
Basically, an enhanced array of Cert.
|
"""
|
def __init__(self, certList, cert0=None):
|
"""
|
Construct a chain of certificates starting with a self-signed
|
certificate (or any certificate submitted by the user)
|
and following issuer/subject matching and signature validity.
|
If there is exactly one chain to be constructed, it will be,
|
but if there are multiple potential chains, there is no guarantee
|
that the retained one will be the longest one.
|
As Cert and CRL classes both share an isIssuerCert() method,
|
the trailing element of a Chain may alternatively be a CRL.
|
|
Note that we do not check AKID/{SKID/issuer/serial} matching,
|
nor the presence of keyCertSign in keyUsage extension (if present).
|
"""
|
list.__init__(self, ())
|
if cert0:
|
self.append(cert0)
|
else:
|
for root_candidate in certList:
|
if root_candidate.isSelfSigned():
|
self.append(root_candidate)
|
certList.remove(root_candidate)
|
break
|
|
if len(self) > 0:
|
while certList:
|
l = len(self)
|
for c in certList:
|
if c.isIssuerCert(self[-1]):
|
self.append(c)
|
certList.remove(c)
|
break
|
if len(self) == l:
|
# no new certificate appended to self
|
break
|
|
def verifyChain(self, anchors, untrusted=None):
|
"""
|
Perform verification of certificate chains for that certificate.
|
A list of anchors is required. The certificates in the optional
|
untrusted list may be used as additional elements to the final chain.
|
On par with chain instantiation, only one chain constructed with the
|
untrusted candidates will be retained. Eventually, dates are checked.
|
"""
|
untrusted = untrusted or []
|
for a in anchors:
|
chain = Chain(self + untrusted, a)
|
if len(chain) == 1: # anchor only
|
continue
|
# check that the chain does not exclusively rely on untrusted
|
found = False
|
for c in self:
|
if c in chain[1:]:
|
found = True
|
if found:
|
for c in chain:
|
if c.remainingDays() < 0:
|
break
|
if c is chain[-1]: # we got to the end of the chain
|
return chain
|
return None
|
|
def verifyChainFromCAFile(self, cafile, untrusted_file=None):
|
"""
|
Does the same job as .verifyChain() but using the list of anchors
|
from the cafile. As for .verifyChain(), a list of untrusted
|
certificates can be passed (as a file, this time).
|
"""
|
try:
|
f = open(cafile)
|
ca_certs = f.read()
|
f.close()
|
except:
|
raise Exception("Could not read from cafile")
|
|
anchors = [Cert(c) for c in split_pem(ca_certs)]
|
|
untrusted = None
|
if untrusted_file:
|
try:
|
f = open(untrusted_file)
|
untrusted_certs = f.read()
|
f.close()
|
except:
|
raise Exception("Could not read from untrusted_file")
|
untrusted = [Cert(c) for c in split_pem(untrusted_certs)]
|
|
return self.verifyChain(anchors, untrusted)
|
|
def verifyChainFromCAPath(self, capath, untrusted_file=None):
|
"""
|
Does the same job as .verifyChainFromCAFile() but using the list
|
of anchors in capath directory. The directory should (only) contain
|
certificates files in PEM format. As for .verifyChainFromCAFile(),
|
a list of untrusted certificates can be passed as a file
|
(concatenation of the certificates in PEM format).
|
"""
|
try:
|
anchors = []
|
for cafile in os.listdir(capath):
|
anchors.append(Cert(open(cafile).read()))
|
except:
|
raise Exception("capath provided is not a valid cert path")
|
|
untrusted = None
|
if untrusted_file:
|
try:
|
f = open(untrusted_file)
|
untrusted_certs = f.read()
|
f.close()
|
except:
|
raise Exception("Could not read from untrusted_file")
|
untrusted = [Cert(c) for c in split_pem(untrusted_certs)]
|
|
return self.verifyChain(anchors, untrusted)
|
|
def __repr__(self):
|
llen = len(self) - 1
|
if llen < 0:
|
return ""
|
c = self[0]
|
s = "__ "
|
if not c.isSelfSigned():
|
s += "%s [Not Self Signed]\n" % c.subject_str
|
else:
|
s += "%s [Self Signed]\n" % c.subject_str
|
idx = 1
|
while idx <= llen:
|
c = self[idx]
|
s += "%s\_ %s" % (" "*idx*2, c.subject_str)
|
if idx != llen:
|
s += "\n"
|
idx += 1
|
return s
|
|
|
##############################
|
# Certificate export helpers #
|
##############################
|
|
def _create_ca_file(anchor_list, filename):
|
"""
|
Concatenate all the certificates (PEM format for the export) in
|
'anchor_list' and write the result to file 'filename'. On success
|
'filename' is returned, None otherwise.
|
|
If you are used to OpenSSL tools, this function builds a CAfile
|
that can be used for certificate and CRL check.
|
"""
|
try:
|
f = open(filename, "w")
|
for a in anchor_list:
|
s = a.output(fmt="PEM")
|
f.write(s)
|
f.close()
|
except IOError:
|
return None
|
return filename
|
