1 --- /dev/null Thu Jan 01 00:00:00 1970 +0000
2 +++ b/scapy/crypto/__init__.py Mon Nov 02 22:09:11 2009 +0100
3 @@ -0,0 +1,6 @@
4 +## This file is part of Scapy
5 +## See http://www.secdev.org/projects/scapy for more informations
6 +## Copyright (C) Arnaud Ebalard <arno@natisbad.org>
7 +## This program is published under a GPLv2 license
8 +
9 +__all__ = ["cert"]
1.1 --- /dev/null Thu Jan 01 00:00:00 1970 +0000
1.2 +++ b/scapy/crypto/cert.py Mon Nov 02 22:09:11 2009 +0100
1.3 @@ -0,0 +1,2470 @@
1.4 +## This file is part of Scapy
1.5 +## See http://www.secdev.org/projects/scapy for more informations
1.6 +## Copyright (C) Arnaud Ebalard <arno@natisbad.org>
1.7 +## This program is published under a GPLv2 license
1.8 +
1.9 +import os, sys, math, socket, struct, sha, hmac, string, time
1.10 +import random, popen2, tempfile
1.11 +from scapy.utils import strxor
1.12 +try:
1.13 + HAS_HASHLIB=True
1.14 + import hashlib
1.15 +except:
1.16 + HAS_HASHLIB=False
1.17 +
1.18 +from Crypto.PublicKey import *
1.19 +from Crypto.Cipher import *
1.20 +from Crypto.Hash import *
1.21 +
1.22 +# Maximum allowed size in bytes for a certificate file, to avoid
1.23 +# loading huge file when importing a cert
1.24 +MAX_KEY_SIZE=50*1024
1.25 +MAX_CERT_SIZE=50*1024
1.26 +MAX_CRL_SIZE=10*1024*1024 # some are that big
1.27 +
1.28 +#####################################################################
1.29 +# Some helpers
1.30 +#####################################################################
1.31 +
1.32 +def warning(m):
1.33 + print "WARNING: %s" % m
1.34 +
1.35 +def randstring(l):
1.36 + """
1.37 + Returns a random string of length l (l >= 0)
1.38 + """
1.39 + tmp = map(lambda x: struct.pack("B", random.randrange(0, 256, 1)), [""]*l)
1.40 + return "".join(tmp)
1.41 +
1.42 +def zerofree_randstring(l):
1.43 + """
1.44 + Returns a random string of length l (l >= 0) without zero in it.
1.45 + """
1.46 + tmp = map(lambda x: struct.pack("B", random.randrange(1, 256, 1)), [""]*l)
1.47 + return "".join(tmp)
1.48 +
1.49 +def strand(s1, s2):
1.50 + """
1.51 + Returns the binary AND of the 2 provided strings s1 and s2. s1 and s2
1.52 + must be of same length.
1.53 + """
1.54 + return "".join(map(lambda x,y:chr(ord(x)&ord(y)), s1, s2))
1.55 +
1.56 +# OS2IP function defined in RFC 3447 for octet string to integer conversion
1.57 +def pkcs_os2ip(x):
1.58 + """
1.59 + Accepts a byte string as input parameter and return the associated long
1.60 + value:
1.61 +
1.62 + Input : x octet string to be converted
1.63 +
1.64 + Output: x corresponding nonnegative integer
1.65 +
1.66 + Reverse function is pkcs_i2osp()
1.67 + """
1.68 + return RSA.number.bytes_to_long(x)
1.69 +
1.70 +# IP2OS function defined in RFC 3447 for octet string to integer conversion
1.71 +def pkcs_i2osp(x,xLen):
1.72 + """
1.73 + Converts a long (the first parameter) to the associated byte string
1.74 + representation of length l (second parameter). Basically, the length
1.75 + parameters allow the function to perform the associated padding.
1.76 +
1.77 + Input : x nonnegative integer to be converted
1.78 + xLen intended length of the resulting octet string
1.79 +
1.80 + Output: x corresponding nonnegative integer
1.81 +
1.82 + Reverse function is pkcs_os2ip().
1.83 + """
1.84 + z = RSA.number.long_to_bytes(x)
1.85 + padlen = max(0, xLen-len(z))
1.86 + return '\x00'*padlen + z
1.87 +
1.88 +# for every hash function a tuple is provided, giving access to
1.89 +# - hash output length in byte
1.90 +# - associated hash function that take data to be hashed as parameter
1.91 +# XXX I do not provide update() at the moment.
1.92 +# - DER encoding of the leading bits of digestInfo (the hash value
1.93 +# will be concatenated to create the complete digestInfo).
1.94 +#
1.95 +# Notes:
1.96 +# - MD4 asn.1 value should be verified. Also, as stated in
1.97 +# PKCS#1 v2.1, MD4 should not be used.
1.98 +# - hashlib is available from http://code.krypto.org/python/hashlib/
1.99 +# - 'tls' one is the concatenation of both md5 and sha1 hashes used
1.100 +# by SSL/TLS when signing/verifying things
1.101 +_hashFuncParams = {
1.102 + "md2" : (16,
1.103 + lambda x: MD2.new(x).digest(),
1.104 + '\x30\x20\x30\x0c\x06\x08\x2a\x86\x48\x86\xf7\x0d\x02\x02\x05\x00\x04\x10'),
1.105 + "md4" : (16,
1.106 + lambda x: MD4.new(x).digest(),
1.107 + '\x30\x20\x30\x0c\x06\x08\x2a\x86\x48\x86\xf7\x0d\x02\x04\x05\x00\x04\x10'), # is that right ?
1.108 + "md5" : (16,
1.109 + lambda x: MD5.new(x).digest(),
1.110 + '\x30\x20\x30\x0c\x06\x08\x2a\x86\x48\x86\xf7\x0d\x02\x05\x05\x00\x04\x10'),
1.111 + "sha1" : (20,
1.112 + lambda x: SHA.new(x).digest(),
1.113 + '\x30\x21\x30\x09\x06\x05\x2b\x0e\x03\x02\x1a\x05\x00\x04\x14'),
1.114 + "tls" : (36,
1.115 + lambda x: MD5.new(x).digest() + SHA.new(x).digest(),
1.116 + '') }
1.117 +
1.118 +if HAS_HASHLIB:
1.119 + _hashFuncParams["sha224"] = (28,
1.120 + lambda x: hashlib.sha224(x).digest(),
1.121 + '\x30\x2d\x30\x0d\x06\x09\x60\x86\x48\x01\x65\x03\x04\x02\x04\x05\x00\x04\x1c')
1.122 + _hashFuncParams["sha256"] = (32,
1.123 + lambda x: hashlib.sha256(x).digest(),
1.124 + '\x30\x31\x30\x0d\x06\x09\x60\x86\x48\x01\x65\x03\x04\x02\x01\x05\x00\x04\x20')
1.125 + _hashFuncParams["sha384"] = (48,
1.126 + lambda x: hashlib.sha384(x).digest(),
1.127 + '\x30\x41\x30\x0d\x06\x09\x60\x86\x48\x01\x65\x03\x04\x02\x02\x05\x00\x04\x30')
1.128 + _hashFuncParams["sha512"] = (64,
1.129 + lambda x: hashlib.sha512(x).digest(),
1.130 + '\x30\x51\x30\x0d\x06\x09\x60\x86\x48\x01\x65\x03\x04\x02\x03\x05\x00\x04\x40')
1.131 +else:
1.132 + warning("hashlib support is not available. Consider installing it")
1.133 + warning("if you need sha224, sha256, sha384 and sha512 algs.")
1.134 +
1.135 +def pkcs_mgf1(mgfSeed, maskLen, h):
1.136 + """
1.137 + Implements generic MGF1 Mask Generation function as described in
1.138 + Appendix B.2.1 of RFC 3447. The hash function is passed by name.
1.139 + valid values are 'md2', 'md4', 'md5', 'sha1', 'tls, 'sha256',
1.140 + 'sha384' and 'sha512'. Returns None on error.
1.141 +
1.142 + Input:
1.143 + mgfSeed: seed from which mask is generated, an octet string
1.144 + maskLen: intended length in octets of the mask, at most 2^32 * hLen
1.145 + hLen (see below)
1.146 + h : hash function name (in 'md2', 'md4', 'md5', 'sha1', 'tls',
1.147 + 'sha256', 'sha384'). hLen denotes the length in octets of
1.148 + the hash function output.
1.149 +
1.150 + Output:
1.151 + an octet string of length maskLen
1.152 + """
1.153 +
1.154 + # steps are those of Appendix B.2.1
1.155 + if not _hashFuncParams.has_key(h):
1.156 + warning("pkcs_mgf1: invalid hash (%s) provided")
1.157 + return None
1.158 + hLen = _hashFuncParams[h][0]
1.159 + hFunc = _hashFuncParams[h][1]
1.160 + if maskLen > 2**32 * hLen: # 1)
1.161 + warning("pkcs_mgf1: maskLen > 2**32 * hLen")
1.162 + return None
1.163 + T = "" # 2)
1.164 + maxCounter = math.ceil(float(maskLen) / float(hLen)) # 3)
1.165 + counter = 0
1.166 + while counter < maxCounter:
1.167 + C = pkcs_i2osp(counter, 4)
1.168 + T += hFunc(mgfSeed + C)
1.169 + counter += 1
1.170 + return T[:maskLen]
1.171 +
1.172 +
1.173 +def pkcs_emsa_pss_encode(M, emBits, h, mgf, sLen):
1.174 + """
1.175 + Implements EMSA-PSS-ENCODE() function described in Sect. 9.1.1 of RFC 3447
1.176 +
1.177 + Input:
1.178 + M : message to be encoded, an octet string
1.179 + emBits: maximal bit length of the integer resulting of pkcs_os2ip(EM),
1.180 + where EM is the encoded message, output of the function.
1.181 + h : hash function name (in 'md2', 'md4', 'md5', 'sha1', 'tls',
1.182 + 'sha256', 'sha384'). hLen denotes the length in octets of
1.183 + the hash function output.
1.184 + mgf : the mask generation function f : seed, maskLen -> mask
1.185 + sLen : intended length in octets of the salt
1.186 +
1.187 + Output:
1.188 + encoded message, an octet string of length emLen = ceil(emBits/8)
1.189 +
1.190 + On error, None is returned.
1.191 + """
1.192 +
1.193 + # 1) is not done
1.194 + hLen = _hashFuncParams[h][0] # 2)
1.195 + hFunc = _hashFuncParams[h][1]
1.196 + mHash = hFunc(M)
1.197 + emLen = int(math.ceil(emBits/8.))
1.198 + if emLen < hLen + sLen + 2: # 3)
1.199 + warning("encoding error (emLen < hLen + sLen + 2)")
1.200 + return None
1.201 + salt = randstring(sLen) # 4)
1.202 + MPrime = '\x00'*8 + mHash + salt # 5)
1.203 + H = hFunc(MPrime) # 6)
1.204 + PS = '\x00'*(emLen - sLen - hLen - 2) # 7)
1.205 + DB = PS + '\x01' + salt # 8)
1.206 + dbMask = mgf(H, emLen - hLen - 1) # 9)
1.207 + maskedDB = strxor(DB, dbMask) # 10)
1.208 + l = (8*emLen - emBits)/8 # 11)
1.209 + rem = 8*emLen - emBits - 8*l # additionnal bits
1.210 + andMask = l*'\x00'
1.211 + if rem:
1.212 + j = chr(reduce(lambda x,y: x+y, map(lambda x: 1<<x, range(8-rem))))
1.213 + andMask += j
1.214 + l += 1
1.215 + maskedDB = strand(maskedDB[:l], andMask) + maskedDB[l:]
1.216 + EM = maskedDB + H + '\xbc' # 12)
1.217 + return EM # 13)
1.218 +
1.219 +
1.220 +def pkcs_emsa_pss_verify(M, EM, emBits, h, mgf, sLen):
1.221 + """
1.222 + Implements EMSA-PSS-VERIFY() function described in Sect. 9.1.2 of RFC 3447
1.223 +
1.224 + Input:
1.225 + M : message to be encoded, an octet string
1.226 + EM : encoded message, an octet string of length emLen = ceil(emBits/8)
1.227 + emBits: maximal bit length of the integer resulting of pkcs_os2ip(EM)
1.228 + h : hash function name (in 'md2', 'md4', 'md5', 'sha1', 'tls',
1.229 + 'sha256', 'sha384'). hLen denotes the length in octets of
1.230 + the hash function output.
1.231 + mgf : the mask generation function f : seed, maskLen -> mask
1.232 + sLen : intended length in octets of the salt
1.233 +
1.234 + Output:
1.235 + True if the verification is ok, False otherwise.
1.236 + """
1.237 +
1.238 + # 1) is not done
1.239 + hLen = _hashFuncParams[h][0] # 2)
1.240 + hFunc = _hashFuncParams[h][1]
1.241 + mHash = hFunc(M)
1.242 + emLen = int(math.ceil(emBits/8.)) # 3)
1.243 + if emLen < hLen + sLen + 2:
1.244 + return False
1.245 + if EM[-1] != '\xbc': # 4)
1.246 + return False
1.247 + l = emLen - hLen - 1 # 5)
1.248 + maskedDB = EM[:l]
1.249 + H = EM[l:l+hLen]
1.250 + l = (8*emLen - emBits)/8 # 6)
1.251 + rem = 8*emLen - emBits - 8*l # additionnal bits
1.252 + andMask = l*'\xff'
1.253 + if rem:
1.254 + val = reduce(lambda x,y: x+y, map(lambda x: 1<<x, range(8-rem)))
1.255 + j = chr(~val & 0xff)
1.256 + andMask += j
1.257 + l += 1
1.258 + if strand(maskedDB[:l], andMask) != '\x00'*l:
1.259 + return False
1.260 + dbMask = mgf(H, emLen - hLen - 1) # 7)
1.261 + DB = strxor(maskedDB, dbMask) # 8)
1.262 + l = (8*emLen - emBits)/8 # 9)
1.263 + rem = 8*emLen - emBits - 8*l # additionnal bits
1.264 + andMask = l*'\x00'
1.265 + if rem:
1.266 + j = chr(reduce(lambda x,y: x+y, map(lambda x: 1<<x, range(8-rem))))
1.267 + andMask += j
1.268 + l += 1
1.269 + DB = strand(DB[:l], andMask) + DB[l:]
1.270 + l = emLen - hLen - sLen - 1 # 10)
1.271 + if DB[:l] != '\x00'*(l-1) + '\x01':
1.272 + return False
1.273 + salt = DB[-sLen:] # 11)
1.274 + MPrime = '\x00'*8 + mHash + salt # 12)
1.275 + HPrime = hFunc(MPrime) # 13)
1.276 + return H == HPrime # 14)
1.277 +
1.278 +
1.279 +def pkcs_emsa_pkcs1_v1_5_encode(M, emLen, h): # section 9.2 of RFC 3447
1.280 + """
1.281 + Implements EMSA-PKCS1-V1_5-ENCODE() function described in Sect.
1.282 + 9.2 of RFC 3447.
1.283 +
1.284 + Input:
1.285 + M : message to be encode, an octet string
1.286 + emLen: intended length in octets of the encoded message, at least
1.287 + tLen + 11, where tLen is the octet length of the DER encoding
1.288 + T of a certain value computed during the encoding operation.
1.289 + h : hash function name (in 'md2', 'md4', 'md5', 'sha1', 'tls',
1.290 + 'sha256', 'sha384'). hLen denotes the length in octets of
1.291 + the hash function output.
1.292 +
1.293 + Output:
1.294 + encoded message, an octet string of length emLen
1.295 +
1.296 + On error, None is returned.
1.297 + """
1.298 + hLen = _hashFuncParams[h][0] # 1)
1.299 + hFunc = _hashFuncParams[h][1]
1.300 + H = hFunc(M)
1.301 + hLeadingDigestInfo = _hashFuncParams[h][2] # 2)
1.302 + T = hLeadingDigestInfo + H
1.303 + tLen = len(T)
1.304 + if emLen < tLen + 11: # 3)
1.305 + warning("pkcs_emsa_pkcs1_v1_5_encode: intended encoded message length too short")
1.306 + return None
1.307 + PS = '\xff'*(emLen - tLen - 3) # 4)
1.308 + EM = '\x00' + '\x01' + PS + '\x00' + T # 5)
1.309 + return EM # 6)
1.310 +
1.311 +
1.312 +# XXX should add other pgf1 instance in a better fashion.
1.313 +
1.314 +def create_ca_file(anchor_list, filename):
1.315 + """
1.316 + Concatenate all the certificates (PEM format for the export) in
1.317 + 'anchor_list' and write the result to file 'filename'. On success
1.318 + 'filename' is returned, None otherwise.
1.319 +
1.320 + If you are used to OpenSSL tools, this function builds a CAfile
1.321 + that can be used for certificate and CRL check.
1.322 +
1.323 + Also see create_temporary_ca_file().
1.324 + """
1.325 + try:
1.326 + f = open(filename, "w")
1.327 + for a in anchor_list:
1.328 + s = a.output(fmt="PEM")
1.329 + f.write(s)
1.330 + f.close()
1.331 + except:
1.332 + return None
1.333 + return filename
1.334 +
1.335 +def create_temporary_ca_file(anchor_list):
1.336 + """
1.337 + Concatenate all the certificates (PEM format for the export) in
1.338 + 'anchor_list' and write the result to file to a temporary file
1.339 + using mkstemp() from tempfile module. On success 'filename' is
1.340 + returned, None otherwise.
1.341 +
1.342 + If you are used to OpenSSL tools, this function builds a CAfile
1.343 + that can be used for certificate and CRL check.
1.344 +
1.345 + Also see create_temporary_ca_file().
1.346 + """
1.347 + try:
1.348 + f, fname = tempfile.mkstemp()
1.349 + for a in anchor_list:
1.350 + s = a.output(fmt="PEM")
1.351 + l = os.write(f, s)
1.352 + os.close(f)
1.353 + except:
1.354 + return None
1.355 + return fname
1.356 +
1.357 +def create_temporary_ca_path(anchor_list, folder):
1.358 + """
1.359 + Create a CA path folder as defined in OpenSSL terminology, by
1.360 + storing all certificates in 'anchor_list' list in PEM format
1.361 + under provided 'folder' and then creating the associated links
1.362 + using the hash as usually done by c_rehash.
1.363 +
1.364 + Note that you can also include CRL in 'anchor_list'. In that
1.365 + case, they will also be stored under 'folder' and associated
1.366 + links will be created.
1.367 +
1.368 + In folder, the files are created with names of the form
1.369 + 0...ZZ.pem. If you provide an empty list, folder will be created
1.370 + if it does not already exist, but that's all.
1.371 +
1.372 + The number of certificates written to folder is returned on
1.373 + success, None on error.
1.374 + """
1.375 + # We should probably avoid writing duplicate anchors and also
1.376 + # check if they are all certs.
1.377 + try:
1.378 + if not os.path.isdir(folder):
1.379 + os.makedirs(folder)
1.380 + except:
1.381 + return None
1.382 +
1.383 + l = len(anchor_list)
1.384 + if l == 0:
1.385 + return None
1.386 + fmtstr = "%%0%sd.pem" % math.ceil(math.log(l, 10))
1.387 + i = 0
1.388 + try:
1.389 + for a in anchor_list:
1.390 + fname = os.path.join(folder, fmtstr % i)
1.391 + f = open(fname, "w")
1.392 + s = a.output(fmt="PEM")
1.393 + f.write(s)
1.394 + f.close()
1.395 + i += 1
1.396 + except:
1.397 + return None
1.398 +
1.399 + r,w=popen2.popen2("c_rehash %s" % folder)
1.400 + r.close(); w.close()
1.401 +
1.402 + return l
1.403 +
1.404 +
1.405 +#####################################################################
1.406 +# Public Key Cryptography related stuff
1.407 +#####################################################################
1.408 +
1.409 +class OSSLHelper:
1.410 + def _apply_ossl_cmd(self, osslcmd, rawdata):
1.411 + r,w=popen2.popen2(osslcmd)
1.412 + w.write(rawdata)
1.413 + w.close()
1.414 + res = r.read()
1.415 + r.close()
1.416 + return res
1.417 +
1.418 +class _EncryptAndVerify:
1.419 + ### Below are encryption methods
1.420 +
1.421 + def _rsaep(self, m):
1.422 + """
1.423 + Internal method providing raw RSA encryption, i.e. simple modular
1.424 + exponentiation of the given message representative 'm', a long
1.425 + between 0 and n-1.
1.426 +
1.427 + This is the encryption primitive RSAEP described in PKCS#1 v2.1,
1.428 + i.e. RFC 3447 Sect. 5.1.1.
1.429 +
1.430 + Input:
1.431 + m: message representative, a long between 0 and n-1, where
1.432 + n is the key modulus.
1.433 +
1.434 + Output:
1.435 + ciphertext representative, a long between 0 and n-1
1.436 +
1.437 + Not intended to be used directly. Please, see encrypt() method.
1.438 + """
1.439 +
1.440 + n = self.modulus
1.441 + if type(m) is int:
1.442 + m = long(m)
1.443 + if type(m) is not long or m > n-1:
1.444 + warning("Key._rsaep() expects a long between 0 and n-1")
1.445 + return None
1.446 +
1.447 + return self.key.encrypt(m, "")[0]
1.448 +
1.449 +
1.450 + def _rsaes_pkcs1_v1_5_encrypt(self, M):
1.451 + """
1.452 + Implements RSAES-PKCS1-V1_5-ENCRYPT() function described in section
1.453 + 7.2.1 of RFC 3447.
1.454 +
1.455 + Input:
1.456 + M: message to be encrypted, an octet string of length mLen, where
1.457 + mLen <= k - 11 (k denotes the length in octets of the key modulus)
1.458 +
1.459 + Output:
1.460 + ciphertext, an octet string of length k
1.461 +
1.462 + On error, None is returned.
1.463 + """
1.464 +
1.465 + # 1) Length checking
1.466 + mLen = len(M)
1.467 + k = self.modulusLen / 8
1.468 + if mLen > k - 11:
1.469 + warning("Key._rsaes_pkcs1_v1_5_encrypt(): message too "
1.470 + "long (%d > %d - 11)" % (mLen, k))
1.471 + return None
1.472 +
1.473 + # 2) EME-PKCS1-v1_5 encoding
1.474 + PS = zerofree_randstring(k - mLen - 3) # 2.a)
1.475 + EM = '\x00' + '\x02' + PS + '\x00' + M # 2.b)
1.476 +
1.477 + # 3) RSA encryption
1.478 + m = pkcs_os2ip(EM) # 3.a)
1.479 + c = self._rsaep(m) # 3.b)
1.480 + C = pkcs_i2osp(c, k) # 3.c)
1.481 +
1.482 + return C # 4)
1.483 +
1.484 +
1.485 + def _rsaes_oaep_encrypt(self, M, h=None, mgf=None, L=None):
1.486 + """
1.487 + Internal method providing RSAES-OAEP-ENCRYPT as defined in Sect.
1.488 + 7.1.1 of RFC 3447. Not intended to be used directly. Please, see
1.489 + encrypt() method for type "OAEP".
1.490 +
1.491 +
1.492 + Input:
1.493 + M : message to be encrypted, an octet string of length mLen
1.494 + where mLen <= k - 2*hLen - 2 (k denotes the length in octets
1.495 + of the RSA modulus and hLen the length in octets of the hash
1.496 + function output)
1.497 + h : hash function name (in 'md2', 'md4', 'md5', 'sha1', 'tls',
1.498 + 'sha256', 'sha384'). hLen denotes the length in octets of
1.499 + the hash function output. 'sha1' is used by default if not
1.500 + provided.
1.501 + mgf: the mask generation function f : seed, maskLen -> mask
1.502 + L : optional label to be associated with the message; the default
1.503 + value for L, if not provided is the empty string
1.504 +
1.505 + Output:
1.506 + ciphertext, an octet string of length k
1.507 +
1.508 + On error, None is returned.
1.509 + """
1.510 + # The steps below are the one described in Sect. 7.1.1 of RFC 3447.
1.511 + # 1) Length Checking
1.512 + # 1.a) is not done
1.513 + mLen = len(M)
1.514 + if h is None:
1.515 + h = "sha1"
1.516 + if not _hashFuncParams.has_key(h):
1.517 + warning("Key._rsaes_oaep_encrypt(): unknown hash function %s.", h)
1.518 + return None
1.519 + hLen = _hashFuncParams[h][0]
1.520 + hFun = _hashFuncParams[h][1]
1.521 + k = self.modulusLen / 8
1.522 + if mLen > k - 2*hLen - 2: # 1.b)
1.523 + warning("Key._rsaes_oaep_encrypt(): message too long.")
1.524 + return None
1.525 +
1.526 + # 2) EME-OAEP encoding
1.527 + if L is None: # 2.a)
1.528 + L = ""
1.529 + lHash = hFun(L)
1.530 + PS = '\x00'*(k - mLen - 2*hLen - 2) # 2.b)
1.531 + DB = lHash + PS + '\x01' + M # 2.c)
1.532 + seed = randstring(hLen) # 2.d)
1.533 + if mgf is None: # 2.e)
1.534 + mgf = lambda x,y: pkcs_mgf1(x,y,h)
1.535 + dbMask = mgf(seed, k - hLen - 1)
1.536 + maskedDB = strxor(DB, dbMask) # 2.f)
1.537 + seedMask = mgf(maskedDB, hLen) # 2.g)
1.538 + maskedSeed = strxor(seed, seedMask) # 2.h)
1.539 + EM = '\x00' + maskedSeed + maskedDB # 2.i)
1.540 +
1.541 + # 3) RSA Encryption
1.542 + m = pkcs_os2ip(EM) # 3.a)
1.543 + c = self._rsaep(m) # 3.b)
1.544 + C = pkcs_i2osp(c, k) # 3.c)
1.545 +
1.546 + return C # 4)
1.547 +
1.548 +
1.549 + def encrypt(self, m, t=None, h=None, mgf=None, L=None):
1.550 + """
1.551 + Encrypt message 'm' using 't' encryption scheme where 't' can be:
1.552 +
1.553 + - None: the message 'm' is directly applied the RSAEP encryption
1.554 + primitive, as described in PKCS#1 v2.1, i.e. RFC 3447
1.555 + Sect 5.1.1. Simply put, the message undergo a modular
1.556 + exponentiation using the public key. Additionnal method
1.557 + parameters are just ignored.
1.558 +
1.559 + - 'pkcs': the message 'm' is applied RSAES-PKCS1-V1_5-ENCRYPT encryption
1.560 + scheme as described in section 7.2.1 of RFC 3447. In that
1.561 + context, other parameters ('h', 'mgf', 'l') are not used.
1.562 +
1.563 + - 'oaep': the message 'm' is applied the RSAES-OAEP-ENCRYPT encryption
1.564 + scheme, as described in PKCS#1 v2.1, i.e. RFC 3447 Sect
1.565 + 7.1.1. In that context,
1.566 +
1.567 + o 'h' parameter provides the name of the hash method to use.
1.568 + Possible values are "md2", "md4", "md5", "sha1", "tls",
1.569 + "sha224", "sha256", "sha384" and "sha512". if none is provided,
1.570 + sha1 is used.
1.571 +
1.572 + o 'mgf' is the mask generation function. By default, mgf
1.573 + is derived from the provided hash function using the
1.574 + generic MGF1 (see pkcs_mgf1() for details).
1.575 +
1.576 + o 'L' is the optional label to be associated with the
1.577 + message. If not provided, the default value is used, i.e
1.578 + the empty string. No check is done on the input limitation
1.579 + of the hash function regarding the size of 'L' (for
1.580 + instance, 2^61 - 1 for SHA-1). You have been warned.
1.581 + """
1.582 +
1.583 + if t is None: # Raw encryption
1.584 + m = pkcs_os2ip(m)
1.585 + c = self._rsaep(m)
1.586 + return pkcs_i2osp(c, self.modulusLen/8)
1.587 +
1.588 + elif t == "pkcs":
1.589 + return self._rsaes_pkcs1_v1_5_encrypt(m)
1.590 +
1.591 + elif t == "oaep":
1.592 + return self._rsaes_oaep_encrypt(m, h, mgf, L)
1.593 +
1.594 + else:
1.595 + warning("Key.encrypt(): Unknown encryption type (%s) provided" % t)
1.596 + return None
1.597 +
1.598 + ### Below are verification related methods
1.599 +
1.600 + def _rsavp1(self, s):
1.601 + """
1.602 + Internal method providing raw RSA verification, i.e. simple modular
1.603 + exponentiation of the given signature representative 'c', an integer
1.604 + between 0 and n-1.
1.605 +
1.606 + This is the signature verification primitive RSAVP1 described in
1.607 + PKCS#1 v2.1, i.e. RFC 3447 Sect. 5.2.2.
1.608 +
1.609 + Input:
1.610 + s: signature representative, an integer between 0 and n-1,
1.611 + where n is the key modulus.
1.612 +
1.613 + Output:
1.614 + message representative, an integer between 0 and n-1
1.615 +
1.616 + Not intended to be used directly. Please, see verify() method.
1.617 + """
1.618 + return self._rsaep(s)
1.619 +
1.620 + def _rsassa_pss_verify(self, M, S, h=None, mgf=None, sLen=None):
1.621 + """
1.622 + Implements RSASSA-PSS-VERIFY() function described in Sect 8.1.2
1.623 + of RFC 3447
1.624 +
1.625 + Input:
1.626 + M: message whose signature is to be verified
1.627 + S: signature to be verified, an octet string of length k, where k
1.628 + is the length in octets of the RSA modulus n.
1.629 +
1.630 + Output:
1.631 + True is the signature is valid. False otherwise.
1.632 + """
1.633 +
1.634 + # Set default parameters if not provided
1.635 + if h is None: # By default, sha1
1.636 + h = "sha1"
1.637 + if not _hashFuncParams.has_key(h):
1.638 + warning("Key._rsassa_pss_verify(): unknown hash function "
1.639 + "provided (%s)" % h)
1.640 + return False
1.641 + if mgf is None: # use mgf1 with underlying hash function
1.642 + mgf = lambda x,y: pkcs_mgf1(x, y, h)
1.643 + if sLen is None: # use Hash output length (A.2.3 of RFC 3447)
1.644 + hLen = _hashFuncParams[h][0]
1.645 + sLen = hLen
1.646 +
1.647 + # 1) Length checking
1.648 + modBits = self.modulusLen
1.649 + k = modBits / 8
1.650 + if len(S) != k:
1.651 + return False
1.652 +
1.653 + # 2) RSA verification
1.654 + s = pkcs_os2ip(S) # 2.a)
1.655 + m = self._rsavp1(s) # 2.b)
1.656 + emLen = math.ceil((modBits - 1) / 8.) # 2.c)
1.657 + EM = pkcs_i2osp(m, emLen)
1.658 +
1.659 + # 3) EMSA-PSS verification
1.660 + Result = pkcs_emsa_pss_verify(M, EM, modBits - 1, h, mgf, sLen)
1.661 +
1.662 + return Result # 4)
1.663 +
1.664 +
1.665 + def _rsassa_pkcs1_v1_5_verify(self, M, S, h):
1.666 + """
1.667 + Implements RSASSA-PKCS1-v1_5-VERIFY() function as described in
1.668 + Sect. 8.2.2 of RFC 3447.
1.669 +
1.670 + Input:
1.671 + M: message whose signature is to be verified, an octet string
1.672 + S: signature to be verified, an octet string of length k, where
1.673 + k is the length in octets of the RSA modulus n
1.674 + h: hash function name (in 'md2', 'md4', 'md5', 'sha1', 'tls',
1.675 + 'sha256', 'sha384').
1.676 +
1.677 + Output:
1.678 + True if the signature is valid. False otherwise.
1.679 + """
1.680 +
1.681 + # 1) Length checking
1.682 + k = self.modulusLen / 8
1.683 + if len(S) != k:
1.684 + warning("invalid signature (len(S) != k)")
1.685 + return False
1.686 +
1.687 + # 2) RSA verification
1.688 + s = pkcs_os2ip(S) # 2.a)
1.689 + m = self._rsavp1(s) # 2.b)
1.690 + EM = pkcs_i2osp(m, k) # 2.c)
1.691 +
1.692 + # 3) EMSA-PKCS1-v1_5 encoding
1.693 + EMPrime = pkcs_emsa_pkcs1_v1_5_encode(M, k, h)
1.694 + if EMPrime is None:
1.695 + warning("Key._rsassa_pkcs1_v1_5_verify(): unable to encode.")
1.696 + return False
1.697 +
1.698 + # 4) Comparison
1.699 + return EM == EMPrime
1.700 +
1.701 +
1.702 + def verify(self, M, S, t=None, h=None, mgf=None, sLen=None):
1.703 + """
1.704 + Verify alleged signature 'S' is indeed the signature of message 'M' using
1.705 + 't' signature scheme where 't' can be:
1.706 +
1.707 + - None: the alleged signature 'S' is directly applied the RSAVP1 signature
1.708 + primitive, as described in PKCS#1 v2.1, i.e. RFC 3447 Sect
1.709 + 5.2.1. Simply put, the provided signature is applied a moular
1.710 + exponentiation using the public key. Then, a comparison of the
1.711 + result is done against 'M'. On match, True is returned.
1.712 + Additionnal method parameters are just ignored.
1.713 +
1.714 + - 'pkcs': the alleged signature 'S' and message 'M' are applied
1.715 + RSASSA-PKCS1-v1_5-VERIFY signature verification scheme as
1.716 + described in Sect. 8.2.2 of RFC 3447. In that context,
1.717 + the hash function name is passed using 'h'. Possible values are
1.718 + "md2", "md4", "md5", "sha1", "tls", "sha224", "sha256", "sha384"
1.719 + and "sha512". If none is provided, sha1 is used. Other additionnal
1.720 + parameters are ignored.
1.721 +
1.722 + - 'pss': the alleged signature 'S' and message 'M' are applied
1.723 + RSASSA-PSS-VERIFY signature scheme as described in Sect. 8.1.2.
1.724 + of RFC 3447. In that context,
1.725 +
1.726 + o 'h' parameter provides the name of the hash method to use.
1.727 + Possible values are "md2", "md4", "md5", "sha1", "tls", "sha224",
1.728 + "sha256", "sha384" and "sha512". if none is provided, sha1
1.729 + is used.
1.730 +
1.731 + o 'mgf' is the mask generation function. By default, mgf
1.732 + is derived from the provided hash function using the
1.733 + generic MGF1 (see pkcs_mgf1() for details).
1.734 +
1.735 + o 'sLen' is the length in octet of the salt. You can overload the
1.736 + default value (the octet length of the hash value for provided
1.737 + algorithm) by providing another one with that parameter.
1.738 + """
1.739 + if t is None: # RSAVP1
1.740 + S = pkcs_os2ip(S)
1.741 + n = self.modulus
1.742 + if S > n-1:
1.743 + warning("Signature to be verified is too long for key modulus")
1.744 + return False
1.745 + m = self._rsavp1(S)
1.746 + if m is None:
1.747 + return False
1.748 + l = int(math.ceil(math.log(m, 2) / 8.)) # Hack
1.749 + m = pkcs_i2osp(m, l)
1.750 + return M == m
1.751 +
1.752 + elif t == "pkcs": # RSASSA-PKCS1-v1_5-VERIFY
1.753 + if h is None:
1.754 + h = "sha1"
1.755 + return self._rsassa_pkcs1_v1_5_verify(M, S, h)
1.756 +
1.757 + elif t == "pss": # RSASSA-PSS-VERIFY
1.758 + return self._rsassa_pss_verify(M, S, h, mgf, sLen)
1.759 +
1.760 + else:
1.761 + warning("Key.verify(): Unknown signature type (%s) provided" % t)
1.762 + return None
1.763 +
1.764 +class _DecryptAndSignMethods(OSSLHelper):
1.765 + ### Below are decryption related methods. Encryption ones are inherited
1.766 + ### from PubKey
1.767 +
1.768 + def _rsadp(self, c):
1.769 + """
1.770 + Internal method providing raw RSA decryption, i.e. simple modular
1.771 + exponentiation of the given ciphertext representative 'c', a long
1.772 + between 0 and n-1.
1.773 +
1.774 + This is the decryption primitive RSADP described in PKCS#1 v2.1,
1.775 + i.e. RFC 3447 Sect. 5.1.2.
1.776 +
1.777 + Input:
1.778 + c: ciphertest representative, a long between 0 and n-1, where
1.779 + n is the key modulus.
1.780 +
1.781 + Output:
1.782 + ciphertext representative, a long between 0 and n-1
1.783 +
1.784 + Not intended to be used directly. Please, see encrypt() method.
1.785 + """
1.786 +
1.787 + n = self.modulus
1.788 + if type(c) is int:
1.789 + c = long(c)
1.790 + if type(c) is not long or c > n-1:
1.791 + warning("Key._rsaep() expects a long between 0 and n-1")
1.792 + return None
1.793 +
1.794 + return self.key.decrypt(c)
1.795 +
1.796 +
1.797 + def _rsaes_pkcs1_v1_5_decrypt(self, C):
1.798 + """
1.799 + Implements RSAES-PKCS1-V1_5-DECRYPT() function described in section
1.800 + 7.2.2 of RFC 3447.
1.801 +
1.802 + Input:
1.803 + C: ciphertext to be decrypted, an octet string of length k, where
1.804 + k is the length in octets of the RSA modulus n.
1.805 +
1.806 + Output:
1.807 + an octet string of length k at most k - 11
1.808 +
1.809 + on error, None is returned.
1.810 + """
1.811 +
1.812 + # 1) Length checking
1.813 + cLen = len(C)
1.814 + k = self.modulusLen / 8
1.815 + if cLen != k or k < 11:
1.816 + warning("Key._rsaes_pkcs1_v1_5_decrypt() decryption error "
1.817 + "(cLen != k or k < 11)")
1.818 + return None
1.819 +
1.820 + # 2) RSA decryption
1.821 + c = pkcs_os2ip(C) # 2.a)
1.822 + m = self._rsadp(c) # 2.b)
1.823 + EM = pkcs_i2osp(m, k) # 2.c)
1.824 +
1.825 + # 3) EME-PKCS1-v1_5 decoding
1.826 +
1.827 + # I am aware of the note at the end of 7.2.2 regarding error
1.828 + # conditions reporting but the one provided below are for _local_
1.829 + # debugging purposes. --arno
1.830 +
1.831 + if EM[0] != '\x00':
1.832 + warning("Key._rsaes_pkcs1_v1_5_decrypt(): decryption error "
1.833 + "(first byte is not 0x00)")
1.834 + return None
1.835 +
1.836 + if EM[1] != '\x02':
1.837 + warning("Key._rsaes_pkcs1_v1_5_decrypt(): decryption error "
1.838 + "(second byte is not 0x02)")
1.839 + return None
1.840 +
1.841 + tmp = EM[2:].split('\x00', 1)
1.842 + if len(tmp) != 2:
1.843 + warning("Key._rsaes_pkcs1_v1_5_decrypt(): decryption error "
1.844 + "(no 0x00 to separate PS from M)")
1.845 + return None
1.846 +
1.847 + PS, M = tmp
1.848 + if len(PS) < 8:
1.849 + warning("Key._rsaes_pkcs1_v1_5_decrypt(): decryption error "
1.850 + "(PS is less than 8 byte long)")
1.851 + return None
1.852 +
1.853 + return M # 4)
1.854 +
1.855 +
1.856 + def _rsaes_oaep_decrypt(self, C, h=None, mgf=None, L=None):
1.857 + """
1.858 + Internal method providing RSAES-OAEP-DECRYPT as defined in Sect.
1.859 + 7.1.2 of RFC 3447. Not intended to be used directly. Please, see
1.860 + encrypt() method for type "OAEP".
1.861 +
1.862 +
1.863 + Input:
1.864 + C : ciphertext to be decrypted, an octet string of length k, where
1.865 + k = 2*hLen + 2 (k denotes the length in octets of the RSA modulus
1.866 + and hLen the length in octets of the hash function output)
1.867 + h : hash function name (in 'md2', 'md4', 'md5', 'sha1', 'tls',
1.868 + 'sha256', 'sha384'). 'sha1' is used if none is provided.
1.869 + mgf: the mask generation function f : seed, maskLen -> mask
1.870 + L : optional label whose association with the message is to be
1.871 + verified; the default value for L, if not provided is the empty
1.872 + string.
1.873 +
1.874 + Output:
1.875 + message, an octet string of length k mLen, where mLen <= k - 2*hLen - 2
1.876 +
1.877 + On error, None is returned.
1.878 + """
1.879 + # The steps below are the one described in Sect. 7.1.2 of RFC 3447.
1.880 +
1.881 + # 1) Length Checking
1.882 + # 1.a) is not done
1.883 + if h is None:
1.884 + h = "sha1"
1.885 + if not _hashFuncParams.has_key(h):
1.886 + warning("Key._rsaes_oaep_decrypt(): unknown hash function %s.", h)
1.887 + return None
1.888 + hLen = _hashFuncParams[h][0]
1.889 + hFun = _hashFuncParams[h][1]
1.890 + k = self.modulusLen / 8
1.891 + cLen = len(C)
1.892 + if cLen != k: # 1.b)
1.893 + warning("Key._rsaes_oaep_decrypt(): decryption error. "
1.894 + "(cLen != k)")
1.895 + return None
1.896 + if k < 2*hLen + 2:
1.897 + warning("Key._rsaes_oaep_decrypt(): decryption error. "
1.898 + "(k < 2*hLen + 2)")
1.899 + return None
1.900 +
1.901 + # 2) RSA decryption
1.902 + c = pkcs_os2ip(C) # 2.a)
1.903 + m = self._rsadp(c) # 2.b)
1.904 + EM = pkcs_i2osp(m, k) # 2.c)
1.905 +
1.906 + # 3) EME-OAEP decoding
1.907 + if L is None: # 3.a)
1.908 + L = ""
1.909 + lHash = hFun(L)
1.910 + Y = EM[:1] # 3.b)
1.911 + if Y != '\x00':
1.912 + warning("Key._rsaes_oaep_decrypt(): decryption error. "
1.913 + "(Y is not zero)")
1.914 + return None
1.915 + maskedSeed = EM[1:1+hLen]
1.916 + maskedDB = EM[1+hLen:]
1.917 + if mgf is None:
1.918 + mgf = lambda x,y: pkcs_mgf1(x, y, h)
1.919 + seedMask = mgf(maskedDB, hLen) # 3.c)
1.920 + seed = strxor(maskedSeed, seedMask) # 3.d)
1.921 + dbMask = mgf(seed, k - hLen - 1) # 3.e)
1.922 + DB = strxor(maskedDB, dbMask) # 3.f)
1.923 +
1.924 + # I am aware of the note at the end of 7.1.2 regarding error
1.925 + # conditions reporting but the one provided below are for _local_
1.926 + # debugging purposes. --arno
1.927 +
1.928 + lHashPrime = DB[:hLen] # 3.g)
1.929 + tmp = DB[hLen:].split('\x01', 1)
1.930 + if len(tmp) != 2:
1.931 + warning("Key._rsaes_oaep_decrypt(): decryption error. "
1.932 + "(0x01 separator not found)")
1.933 + return None
1.934 + PS, M = tmp
1.935 + if PS != '\x00'*len(PS):
1.936 + warning("Key._rsaes_oaep_decrypt(): decryption error. "
1.937 + "(invalid padding string)")
1.938 + return None
1.939 + if lHash != lHashPrime:
1.940 + warning("Key._rsaes_oaep_decrypt(): decryption error. "
1.941 + "(invalid hash)")
1.942 + return None
1.943 + return M # 4)
1.944 +
1.945 +
1.946 + def decrypt(self, C, t=None, h=None, mgf=None, L=None):
1.947 + """
1.948 + Decrypt ciphertext 'C' using 't' decryption scheme where 't' can be:
1.949 +
1.950 + - None: the ciphertext 'C' is directly applied the RSADP decryption
1.951 + primitive, as described in PKCS#1 v2.1, i.e. RFC 3447
1.952 + Sect 5.1.2. Simply, put the message undergo a modular
1.953 + exponentiation using the private key. Additionnal method
1.954 + parameters are just ignored.
1.955 +
1.956 + - 'pkcs': the ciphertext 'C' is applied RSAES-PKCS1-V1_5-DECRYPT
1.957 + decryption scheme as described in section 7.2.2 of RFC 3447.
1.958 + In that context, other parameters ('h', 'mgf', 'l') are not
1.959 + used.
1.960 +
1.961 + - 'oaep': the ciphertext 'C' is applied the RSAES-OAEP-DECRYPT decryption
1.962 + scheme, as described in PKCS#1 v2.1, i.e. RFC 3447 Sect
1.963 + 7.1.2. In that context,
1.964 +
1.965 + o 'h' parameter provides the name of the hash method to use.
1.966 + Possible values are "md2", "md4", "md5", "sha1", "tls",
1.967 + "sha224", "sha256", "sha384" and "sha512". if none is provided,
1.968 + sha1 is used by default.
1.969 +
1.970 + o 'mgf' is the mask generation function. By default, mgf
1.971 + is derived from the provided hash function using the
1.972 + generic MGF1 (see pkcs_mgf1() for details).
1.973 +
1.974 + o 'L' is the optional label to be associated with the
1.975 + message. If not provided, the default value is used, i.e
1.976 + the empty string. No check is done on the input limitation
1.977 + of the hash function regarding the size of 'L' (for
1.978 + instance, 2^61 - 1 for SHA-1). You have been warned.
1.979 + """
1.980 + if t is None:
1.981 + C = pkcs_os2ip(C)
1.982 + c = self._rsadp(C)
1.983 + l = int(math.ceil(math.log(c, 2) / 8.)) # Hack
1.984 + return pkcs_i2osp(c, l)
1.985 +
1.986 + elif t == "pkcs":
1.987 + return self._rsaes_pkcs1_v1_5_decrypt(C)
1.988 +
1.989 + elif t == "oaep":
1.990 + return self._rsaes_oaep_decrypt(C, h, mgf, L)
1.991 +
1.992 + else:
1.993 + warning("Key.decrypt(): Unknown decryption type (%s) provided" % t)
1.994 + return None
1.995 +
1.996 + ### Below are signature related methods. Verification ones are inherited from
1.997 + ### PubKey
1.998 +
1.999 + def _rsasp1(self, m):
1.1000 + """
1.1001 + Internal method providing raw RSA signature, i.e. simple modular
1.1002 + exponentiation of the given message representative 'm', an integer
1.1003 + between 0 and n-1.
1.1004 +
1.1005 + This is the signature primitive RSASP1 described in PKCS#1 v2.1,
1.1006 + i.e. RFC 3447 Sect. 5.2.1.
1.1007 +
1.1008 + Input:
1.1009 + m: message representative, an integer between 0 and n-1, where
1.1010 + n is the key modulus.
1.1011 +
1.1012 + Output:
1.1013 + signature representative, an integer between 0 and n-1
1.1014 +
1.1015 + Not intended to be used directly. Please, see sign() method.
1.1016 + """
1.1017 + return self._rsadp(m)
1.1018 +
1.1019 +
1.1020 + def _rsassa_pss_sign(self, M, h=None, mgf=None, sLen=None):
1.1021 + """
1.1022 + Implements RSASSA-PSS-SIGN() function described in Sect. 8.1.1 of
1.1023 + RFC 3447.
1.1024 +
1.1025 + Input:
1.1026 + M: message to be signed, an octet string
1.1027 +
1.1028 + Output:
1.1029 + signature, an octet string of length k, where k is the length in
1.1030 + octets of the RSA modulus n.
1.1031 +
1.1032 + On error, None is returned.
1.1033 + """
1.1034 +
1.1035 + # Set default parameters if not provided
1.1036 + if h is None: # By default, sha1
1.1037 + h = "sha1"
1.1038 + if not _hashFuncParams.has_key(h):
1.1039 + warning("Key._rsassa_pss_sign(): unknown hash function "
1.1040 + "provided (%s)" % h)
1.1041 + return None
1.1042 + if mgf is None: # use mgf1 with underlying hash function
1.1043 + mgf = lambda x,y: pkcs_mgf1(x, y, h)
1.1044 + if sLen is None: # use Hash output length (A.2.3 of RFC 3447)
1.1045 + hLen = _hashFuncParams[h][0]
1.1046 + sLen = hLen
1.1047 +
1.1048 + # 1) EMSA-PSS encoding
1.1049 + modBits = self.modulusLen
1.1050 + k = modBits / 8
1.1051 + EM = pkcs_emsa_pss_encode(M, modBits - 1, h, mgf, sLen)
1.1052 + if EM is None:
1.1053 + warning("Key._rsassa_pss_sign(): unable to encode")
1.1054 + return None
1.1055 +
1.1056 + # 2) RSA signature
1.1057 + m = pkcs_os2ip(EM) # 2.a)
1.1058 + s = self._rsasp1(m) # 2.b)
1.1059 + S = pkcs_i2osp(s, k) # 2.c)
1.1060 +
1.1061 + return S # 3)
1.1062 +
1.1063 +
1.1064 + def _rsassa_pkcs1_v1_5_sign(self, M, h):
1.1065 + """
1.1066 + Implements RSASSA-PKCS1-v1_5-SIGN() function as described in
1.1067 + Sect. 8.2.1 of RFC 3447.
1.1068 +
1.1069 + Input:
1.1070 + M: message to be signed, an octet string
1.1071 + h: hash function name (in 'md2', 'md4', 'md5', 'sha1', 'tls'
1.1072 + 'sha256', 'sha384').
1.1073 +
1.1074 + Output:
1.1075 + the signature, an octet string.
1.1076 + """
1.1077 +
1.1078 + # 1) EMSA-PKCS1-v1_5 encoding
1.1079 + k = self.modulusLen / 8
1.1080 + EM = pkcs_emsa_pkcs1_v1_5_encode(M, k, h)
1.1081 + if EM is None:
1.1082 + warning("Key._rsassa_pkcs1_v1_5_sign(): unable to encode")
1.1083 + return None
1.1084 +
1.1085 + # 2) RSA signature
1.1086 + m = pkcs_os2ip(EM) # 2.a)
1.1087 + s = self._rsasp1(m) # 2.b)
1.1088 + S = pkcs_i2osp(s, k) # 2.c)
1.1089 +
1.1090 + return S # 3)
1.1091 +
1.1092 +
1.1093 + def sign(self, M, t=None, h=None, mgf=None, sLen=None):
1.1094 + """
1.1095 + Sign message 'M' using 't' signature scheme where 't' can be:
1.1096 +
1.1097 + - None: the message 'M' is directly applied the RSASP1 signature
1.1098 + primitive, as described in PKCS#1 v2.1, i.e. RFC 3447 Sect
1.1099 + 5.2.1. Simply put, the message undergo a modular exponentiation
1.1100 + using the private key. Additionnal method parameters are just
1.1101 + ignored.
1.1102 +
1.1103 + - 'pkcs': the message 'M' is applied RSASSA-PKCS1-v1_5-SIGN signature
1.1104 + scheme as described in Sect. 8.2.1 of RFC 3447. In that context,
1.1105 + the hash function name is passed using 'h'. Possible values are
1.1106 + "md2", "md4", "md5", "sha1", "tls", "sha224", "sha256", "sha384"
1.1107 + and "sha512". If none is provided, sha1 is used. Other additionnal
1.1108 + parameters are ignored.
1.1109 +
1.1110 + - 'pss' : the message 'M' is applied RSASSA-PSS-SIGN signature scheme as
1.1111 + described in Sect. 8.1.1. of RFC 3447. In that context,
1.1112 +
1.1113 + o 'h' parameter provides the name of the hash method to use.
1.1114 + Possible values are "md2", "md4", "md5", "sha1", "tls", "sha224",
1.1115 + "sha256", "sha384" and "sha512". if none is provided, sha1
1.1116 + is used.
1.1117 +
1.1118 + o 'mgf' is the mask generation function. By default, mgf
1.1119 + is derived from the provided hash function using the
1.1120 + generic MGF1 (see pkcs_mgf1() for details).
1.1121 +
1.1122 + o 'sLen' is the length in octet of the salt. You can overload the
1.1123 + default value (the octet length of the hash value for provided
1.1124 + algorithm) by providing another one with that parameter.
1.1125 + """
1.1126 +
1.1127 + if t is None: # RSASP1
1.1128 + M = pkcs_os2ip(M)
1.1129 + n = self.modulus
1.1130 + if M > n-1:
1.1131 + warning("Message to be signed is too long for key modulus")
1.1132 + return None
1.1133 + s = self._rsasp1(M)
1.1134 + if s is None:
1.1135 + return None
1.1136 + return pkcs_i2osp(s, self.modulusLen/8)
1.1137 +
1.1138 + elif t == "pkcs": # RSASSA-PKCS1-v1_5-SIGN
1.1139 + if h is None:
1.1140 + h = "sha1"
1.1141 + return self._rsassa_pkcs1_v1_5_sign(M, h)
1.1142 +
1.1143 + elif t == "pss": # RSASSA-PSS-SIGN
1.1144 + return self._rsassa_pss_sign(M, h, mgf, sLen)
1.1145 +
1.1146 + else:
1.1147 + warning("Key.sign(): Unknown signature type (%s) provided" % t)
1.1148 + return None
1.1149 +
1.1150 +
1.1151 +
1.1152 +
1.1153 +class PubKey(OSSLHelper, _EncryptAndVerify):
1.1154 + # Below are the fields we recognize in the -text output of openssl
1.1155 + # and from which we extract information. We expect them in that
1.1156 + # order. Number of spaces does matter.
1.1157 + possible_fields = [ "Modulus (",
1.1158 + "Exponent:" ]
1.1159 + possible_fields_count = len(possible_fields)
1.1160 +
1.1161 + def __init__(self, keypath):
1.1162 + error_msg = "Unable to import key."
1.1163 +
1.1164 + # XXX Temporary hack to use PubKey inside Cert
1.1165 + if type(keypath) is tuple:
1.1166 + e, m, mLen = keypath
1.1167 + self.modulus = m
1.1168 + self.modulusLen = mLen
1.1169 + self.pubExp = e
1.1170 + return
1.1171 +
1.1172 + fields_dict = {}
1.1173 + for k in self.possible_fields:
1.1174 + fields_dict[k] = None
1.1175 +
1.1176 + self.keypath = None
1.1177 + rawkey = None
1.1178 +
1.1179 + if (not '\x00' in keypath) and os.path.isfile(keypath): # file
1.1180 + self.keypath = keypath
1.1181 + key_size = os.path.getsize(keypath)
1.1182 + if key_size > MAX_KEY_SIZE:
1.1183 + raise Exception(error_msg)
1.1184 + try:
1.1185 + f = open(keypath)
1.1186 + rawkey = f.read()
1.1187 + f.close()
1.1188 + except:
1.1189 + raise Exception(error_msg)
1.1190 + else:
1.1191 + rawkey = keypath
1.1192 +
1.1193 + if rawkey is None:
1.1194 + raise Exception(error_msg)
1.1195 +
1.1196 + self.rawkey = rawkey
1.1197 +
1.1198 + # Let's try to get file format : PEM or DER.
1.1199 + fmtstr = 'openssl rsa -text -pubin -inform %s -noout '
1.1200 + convertstr = 'openssl rsa -pubin -inform %s -outform %s 2>/dev/null'
1.1201 + key_header = "-----BEGIN PUBLIC KEY-----"
1.1202 + key_footer = "-----END PUBLIC KEY-----"
1.1203 + l = rawkey.split(key_header, 1)
1.1204 + if len(l) == 2: # looks like PEM
1.1205 + tmp = l[1]
1.1206 + l = tmp.split(key_footer, 1)
1.1207 + if len(l) == 2:
1.1208 + tmp = l[0]
1.1209 + rawkey = "%s%s%s\n" % (key_header, tmp, key_footer)
1.1210 + else:
1.1211 + raise Exception(error_msg)
1.1212 + r,w,e = popen2.popen3(fmtstr % "PEM")
1.1213 + w.write(rawkey)
1.1214 + w.close()
1.1215 + textkey = r.read()
1.1216 + r.close()
1.1217 + res = e.read()
1.1218 + e.close()
1.1219 + if res == '':
1.1220 + self.format = "PEM"
1.1221 + self.pemkey = rawkey
1.1222 + self.textkey = textkey
1.1223 + cmd = convertstr % ("PEM", "DER")
1.1224 + self.derkey = self._apply_ossl_cmd(cmd, rawkey)
1.1225 + else:
1.1226 + raise Exception(error_msg)
1.1227 + else: # not PEM, try DER
1.1228 + r,w,e = popen2.popen3(fmtstr % "DER")
1.1229 + w.write(rawkey)
1.1230 + w.close()
1.1231 + textkey = r.read()
1.1232 + r.close()
1.1233 + res = e.read()
1.1234 + if res == '':
1.1235 + self.format = "DER"
1.1236 + self.derkey = rawkey
1.1237 + self.textkey = textkey
1.1238 + cmd = convertstr % ("DER", "PEM")
1.1239 + self.pemkey = self._apply_ossl_cmd(cmd, rawkey)
1.1240 + cmd = convertstr % ("DER", "DER")
1.1241 + self.derkey = self._apply_ossl_cmd(cmd, rawkey)
1.1242 + else:
1.1243 + try: # Perhaps it is a cert
1.1244 + c = Cert(keypath)
1.1245 + except:
1.1246 + raise Exception(error_msg)
1.1247 + # TODO:
1.1248 + # Reconstruct a key (der and pem) and provide:
1.1249 + # self.format
1.1250 + # self.derkey
1.1251 + # self.pemkey
1.1252 + # self.textkey
1.1253 + # self.keypath
1.1254 +
1.1255 + self.osslcmdbase = 'openssl rsa -pubin -inform %s ' % self.format
1.1256 +
1.1257 + self.keypath = keypath
1.1258 +
1.1259 + # Parse the -text output of openssl to make things available
1.1260 + l = self.textkey.split('\n', 1)
1.1261 + if len(l) != 2:
1.1262 + raise Exception(error_msg)
1.1263 + cur, tmp = l
1.1264 + i = 0
1.1265 + k = self.possible_fields[i] # Modulus (
1.1266 + cur = cur[len(k):] + '\n'
1.1267 + while k:
1.1268 + l = tmp.split('\n', 1)
1.1269 + if len(l) != 2: # Over
1.1270 + fields_dict[k] = cur
1.1271 + break
1.1272 + l, tmp = l
1.1273 +
1.1274 + newkey = 0
1.1275 + # skip fields we have already seen, this is the purpose of 'i'
1.1276 + for j in range(i, self.possible_fields_count):
1.1277 + f = self.possible_fields[j]
1.1278 + if l.startswith(f):
1.1279 + fields_dict[k] = cur
1.1280 + cur = l[len(f):] + '\n'
1.1281 + k = f
1.1282 + newkey = 1
1.1283 + i = j+1
1.1284 + break
1.1285 + if newkey == 1:
1.1286 + continue
1.1287 + cur += l + '\n'
1.1288 +
1.1289 + # modulus and modulus length
1.1290 + v = fields_dict["Modulus ("]
1.1291 + self.modulusLen = None
1.1292 + if v:
1.1293 + v, rem = v.split(' bit):', 1)
1.1294 + self.modulusLen = int(v)
1.1295 + rem = rem.replace('\n','').replace(' ','').replace(':','')
1.1296 + self.modulus = long(rem, 16)
1.1297 + if self.modulus is None:
1.1298 + raise Exception(error_msg)
1.1299 +
1.1300 + # public exponent
1.1301 + v = fields_dict["Exponent:"]
1.1302 + self.pubExp = None
1.1303 + if v:
1.1304 + self.pubExp = long(v.split('(', 1)[0])
1.1305 + if self.pubExp is None:
1.1306 + raise Exception(error_msg)
1.1307 +
1.1308 + self.key = RSA.construct((self.modulus, self.pubExp, ))
1.1309 +
1.1310 + def __str__(self):
1.1311 + return self.derkey
1.1312 +
1.1313 +
1.1314 +class Key(OSSLHelper, _DecryptAndSignMethods, _EncryptAndVerify):
1.1315 + # Below are the fields we recognize in the -text output of openssl
1.1316 + # and from which we extract information. We expect them in that
1.1317 + # order. Number of spaces does matter.
1.1318 + possible_fields = [ "Private-Key: (",
1.1319 + "modulus:",
1.1320 + "publicExponent:",
1.1321 + "privateExponent:",
1.1322 + "prime1:",
1.1323 + "prime2:",
1.1324 + "exponent1:",
1.1325 + "exponent2:",
1.1326 + "coefficient:" ]
1.1327 + possible_fields_count = len(possible_fields)
1.1328 +
1.1329 + def __init__(self, keypath):
1.1330 + error_msg = "Unable to import key."
1.1331 +
1.1332 + fields_dict = {}
1.1333 + for k in self.possible_fields:
1.1334 + fields_dict[k] = None
1.1335 +
1.1336 + self.keypath = None
1.1337 + rawkey = None
1.1338 +
1.1339 + if (not '\x00' in keypath) and os.path.isfile(keypath):
1.1340 + self.keypath = keypath
1.1341 + key_size = os.path.getsize(keypath)
1.1342 + if key_size > MAX_KEY_SIZE:
1.1343 + raise Exception(error_msg)
1.1344 + try:
1.1345 + f = open(keypath)
1.1346 + rawkey = f.read()
1.1347 + f.close()
1.1348 + except:
1.1349 + raise Exception(error_msg)
1.1350 + else:
1.1351 + rawkey = keypath
1.1352 +
1.1353 + if rawkey is None:
1.1354 + raise Exception(error_msg)
1.1355 +
1.1356 + self.rawkey = rawkey
1.1357 +
1.1358 + # Let's try to get file format : PEM or DER.
1.1359 + fmtstr = 'openssl rsa -text -inform %s -noout '
1.1360 + convertstr = 'openssl rsa -inform %s -outform %s 2>/dev/null'
1.1361 + key_header = "-----BEGIN RSA PRIVATE KEY-----"
1.1362 + key_footer = "-----END RSA PRIVATE KEY-----"
1.1363 + l = rawkey.split(key_header, 1)
1.1364 + if len(l) == 2: # looks like PEM
1.1365 + tmp = l[1]
1.1366 + l = tmp.split(key_footer, 1)
1.1367 + if len(l) == 2:
1.1368 + tmp = l[0]
1.1369 + rawkey = "%s%s%s\n" % (key_header, tmp, key_footer)
1.1370 + else:
1.1371 + raise Exception(error_msg)
1.1372 + r,w,e = popen2.popen3(fmtstr % "PEM")
1.1373 + w.write(rawkey)
1.1374 + w.close()
1.1375 + textkey = r.read()
1.1376 + r.close()
1.1377 + res = e.read()
1.1378 + e.close()
1.1379 + if res == '':
1.1380 + self.format = "PEM"
1.1381 + self.pemkey = rawkey
1.1382 + self.textkey = textkey
1.1383 + cmd = convertstr % ("PEM", "DER")
1.1384 + self.derkey = self._apply_ossl_cmd(cmd, rawkey)
1.1385 + else:
1.1386 + raise Exception(error_msg)
1.1387 + else: # not PEM, try DER
1.1388 + r,w,e = popen2.popen3(fmtstr % "DER")
1.1389 + w.write(rawkey)
1.1390 + w.close()
1.1391 + textkey = r.read()
1.1392 + r.close()
1.1393 + res = e.read()
1.1394 + if res == '':
1.1395 + self.format = "DER"
1.1396 + self.derkey = rawkey
1.1397 + self.textkey = textkey
1.1398 + cmd = convertstr % ("DER", "PEM")
1.1399 + self.pemkey = self._apply_ossl_cmd(cmd, rawkey)
1.1400 + cmd = convertstr % ("DER", "DER")
1.1401 + self.derkey = self._apply_ossl_cmd(cmd, rawkey)
1.1402 + else:
1.1403 + raise Exception(error_msg)
1.1404 +
1.1405 + self.osslcmdbase = 'openssl rsa -inform %s ' % self.format
1.1406 +
1.1407 + r,w,e = popen2.popen3('openssl asn1parse -inform DER ')
1.1408 + w.write(self.derkey)
1.1409 + w.close()
1.1410 + self.asn1parsekey = r.read()
1.1411 + r.close()
1.1412 + res = e.read()
1.1413 + e.close()
1.1414 + if res != '':
1.1415 + raise Exception(error_msg)
1.1416 +
1.1417 + self.keypath = keypath
1.1418 +
1.1419 + # Parse the -text output of openssl to make things available
1.1420 + l = self.textkey.split('\n', 1)
1.1421 + if len(l) != 2:
1.1422 + raise Exception(error_msg)
1.1423 + cur, tmp = l
1.1424 + i = 0
1.1425 + k = self.possible_fields[i] # Private-Key: (
1.1426 + cur = cur[len(k):] + '\n'
1.1427 + while k:
1.1428 + l = tmp.split('\n', 1)
1.1429 + if len(l) != 2: # Over
1.1430 + fields_dict[k] = cur
1.1431 + break
1.1432 + l, tmp = l
1.1433 +
1.1434 + newkey = 0
1.1435 + # skip fields we have already seen, this is the purpose of 'i'
1.1436 + for j in range(i, self.possible_fields_count):
1.1437 + f = self.possible_fields[j]
1.1438 + if l.startswith(f):
1.1439 + fields_dict[k] = cur
1.1440 + cur = l[len(f):] + '\n'
1.1441 + k = f
1.1442 + newkey = 1
1.1443 + i = j+1
1.1444 + break
1.1445 + if newkey == 1:
1.1446 + continue
1.1447 + cur += l + '\n'
1.1448 +
1.1449 + # modulus length
1.1450 + v = fields_dict["Private-Key: ("]
1.1451 + self.modulusLen = None
1.1452 + if v:
1.1453 + self.modulusLen = int(v.split(' bit', 1)[0])
1.1454 + if self.modulusLen is None:
1.1455 + raise Exception(error_msg)
1.1456 +
1.1457 + # public exponent
1.1458 + v = fields_dict["publicExponent:"]
1.1459 + self.pubExp = None
1.1460 + if v:
1.1461 + self.pubExp = long(v.split('(', 1)[0])
1.1462 + if self.pubExp is None:
1.1463 + raise Exception(error_msg)
1.1464 +
1.1465 + tmp = {}
1.1466 + for k in ["modulus:", "privateExponent:", "prime1:", "prime2:",
1.1467 + "exponent1:", "exponent2:", "coefficient:"]:
1.1468 + v = fields_dict[k]
1.1469 + if v:
1.1470 + s = v.replace('\n', '').replace(' ', '').replace(':', '')
1.1471 + tmp[k] = long(s, 16)
1.1472 + else:
1.1473 + raise Exception(error_msg)
1.1474 +
1.1475 + self.modulus = tmp["modulus:"]
1.1476 + self.privExp = tmp["privateExponent:"]
1.1477 + self.prime1 = tmp["prime1:"]
1.1478 + self.prime2 = tmp["prime2:"]
1.1479 + self.exponent1 = tmp["exponent1:"]
1.1480 + self.exponent2 = tmp["exponent2:"]
1.1481 + self.coefficient = tmp["coefficient:"]
1.1482 +
1.1483 + self.key = RSA.construct((self.modulus, self.pubExp, self.privExp))
1.1484 +
1.1485 + def __str__(self):
1.1486 + return self.derkey
1.1487 +
1.1488 +
1.1489 +# We inherit from PubKey to get access to all encryption and verification
1.1490 +# methods. To have that working, we simply need Cert to provide
1.1491 +# modulusLen and key attribute.
1.1492 +# XXX Yes, it is a hack.
1.1493 +class Cert(OSSLHelper, _EncryptAndVerify):
1.1494 + # Below are the fields we recognize in the -text output of openssl
1.1495 + # and from which we extract information. We expect them in that
1.1496 + # order. Number of spaces does matter.
1.1497 + possible_fields = [ " Version:",
1.1498 + " Serial Number:",
1.1499 + " Signature Algorithm:",
1.1500 + " Issuer:",
1.1501 + " Not Before:",
1.1502 + " Not After :",
1.1503 + " Subject:",
1.1504 + " Public Key Algorithm:",
1.1505 + " Modulus (",
1.1506 + " Exponent:",
1.1507 + " X509v3 Subject Key Identifier:",
1.1508 + " X509v3 Authority Key Identifier:",
1.1509 + " keyid:",
1.1510 + " DirName:",
1.1511 + " serial:",
1.1512 + " X509v3 Basic Constraints:",
1.1513 + " X509v3 Key Usage:",
1.1514 + " X509v3 Extended Key Usage:",
1.1515 + " X509v3 CRL Distribution Points:",
1.1516 + " Authority Information Access:",
1.1517 + " Signature Algorithm:" ]
1.1518 + possible_fields_count = len(possible_fields)
1.1519 +
1.1520 + def __init__(self, certpath):
1.1521 + error_msg = "Unable to import certificate."
1.1522 +
1.1523 + fields_dict = {}
1.1524 + for k in self.possible_fields:
1.1525 + fields_dict[k] = None
1.1526 +
1.1527 + self.certpath = None
1.1528 + rawcert = None
1.1529 +
1.1530 + if (not '\x00' in certpath) and os.path.isfile(certpath): # file
1.1531 + self.certpath = certpath
1.1532 + cert_size = os.path.getsize(certpath)
1.1533 + if cert_size > MAX_CERT_SIZE:
1.1534 + raise Exception(error_msg)
1.1535 + try:
1.1536 + f = open(certpath)
1.1537 + rawcert = f.read()
1.1538 + f.close()
1.1539 + except:
1.1540 + raise Exception(error_msg)
1.1541 + else:
1.1542 + rawcert = certpath
1.1543 +
1.1544 + if rawcert is None:
1.1545 + raise Exception(error_msg)
1.1546 +
1.1547 + self.rawcert = rawcert
1.1548 +
1.1549 + # Let's try to get file format : PEM or DER.
1.1550 + fmtstr = 'openssl x509 -text -inform %s -noout '
1.1551 + convertstr = 'openssl x509 -inform %s -outform %s '
1.1552 + cert_header = "-----BEGIN CERTIFICATE-----"
1.1553 + cert_footer = "-----END CERTIFICATE-----"
1.1554 + l = rawcert.split(cert_header, 1)
1.1555 + if len(l) == 2: # looks like PEM
1.1556 + tmp = l[1]
1.1557 + l = tmp.split(cert_footer, 1)
1.1558 + if len(l) == 2:
1.1559 + tmp = l[0]
1.1560 + rawcert = "%s%s%s\n" % (cert_header, tmp, cert_footer)
1.1561 + else:
1.1562 + raise Exception(error_msg)
1.1563 + r,w,e = popen2.popen3(fmtstr % "PEM")
1.1564 + w.write(rawcert)
1.1565 + w.close()
1.1566 + textcert = r.read()
1.1567 + r.close()
1.1568 + res = e.read()
1.1569 + e.close()
1.1570 + if res == '':
1.1571 + self.format = "PEM"
1.1572 + self.pemcert = rawcert
1.1573 + self.textcert = textcert
1.1574 + cmd = convertstr % ("PEM", "DER")
1.1575 + self.dercert = self._apply_ossl_cmd(cmd, rawcert)
1.1576 + else:
1.1577 + raise Exception(error_msg)
1.1578 + else: # not PEM, try DER
1.1579 + r,w,e = popen2.popen3(fmtstr % "DER")
1.1580 + w.write(rawcert)
1.1581 + w.close()
1.1582 + textcert = r.read()
1.1583 + r.close()
1.1584 + res = e.read()
1.1585 + if res == '':
1.1586 + self.format = "DER"
1.1587 + self.dercert = rawcert
1.1588 + self.textcert = textcert
1.1589 + cmd = convertstr % ("DER", "PEM")
1.1590 + self.pemcert = self._apply_ossl_cmd(cmd, rawcert)
1.1591 + cmd = convertstr % ("DER", "DER")
1.1592 + self.dercert = self._apply_ossl_cmd(cmd, rawcert)
1.1593 + else:
1.1594 + raise Exception(error_msg)
1.1595 +
1.1596 + self.osslcmdbase = 'openssl x509 -inform %s ' % self.format
1.1597 +
1.1598 + r,w,e = popen2.popen3('openssl asn1parse -inform DER ')
1.1599 + w.write(self.dercert)
1.1600 + w.close()
1.1601 + self.asn1parsecert = r.read()
1.1602 + r.close()
1.1603 + res = e.read()
1.1604 + e.close()
1.1605 + if res != '':
1.1606 + raise Exception(error_msg)
1.1607 +
1.1608 + # Grab _raw_ X509v3 Authority Key Identifier, if any.
1.1609 + tmp = self.asn1parsecert.split(":X509v3 Authority Key Identifier", 1)
1.1610 + self.authorityKeyID = None
1.1611 + if len(tmp) == 2:
1.1612 + tmp = tmp[1]
1.1613 + tmp = tmp.split("[HEX DUMP]:", 1)[1]
1.1614 + self.authorityKeyID=tmp.split('\n',1)[0]
1.1615 +
1.1616 + # Grab _raw_ X509v3 Subject Key Identifier, if any.
1.1617 + tmp = self.asn1parsecert.split(":X509v3 Subject Key Identifier", 1)
1.1618 + self.subjectKeyID = None
1.1619 + if len(tmp) == 2:
1.1620 + tmp = tmp[1]
1.1621 + tmp = tmp.split("[HEX DUMP]:", 1)[1]
1.1622 + self.subjectKeyID=tmp.split('\n',1)[0]
1.1623 +
1.1624 + # Get tbsCertificate using the worst hack. output of asn1parse
1.1625 + # looks like that:
1.1626 + #
1.1627 + # 0:d=0 hl=4 l=1298 cons: SEQUENCE
1.1628 + # 4:d=1 hl=4 l=1018 cons: SEQUENCE
1.1629 + # ...
1.1630 + #
1.1631 + l1,l2 = self.asn1parsecert.split('\n', 2)[:2]
1.1632 + hl1 = int(l1.split("hl=",1)[1].split("l=",1)[0])
1.1633 + rem = l2.split("hl=",1)[1]
1.1634 + hl2, rem = rem.split("l=",1)
1.1635 + hl2 = int(hl2)
1.1636 + l = int(rem.split("cons",1)[0])
1.1637 + self.tbsCertificate = self.dercert[hl1:hl1+hl2+l]
1.1638 +
1.1639 + # Parse the -text output of openssl to make things available
1.1640 + tmp = self.textcert.split('\n', 2)[2]
1.1641 + l = tmp.split('\n', 1)
1.1642 + if len(l) != 2:
1.1643 + raise Exception(error_msg)
1.1644 + cur, tmp = l
1.1645 + i = 0
1.1646 + k = self.possible_fields[i] # Version:
1.1647 + cur = cur[len(k):] + '\n'
1.1648 + while k:
1.1649 + l = tmp.split('\n', 1)
1.1650 + if len(l) != 2: # Over
1.1651 + fields_dict[k] = cur
1.1652 + break
1.1653 + l, tmp = l
1.1654 +
1.1655 + newkey = 0
1.1656 + # skip fields we have already seen, this is the purpose of 'i'
1.1657 + for j in range(i, self.possible_fields_count):
1.1658 + f = self.possible_fields[j]
1.1659 + if l.startswith(f):
1.1660 + fields_dict[k] = cur
1.1661 + cur = l[len(f):] + '\n'
1.1662 + k = f
1.1663 + newkey = 1
1.1664 + i = j+1
1.1665 + break
1.1666 + if newkey == 1:
1.1667 + continue
1.1668 + cur += l + '\n'
1.1669 +
1.1670 + # version
1.1671 + v = fields_dict[" Version:"]
1.1672 + self.version = None
1.1673 + if v:
1.1674 + self.version = int(v[1:2])
1.1675 + if self.version is None:
1.1676 + raise Exception(error_msg)
1.1677 +
1.1678 + # serial number
1.1679 + v = fields_dict[" Serial Number:"]
1.1680 + self.serial = None
1.1681 + if v:
1.1682 + v = v.replace('\n', '').strip()
1.1683 + if "0x" in v:
1.1684 + v = v.split("0x", 1)[1].split(')', 1)[0]
1.1685 + v = v.replace(':', '').upper()
1.1686 + if len(v) % 2:
1.1687 + v = '0' + v
1.1688 + self.serial = v
1.1689 + if self.serial is None:
1.1690 + raise Exception(error_msg)
1.1691 +
1.1692 + # Signature Algorithm
1.1693 + v = fields_dict[" Signature Algorithm:"]
1.1694 + self.sigAlg = None
1.1695 + if v:
1.1696 + v = v.split('\n',1)[0]
1.1697 + v = v.strip()
1.1698 + self.sigAlg = v
1.1699 + if self.sigAlg is None:
1.1700 + raise Exception(error_msg)
1.1701 +
1.1702 + # issuer
1.1703 + v = fields_dict[" Issuer:"]
1.1704 + self.issuer = None
1.1705 + if v:
1.1706 + v = v.split('\n',1)[0]
1.1707 + v = v.strip()
1.1708 + self.issuer = v
1.1709 + if self.issuer is None:
1.1710 + raise Exception(error_msg)
1.1711 +
1.1712 + # not before
1.1713 + v = fields_dict[" Not Before:"]
1.1714 + self.notBefore_str = None
1.1715 + if v:
1.1716 + v = v.split('\n',1)[0]
1.1717 + v = v.strip()
1.1718 + self.notBefore_str = v
1.1719 + if self.notBefore_str is None:
1.1720 + raise Exception(error_msg)
1.1721 + self.notBefore = time.strptime(self.notBefore_str,
1.1722 + "%b %d %H:%M:%S %Y %Z")
1.1723 + self.notBefore_str_simple = time.strftime("%x", self.notBefore)
1.1724 +
1.1725 + # not after
1.1726 + v = fields_dict[" Not After :"]
1.1727 + self.notAfter_str = None
1.1728 + if v:
1.1729 + v = v.split('\n',1)[0]
1.1730 + v = v.strip()
1.1731 + self.notAfter_str = v
1.1732 + if self.notAfter_str is None:
1.1733 + raise Exception(error_msg)
1.1734 + self.notAfter = time.strptime(self.notAfter_str,
1.1735 + "%b %d %H:%M:%S %Y %Z")
1.1736 + self.notAfter_str_simple = time.strftime("%x", self.notAfter)
1.1737 +
1.1738 + # subject
1.1739 + v = fields_dict[" Subject:"]
1.1740 + self.subject = None
1.1741 + if v:
1.1742 + v = v.split('\n',1)[0]
1.1743 + v = v.strip()
1.1744 + self.subject = v
1.1745 + if self.subject is None:
1.1746 + raise Exception(error_msg)
1.1747 +
1.1748 + # Public Key Algorithm
1.1749 + v = fields_dict[" Public Key Algorithm:"]
1.1750 + self.pubKeyAlg = None
1.1751 + if v:
1.1752 + v = v.split('\n',1)[0]
1.1753 + v = v.strip()
1.1754 + self.pubKeyAlg = v
1.1755 + if self.pubKeyAlg is None:
1.1756 + raise Exception(error_msg)
1.1757 +
1.1758 + # Modulus
1.1759 + v = fields_dict[" Modulus ("]
1.1760 + self.modulus = None
1.1761 + if v:
1.1762 + v,t = v.split(' bit):',1)
1.1763 + self.modulusLen = int(v)
1.1764 + t = t.replace(' ', '').replace('\n', ''). replace(':', '')
1.1765 + self.modulus_hexdump = t
1.1766 + self.modulus = long(t, 16)
1.1767 + if self.modulus is None:
1.1768 + raise Exception(error_msg)
1.1769 +
1.1770 + # Exponent
1.1771 + v = fields_dict[" Exponent:"]
1.1772 + self.exponent = None
1.1773 + if v:
1.1774 + v = v.split('(',1)[0]
1.1775 + self.exponent = long(v)
1.1776 + if self.exponent is None:
1.1777 + raise Exception(error_msg)
1.1778 +
1.1779 + # Public Key instance
1.1780 + self.key = RSA.construct((self.modulus, self.exponent, ))
1.1781 +
1.1782 + # Subject Key Identifier
1.1783 +
1.1784 + # Authority Key Identifier: keyid, dirname and serial
1.1785 + self.authorityKeyID_keyid = None
1.1786 + self.authorityKeyID_dirname = None
1.1787 + self.authorityKeyID_serial = None
1.1788 + if self.authorityKeyID: # (hex version already done using asn1parse)
1.1789 + v = fields_dict[" keyid:"]
1.1790 + if v:
1.1791 + v = v.split('\n',1)[0]
1.1792 + v = v.strip().replace(':', '')
1.1793 + self.authorityKeyID_keyid = v
1.1794 + v = fields_dict[" DirName:"]
1.1795 + if v:
1.1796 + v = v.split('\n',1)[0]
1.1797 + self.authorityKeyID_dirname = v
1.1798 + v = fields_dict[" serial:"]
1.1799 + if v:
1.1800 + v = v.split('\n',1)[0]
1.1801 + v = v.strip().replace(':', '')
1.1802 + self.authorityKeyID_serial = v
1.1803 +
1.1804 + # Basic constraints
1.1805 + self.basicConstraintsCritical = False
1.1806 + self.basicConstraints=None
1.1807 + v = fields_dict[" X509v3 Basic Constraints:"]
1.1808 + if v:
1.1809 + self.basicConstraints = {}
1.1810 + v,t = v.split('\n',2)[:2]
1.1811 + if "critical" in v:
1.1812 + self.basicConstraintsCritical = True
1.1813 + if "CA:" in t:
1.1814 + self.basicConstraints["CA"] = t.split('CA:')[1][:4] == "TRUE"
1.1815 + if "pathlen:" in t:
1.1816 + self.basicConstraints["pathlen"] = int(t.split('pathlen:')[1])
1.1817 +
1.1818 + # X509v3 Key Usage
1.1819 + self.keyUsage = []
1.1820 + v = fields_dict[" X509v3 Key Usage:"]
1.1821 + if v:
1.1822 + # man 5 x509v3_config
1.1823 + ku_mapping = {"Digital Signature": "digitalSignature",
1.1824 + "Non Repudiation": "nonRepudiation",
1.1825 + "Key Encipherment": "keyEncipherment",
1.1826 + "Data Encipherment": "dataEncipherment",
1.1827 + "Key Agreement": "keyAgreement",
1.1828 + "Certificate Sign": "keyCertSign",
1.1829 + "CRL Sign": "cRLSign",
1.1830 + "Encipher Only": "encipherOnly",
1.1831 + "Decipher Only": "decipherOnly"}
1.1832 + v = v.split('\n',2)[1]
1.1833 + l = map(lambda x: x.strip(), v.split(','))
1.1834 + while l:
1.1835 + c = l.pop()
1.1836 + if ku_mapping.has_key(c):
1.1837 + self.keyUsage.append(ku_mapping[c])
1.1838 + else:
1.1839 + self.keyUsage.append(c) # Add it anyway
1.1840 + print "Found unknown X509v3 Key Usage: '%s'" % c
1.1841 + print "Report it to arno (at) natisbad.org for addition"
1.1842 +
1.1843 + # X509v3 Extended Key Usage
1.1844 + self.extKeyUsage = []
1.1845 + v = fields_dict[" X509v3 Extended Key Usage:"]
1.1846 + if v:
1.1847 + # man 5 x509v3_config:
1.1848 + eku_mapping = {"TLS Web Server Authentication": "serverAuth",
1.1849 + "TLS Web Client Authentication": "clientAuth",
1.1850 + "Code Signing": "codeSigning",
1.1851 + "E-mail Protection": "emailProtection",
1.1852 + "Time Stamping": "timeStamping",
1.1853 + "Microsoft Individual Code Signing": "msCodeInd",
1.1854 + "Microsoft Commercial Code Signing": "msCodeCom",
1.1855 + "Microsoft Trust List Signing": "msCTLSign",
1.1856 + "Microsoft Encrypted File System": "msEFS",
1.1857 + "Microsoft Server Gated Crypto": "msSGC",
1.1858 + "Netscape Server Gated Crypto": "nsSGC",
1.1859 + "IPSec End System": "iPsecEndSystem",
1.1860 + "IPSec Tunnel": "iPsecTunnel",
1.1861 + "IPSec User": "iPsecUser"}
1.1862 + v = v.split('\n',2)[1]
1.1863 + l = map(lambda x: x.strip(), v.split(','))
1.1864 + while l:
1.1865 + c = l.pop()
1.1866 + if eku_mapping.has_key(c):
1.1867 + self.extKeyUsage.append(eku_mapping[c])
1.1868 + else:
1.1869 + self.extKeyUsage.append(c) # Add it anyway
1.1870 + print "Found unknown X509v3 Extended Key Usage: '%s'" % c
1.1871 + print "Report it to arno (at) natisbad.org for addition"
1.1872 +
1.1873 + # CRL Distribution points
1.1874 + self.cRLDistributionPoints = []
1.1875 + v = fields_dict[" X509v3 CRL Distribution Points:"]
1.1876 + if v:
1.1877 + v = v.split("\n\n", 1)[0]
1.1878 + v = v.split("URI:")[1:]
1.1879 + self.CRLDistributionPoints = map(lambda x: x.strip(), v)
1.1880 +
1.1881 + # Authority Information Access: list of tuples ("method", "location")
1.1882 + self.authorityInfoAccess = []
1.1883 + v = fields_dict[" Authority Information Access:"]
1.1884 + if v:
1.1885 + v = v.split("\n\n", 1)[0]
1.1886 + v = v.split("\n")[1:]
1.1887 + for e in v:
1.1888 + method, location = map(lambda x: x.strip(), e.split(" - ", 1))
1.1889 + self.authorityInfoAccess.append((method, location))
1.1890 +
1.1891 + # signature field
1.1892 + v = fields_dict[" Signature Algorithm:" ]
1.1893 + self.sig = None
1.1894 + if v:
1.1895 + v = v.split('\n',1)[1]
1.1896 + v = v.replace(' ', '').replace('\n', '')
1.1897 + self.sig = "".join(map(lambda x: chr(int(x, 16)), v.split(':')))
1.1898 + self.sigLen = len(self.sig)
1.1899 + if self.sig is None:
1.1900 + raise Exception(error_msg)
1.1901 +
1.1902 + def isIssuerCert(self, other):
1.1903 + """
1.1904 + True if 'other' issued 'self', i.e.:
1.1905 + - self.issuer == other.subject
1.1906 + - self is signed by other
1.1907 + """
1.1908 + # XXX should be done on raw values, instead of their textual repr
1.1909 + if self.issuer != other.subject:
1.1910 + return False
1.1911 +
1.1912 + # Sanity check regarding modulus length and the
1.1913 + # signature length
1.1914 + keyLen = (other.modulusLen + 7)/8
1.1915 + if keyLen != self.sigLen:
1.1916 + return False
1.1917 +
1.1918 + unenc = other.encrypt(self.sig) # public key encryption, i.e. decrypt
1.1919 +
1.1920 + # XXX Check block type (00 or 01 and type of padding)
1.1921 + unenc = unenc[1:]
1.1922 + if not '\x00' in unenc:
1.1923 + return False
1.1924 + pos = unenc.index('\x00')
1.1925 + unenc = unenc[pos+1:]
1.1926 +
1.1927 + found = None
1.1928 + for k in _hashFuncParams.keys():
1.1929 + if self.sigAlg.startswith(k):
1.1930 + found = k
1.1931 + break
1.1932 + if not found:
1.1933 + return False
1.1934 + hlen, hfunc, digestInfo = _hashFuncParams[k]
1.1935 +
1.1936 + if len(unenc) != (hlen+len(digestInfo)):
1.1937 + return False
1.1938 +
1.1939 + if not unenc.startswith(digestInfo):
1.1940 + return False
1.1941 +
1.1942 + h = unenc[-hlen:]
1.1943 + myh = hfunc(self.tbsCertificate)
1.1944 +
1.1945 + return h == myh
1.1946 +
1.1947 + def chain(self, certlist):
1.1948 + """
1.1949 + Construct the chain of certificates leading from 'self' to the
1.1950 + self signed root using the certificates in 'certlist'. If the
1.1951 + list does not provide all the required certs to go to the root
1.1952 + the function returns a incomplete chain starting with the
1.1953 + certificate. This fact can be tested by tchecking if the last
1.1954 + certificate of the returned chain is self signed (if c is the
1.1955 + result, c[-1].isSelfSigned())
1.1956 + """
1.1957 + d = {}
1.1958 + for c in certlist:
1.1959 + # XXX we should check if we have duplicate
1.1960 + d[c.subject] = c
1.1961 + res = [self]
1.1962 + cur = self
1.1963 + while not cur.isSelfSigned():
1.1964 + if d.has_key(cur.issuer):
1.1965 + possible_issuer = d[cur.issuer]
1.1966 + if cur.isIssuerCert(possible_issuer):
1.1967 + res.append(possible_issuer)
1.1968 + cur = possible_issuer
1.1969 + else:
1.1970 + break
1.1971 + return res
1.1972 +
1.1973 + def remainingDays(self, now=None):
1.1974 + """
1.1975 + Based on the value of notBefore field, returns the number of
1.1976 + days the certificate will still be valid. The date used for the
1.1977 + comparison is the current and local date, as returned by
1.1978 + time.localtime(), except if 'now' argument is provided another
1.1979 + one. 'now' argument can be given as either a time tuple or a string
1.1980 + representing the date. Accepted format for the string version
1.1981 + are:
1.1982 +
1.1983 + - '%b %d %H:%M:%S %Y %Z' e.g. 'Jan 30 07:38:59 2008 GMT'
1.1984 + - '%m/%d/%y' e.g. '01/30/08' (less precise)
1.1985 +
1.1986 + If the certificate is no more valid at the date considered, then,
1.1987 + a negative value is returned representing the number of days
1.1988 + since it has expired.
1.1989 +
1.1990 + The number of days is returned as a float to deal with the unlikely
1.1991 + case of certificates that are still just valid.
1.1992 + """
1.1993 + if now is None:
1.1994 + now = time.localtime()
1.1995 + elif type(now) is str:
1.1996 + try:
1.1997 + if '/' in now:
1.1998 + now = time.strptime(now, '%m/%d/%y')
1.1999 + else:
1.2000 + now = time.strptime(now, '%b %d %H:%M:%S %Y %Z')
1.2001 + except:
1.2002 + warning("Bad time string provided '%s'. Using current time" % now)
1.2003 + now = time.localtime()
1.2004 +
1.2005 + now = time.mktime(now)
1.2006 + nft = time.mktime(self.notAfter)
1.2007 + diff = (nft - now)/(24.*3600)
1.2008 + return diff
1.2009 +
1.2010 +
1.2011 + # return SHA-1 hash of cert embedded public key
1.2012 + # !! At the moment, the trailing 0 is in the hashed string if any
1.2013 + def keyHash(self):
1.2014 + m = self.modulus_hexdump
1.2015 + res = []
1.2016 + i = 0
1.2017 + l = len(m)
1.2018 + while i<l: # get a string version of modulus
1.2019 + res.append(struct.pack("B", int(m[i:i+2], 16)))
1.2020 + i += 2
1.2021 + return sha.new("".join(res)).digest()
1.2022 +
1.2023 + def output(self, fmt="DER"):
1.2024 + if fmt == "DER":
1.2025 + return self.dercert
1.2026 + elif fmt == "PEM":
1.2027 + return self.pemcert
1.2028 + elif fmt == "TXT":
1.2029 + return self.textcert
1.2030 +
1.2031 + def export(self, filename, fmt="DER"):
1.2032 + """
1.2033 + Export certificate in 'fmt' format (PEM, DER or TXT) to file 'filename'
1.2034 + """
1.2035 + f = open(filename, "wb")
1.2036 + f.write(self.output(fmt))
1.2037 + f.close()
1.2038 +
1.2039 + def isSelfSigned(self):
1.2040 + """
1.2041 + Return True if the certificate is self signed:
1.2042 + - issuer and subject are the same
1.2043 + - the signature of the certificate is valid.
1.2044 + """
1.2045 + if self.issuer == self.subject:
1.2046 + return self.isIssuerCert(self)
1.2047 + return False
1.2048 +
1.2049 + # Print main informations stored in certificate
1.2050 + def show(self):
1.2051 + print "Serial: %s" % self.serial
1.2052 + print "Issuer: " + self.issuer
1.2053 + print "Subject: " + self.subject
1.2054 + print "Validity: %s to %s" % (self.notBefore_str_simple,
1.2055 + self.notAfter_str_simple)
1.2056 +
1.2057 + def __repr__(self):
1.2058 + return "[X.509 Cert. Subject:%s, Issuer:%s]" % (self.subject, self.issuer)
1.2059 +
1.2060 + def __str__(self):
1.2061 + return self.dercert
1.2062 +
1.2063 + def verifychain(self, anchors, untrusted=None):
1.2064 + """
1.2065 + Perform verification of certificate chains for that certificate. The
1.2066 + behavior of verifychain method is mapped (and also based) on openssl
1.2067 + verify userland tool (man 1 verify).
1.2068 + A list of anchors is required. untrusted parameter can be provided
1.2069 + a list of untrusted certificates that can be used to reconstruct the
1.2070 + chain.
1.2071 +
1.2072 + If you have a lot of certificates to verify against the same
1.2073 + list of anchor, consider constructing this list as a cafile
1.2074 + and use .verifychain_from_cafile() instead.
1.2075 + """
1.2076 + cafile = create_temporary_ca_file(anchors)
1.2077 + if not cafile:
1.2078 + return False
1.2079 + untrusted_file = None
1.2080 + if untrusted:
1.2081 + untrusted_file = create_temporary_ca_file(untrusted) # hack
1.2082 + if not untrusted_file:
1.2083 + os.unlink(cafile)
1.2084 + return False
1.2085 + res = self.verifychain_from_cafile(cafile,
1.2086 + untrusted_file=untrusted_file)
1.2087 + os.unlink(cafile)
1.2088 + if untrusted_file:
1.2089 + os.unlink(untrusted_file)
1.2090 + return res
1.2091 +
1.2092 + def verifychain_from_cafile(self, cafile, untrusted_file=None):
1.2093 + """
1.2094 + Does the same job as .verifychain() but using the list of anchors
1.2095 + from the cafile. This is useful (because more efficient) if
1.2096 + you have a lot of certificates to verify do it that way: it
1.2097 + avoids the creation of a cafile from anchors at each call.
1.2098 +
1.2099 + As for .verifychain(), a list of untrusted certificates can be
1.2100 + passed (as a file, this time)
1.2101 + """
1.2102 + u = ""
1.2103 + if untrusted_file:
1.2104 + u = "-untrusted %s" % untrusted_file
1.2105 + try:
1.2106 + cmd = "openssl verify -CAfile %s %s " % (cafile, u)
1.2107 + pemcert = self.output(fmt="PEM")
1.2108 + cmdres = self._apply_ossl_cmd(cmd, pemcert)
1.2109 + except:
1.2110 + return False
1.2111 + return cmdres.endswith("\nOK\n") or cmdres.endswith(": OK\n")
1.2112 +
1.2113 + def verifychain_from_capath(self, capath, untrusted_file=None):
1.2114 + """
1.2115 + Does the same job as .verifychain_from_cafile() but using the list
1.2116 + of anchors in capath directory. The directory should contain
1.2117 + certificates files in PEM format with associated links as
1.2118 + created using c_rehash utility (man c_rehash).
1.2119 +
1.2120 + As for .verifychain_from_cafile(), a list of untrusted certificates
1.2121 + can be passed as a file (concatenation of the certificates in
1.2122 + PEM format)
1.2123 + """
1.2124 + u = ""
1.2125 + if untrusted_file:
1.2126 + u = "-untrusted %s" % untrusted_file
1.2127 + try:
1.2128 + cmd = "openssl verify -CApath %s %s " % (capath, u)
1.2129 + pemcert = self.output(fmt="PEM")
1.2130 + cmdres = self._apply_ossl_cmd(cmd, pemcert)
1.2131 + except:
1.2132 + return False
1.2133 + return cmdres.endswith("\nOK\n") or cmdres.endswith(": OK\n")
1.2134 +
1.2135 + def is_revoked(self, crl_list):
1.2136 + """
1.2137 + Given a list of trusted CRL (their signature has already been
1.2138 + verified with trusted anchors), this function returns True if
1.2139 + the certificate is marked as revoked by one of those CRL.
1.2140 +
1.2141 + Note that if the Certificate was on hold in a previous CRL and
1.2142 + is now valid again in a new CRL and bot are in the list, it
1.2143 + will be considered revoked: this is because _all_ CRLs are
1.2144 + checked (not only the freshest) and revocation status is not
1.2145 + handled.
1.2146 +
1.2147 + Also note that the check on the issuer is performed on the
1.2148 + Authority Key Identifier if available in _both_ the CRL and the
1.2149 + Cert. Otherwise, the issuers are simply compared.
1.2150 + """
1.2151 + for c in crl_list:
1.2152 + if (self.authorityKeyID is not None and
1.2153 + c.authorityKeyID is not None and
1.2154 + self.authorityKeyID == c.authorityKeyID):
1.2155 + return self.serial in map(lambda x: x[0], c.revoked_cert_serials)
1.2156 + elif (self.issuer == c.issuer):
1.2157 + return self.serial in map(lambda x: x[0], c.revoked_cert_serials)
1.2158 + return False
1.2159 +
1.2160 +def print_chain(l):
1.2161 + llen = len(l) - 1
1.2162 + if llen < 0:
1.2163 + return ""
1.2164 + c = l[llen]
1.2165 + llen -= 1
1.2166 + s = "_ "
1.2167 + if not c.isSelfSigned():
1.2168 + s = "_ ... [Missing Root]\n"
1.2169 + else:
1.2170 + s += "%s [Self Signed]\n" % c.subject
1.2171 + i = 1
1.2172 + while (llen != -1):
1.2173 + c = l[llen]
1.2174 + s += "%s\_ %s" % (" "*i, c.subject)
1.2175 + if llen != 0:
1.2176 + s += "\n"
1.2177 + i += 2
1.2178 + llen -= 1
1.2179 + print s
1.2180 +
1.2181 +# import popen2
1.2182 +# a=popen2.Popen3("openssl crl -text -inform DER -noout ", capturestderr=True)
1.2183 +# a.tochild.write(open("samples/klasa1.crl").read())
1.2184 +# a.tochild.close()
1.2185 +# a.poll()
1.2186 +
1.2187 +class CRL(OSSLHelper):
1.2188 + # Below are the fields we recognize in the -text output of openssl
1.2189 + # and from which we extract information. We expect them in that
1.2190 + # order. Number of spaces does matter.
1.2191 + possible_fields = [ " Version",
1.2192 + " Signature Algorithm:",
1.2193 + " Issuer:",
1.2194 + " Last Update:",
1.2195 + " Next Update:",
1.2196 + " CRL extensions:",
1.2197 + " X509v3 Issuer Alternative Name:",
1.2198 + " X509v3 Authority Key Identifier:",
1.2199 + " keyid:",
1.2200 + " DirName:",
1.2201 + " serial:",
1.2202 + " X509v3 CRL Number:",
1.2203 + "Revoked Certificates:",
1.2204 + "No Revoked Certificates.",
1.2205 + " Signature Algorithm:" ]
1.2206 + possible_fields_count = len(possible_fields)
1.2207 +
1.2208 + def __init__(self, crlpath):
1.2209 + error_msg = "Unable to import CRL."
1.2210 +
1.2211 + fields_dict = {}
1.2212 + for k in self.possible_fields:
1.2213 + fields_dict[k] = None
1.2214 +
1.2215 + self.crlpath = None
1.2216 + rawcrl = None
1.2217 +
1.2218 + if (not '\x00' in crlpath) and os.path.isfile(crlpath):
1.2219 + self.crlpath = crlpath
1.2220 + cert_size = os.path.getsize(crlpath)
1.2221 + if cert_size > MAX_CRL_SIZE:
1.2222 + raise Exception(error_msg)
1.2223 + try:
1.2224 + f = open(crlpath)
1.2225 + rawcrl = f.read()
1.2226 + f.close()
1.2227 + except:
1.2228 + raise Exception(error_msg)
1.2229 + else:
1.2230 + rawcrl = crlpath
1.2231 +
1.2232 + if rawcrl is None:
1.2233 + raise Exception(error_msg)
1.2234 +
1.2235 + self.rawcrl = rawcrl
1.2236 +
1.2237 + # Let's try to get file format : PEM or DER.
1.2238 + fmtstr = 'openssl crl -text -inform %s -noout '
1.2239 + convertstr = 'openssl crl -inform %s -outform %s '
1.2240 + crl_header = "-----BEGIN X509 CRL-----"
1.2241 + crl_footer = "-----END X509 CRL-----"
1.2242 + l = rawcrl.split(crl_header, 1)
1.2243 + if len(l) == 2: # looks like PEM
1.2244 + tmp = l[1]
1.2245 + l = tmp.split(crl_footer, 1)
1.2246 + if len(l) == 2:
1.2247 + tmp = l[0]
1.2248 + rawcrl = "%s%s%s\n" % (crl_header, tmp, crl_footer)
1.2249 + else:
1.2250 + raise Exception(error_msg)
1.2251 + r,w,e = popen2.popen3(fmtstr % "PEM")
1.2252 + w.write(rawcrl)
1.2253 + w.close()
1.2254 + textcrl = r.read()
1.2255 + r.close()
1.2256 + res = e.read()
1.2257 + e.close()
1.2258 + if res == '':
1.2259 + self.format = "PEM"
1.2260 + self.pemcrl = rawcrl
1.2261 + self.textcrl = textcrl
1.2262 + cmd = convertstr % ("PEM", "DER")
1.2263 + self.dercrl = self._apply_ossl_cmd(cmd, rawcrl)
1.2264 + else:
1.2265 + raise Exception(error_msg)
1.2266 + else: # not PEM, try DER
1.2267 + r,w,e = popen2.popen3(fmtstr % "DER")
1.2268 + w.write(rawcrl)
1.2269 + w.close()
1.2270 + textcrl = r.read()
1.2271 + r.close()
1.2272 + res = e.read()
1.2273 + if res == '':
1.2274 + self.format = "DER"
1.2275 + self.dercrl = rawcrl
1.2276 + self.textcrl = textcrl
1.2277 + cmd = convertstr % ("DER", "PEM")
1.2278 + self.pemcrl = self._apply_ossl_cmd(cmd, rawcrl)
1.2279 + cmd = convertstr % ("DER", "DER")
1.2280 + self.dercrl = self._apply_ossl_cmd(cmd, rawcrl)
1.2281 + else:
1.2282 + raise Exception(error_msg)
1.2283 +
1.2284 + self.osslcmdbase = 'openssl crl -inform %s ' % self.format
1.2285 +
1.2286 + r,w,e = popen2.popen3('openssl asn1parse -inform DER ')
1.2287 + w.write(self.dercrl)
1.2288 + w.close()
1.2289 + self.asn1parsecrl = r.read()
1.2290 + r.close()
1.2291 + res = e.read()
1.2292 + e.close()
1.2293 + if res != '':
1.2294 + raise Exception(error_msg)
1.2295 +
1.2296 + # Grab _raw_ X509v3 Authority Key Identifier, if any.
1.2297 + tmp = self.asn1parsecrl.split(":X509v3 Authority Key Identifier", 1)
1.2298 + self.authorityKeyID = None
1.2299 + if len(tmp) == 2:
1.2300 + tmp = tmp[1]
1.2301 + tmp = tmp.split("[HEX DUMP]:", 1)[1]
1.2302 + self.authorityKeyID=tmp.split('\n',1)[0]
1.2303 +
1.2304 + # Parse the -text output of openssl to make things available
1.2305 + tmp = self.textcrl.split('\n', 1)[1]
1.2306 + l = tmp.split('\n', 1)
1.2307 + if len(l) != 2:
1.2308 + raise Exception(error_msg)
1.2309 + cur, tmp = l
1.2310 + i = 0
1.2311 + k = self.possible_fields[i] # Version
1.2312 + cur = cur[len(k):] + '\n'
1.2313 + while k:
1.2314 + l = tmp.split('\n', 1)
1.2315 + if len(l) != 2: # Over
1.2316 + fields_dict[k] = cur
1.2317 + break
1.2318 + l, tmp = l
1.2319 +
1.2320 + newkey = 0
1.2321 + # skip fields we have already seen, this is the purpose of 'i'
1.2322 + for j in range(i, self.possible_fields_count):
1.2323 + f = self.possible_fields[j]
1.2324 + if l.startswith(f):
1.2325 + fields_dict[k] = cur
1.2326 + cur = l[len(f):] + '\n'
1.2327 + k = f
1.2328 + newkey = 1
1.2329 + i = j+1
1.2330 + break
1.2331 + if newkey == 1:
1.2332 + continue
1.2333 + cur += l + '\n'
1.2334 +
1.2335 + # version
1.2336 + v = fields_dict[" Version"]
1.2337 + self.version = None
1.2338 + if v:
1.2339 + self.version = int(v[1:2])
1.2340 + if self.version is None:
1.2341 + raise Exception(error_msg)
1.2342 +
1.2343 + # signature algorithm
1.2344 + v = fields_dict[" Signature Algorithm:"]
1.2345 + self.sigAlg = None
1.2346 + if v:
1.2347 + v = v.split('\n',1)[0]
1.2348 + v = v.strip()
1.2349 + self.sigAlg = v
1.2350 + if self.sigAlg is None:
1.2351 + raise Exception(error_msg)
1.2352 +
1.2353 + # issuer
1.2354 + v = fields_dict[" Issuer:"]
1.2355 + self.issuer = None
1.2356 + if v:
1.2357 + v = v.split('\n',1)[0]
1.2358 + v = v.strip()
1.2359 + self.issuer = v
1.2360 + if self.issuer is None:
1.2361 + raise Exception(error_msg)
1.2362 +
1.2363 + # last update
1.2364 + v = fields_dict[" Last Update:"]
1.2365 + self.lastUpdate_str = None
1.2366 + if v:
1.2367 + v = v.split('\n',1)[0]
1.2368 + v = v.strip()
1.2369 + self.lastUpdate_str = v
1.2370 + if self.lastUpdate_str is None:
1.2371 + raise Exception(error_msg)
1.2372 + self.lastUpdate = time.strptime(self.lastUpdate_str,
1.2373 + "%b %d %H:%M:%S %Y %Z")
1.2374 + self.lastUpdate_str_simple = time.strftime("%x", self.lastUpdate)
1.2375 +
1.2376 + # next update
1.2377 + v = fields_dict[" Next Update:"]
1.2378 + self.nextUpdate_str = None
1.2379 + if v:
1.2380 + v = v.split('\n',1)[0]
1.2381 + v = v.strip()
1.2382 + self.nextUpdate_str = v
1.2383 + if self.nextUpdate_str is None:
1.2384 + raise Exception(error_msg)
1.2385 + self.nextUpdate = time.strptime(self.nextUpdate_str,
1.2386 + "%b %d %H:%M:%S %Y %Z")
1.2387 + self.nextUpdate_str_simple = time.strftime("%x", self.nextUpdate)
1.2388 +
1.2389 + # XXX Do something for Issuer Alternative Name
1.2390 +
1.2391 + # Authority Key Identifier: keyid, dirname and serial
1.2392 + self.authorityKeyID_keyid = None
1.2393 + self.authorityKeyID_dirname = None
1.2394 + self.authorityKeyID_serial = None
1.2395 + if self.authorityKeyID: # (hex version already done using asn1parse)
1.2396 + v = fields_dict[" keyid:"]
1.2397 + if v:
1.2398 + v = v.split('\n',1)[0]
1.2399 + v = v.strip().replace(':', '')
1.2400 + self.authorityKeyID_keyid = v
1.2401 + v = fields_dict[" DirName:"]
1.2402 + if v:
1.2403 + v = v.split('\n',1)[0]
1.2404 + self.authorityKeyID_dirname = v
1.2405 + v = fields_dict[" serial:"]
1.2406 + if v:
1.2407 + v = v.split('\n',1)[0]
1.2408 + v = v.strip().replace(':', '')
1.2409 + self.authorityKeyID_serial = v
1.2410 +
1.2411 + # number
1.2412 + v = fields_dict[" X509v3 CRL Number:"]
1.2413 + self.number = None
1.2414 + if v:
1.2415 + v = v.split('\n',2)[1]
1.2416 + v = v.strip()
1.2417 + self.number = int(v)
1.2418 +
1.2419 + # Get the list of serial numbers of revoked certificates
1.2420 + self.revoked_cert_serials = []
1.2421 + v = fields_dict["Revoked Certificates:"]
1.2422 + t = fields_dict["No Revoked Certificates."]
1.2423 + if (t is None and v is not None):
1.2424 + v = v.split("Serial Number: ")[1:]
1.2425 + for r in v:
1.2426 + s,d = r.split('\n', 1)
1.2427 + s = s.split('\n', 1)[0]
1.2428 + d = d.split("Revocation Date:", 1)[1]
1.2429 + d = time.strptime(d.strip(), "%b %d %H:%M:%S %Y %Z")
1.2430 + self.revoked_cert_serials.append((s,d))
1.2431 +
1.2432 + # signature field
1.2433 + v = fields_dict[" Signature Algorithm:" ]
1.2434 + self.sig = None
1.2435 + if v:
1.2436 + v = v.split('\n',1)[1]
1.2437 + v = v.replace(' ', '').replace('\n', '')
1.2438 + self.sig = "".join(map(lambda x: chr(int(x, 16)), v.split(':')))
1.2439 + self.sigLen = len(self.sig)
1.2440 + if self.sig is None:
1.2441 + raise Exception(error_msg)
1.2442 +
1.2443 + def __str__(self):
1.2444 + return self.dercrl
1.2445 +
1.2446 + # Print main informations stored in CRL
1.2447 + def show(self):
1.2448 + print "Version: %d" % self.version
1.2449 + print "sigAlg: " + self.sigAlg
1.2450 + print "Issuer: " + self.issuer
1.2451 + print "lastUpdate: %s" % self.lastUpdate_str_simple
1.2452 + print "nextUpdate: %s" % self.nextUpdate_str_simple
1.2453 +
1.2454 + def verify(self, anchors):
1.2455 + """
1.2456 + Return True if the CRL is signed by one of the provided
1.2457 + anchors. False on error (invalid signature, missing anchorand, ...)
1.2458 + """
1.2459 + cafile = create_temporary_ca_file(anchors)
1.2460 + if cafile is None:
1.2461 + return False
1.2462 + try:
1.2463 + cmd = self.osslcmdbase + '-noout -CAfile %s 2>&1' % cafile
1.2464 + cmdres = self._apply_ossl_cmd(cmd, self.rawcrl)
1.2465 + except:
1.2466 + os.unlink(cafile)
1.2467 + return False
1.2468 + os.unlink(cafile)
1.2469 + return "verify OK" in cmdres
1.2470 +
1.2471 +
1.2472 +
1.2473 +