You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
304 lines
9.8 KiB
304 lines
9.8 KiB
#!/usr/bin/env python |
|
# |
|
# Electrum - lightweight Bitcoin client |
|
# Copyright (C) 2014 Thomas Voegtlin |
|
# |
|
# This program is free software: you can redistribute it and/or modify |
|
# it under the terms of the GNU General Public License as published by |
|
# the Free Software Foundation, either version 3 of the License, or |
|
# (at your option) any later version. |
|
# |
|
# This program is distributed in the hope that it will be useful, |
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of |
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|
# GNU General Public License for more details. |
|
# |
|
# You should have received a copy of the GNU General Public License |
|
# along with this program. If not, see <http://www.gnu.org/licenses/>. |
|
|
|
|
|
from datetime import datetime |
|
import sys |
|
|
|
import tlslite |
|
import util |
|
from util import profiler, print_error |
|
|
|
from asn1tinydecoder import asn1_node_root, asn1_get_all, asn1_get_value, \ |
|
asn1_get_value_of_type, asn1_node_next, asn1_node_first_child, \ |
|
asn1_read_length, asn1_node_is_child_of, \ |
|
bytestr_to_int, bitstr_to_bytestr |
|
|
|
# workaround https://github.com/trevp/tlslite/issues/15 |
|
tlslite.utils.cryptomath.pycryptoLoaded = False |
|
|
|
|
|
# algo OIDs |
|
ALGO_RSA_SHA1 = '1.2.840.113549.1.1.5' |
|
ALGO_RSA_SHA256 = '1.2.840.113549.1.1.11' |
|
ALGO_RSA_SHA384 = '1.2.840.113549.1.1.12' |
|
ALGO_RSA_SHA512 = '1.2.840.113549.1.1.13' |
|
ALGO_ECDSA_SHA256 = '1.2.840.10045.4.3.2' |
|
|
|
# prefixes, see http://stackoverflow.com/questions/3713774/c-sharp-how-to-calculate-asn-1-der-encoding-of-a-particular-hash-algorithm |
|
PREFIX_RSA_SHA256 = bytearray([0x30,0x31,0x30,0x0d,0x06,0x09,0x60,0x86,0x48,0x01,0x65,0x03,0x04,0x02,0x01,0x05,0x00,0x04,0x20]) |
|
PREFIX_RSA_SHA384 = bytearray([0x30,0x41,0x30,0x0d,0x06,0x09,0x60,0x86,0x48,0x01,0x65,0x03,0x04,0x02,0x02,0x05,0x00,0x04,0x30]) |
|
PREFIX_RSA_SHA512 = bytearray([0x30,0x51,0x30,0x0d,0x06,0x09,0x60,0x86,0x48,0x01,0x65,0x03,0x04,0x02,0x03,0x05,0x00,0x04,0x40]) |
|
|
|
|
|
class CertificateError(Exception): |
|
pass |
|
|
|
|
|
def decode_OID(s): |
|
s = map(ord, s) |
|
r = [] |
|
r.append(s[0] / 40) |
|
r.append(s[0] % 40) |
|
k = 0 |
|
for i in s[1:]: |
|
if i < 128: |
|
r.append(i + 128*k) |
|
k = 0 |
|
else: |
|
k = (i - 128) + 128*k |
|
return '.'.join(map(str, r)) |
|
|
|
def encode_OID(oid): |
|
x = map(int, oid.split('.')) |
|
s = chr(x[0]*40 + x[1]) |
|
for i in x[2:]: |
|
ss = chr(i % 128) |
|
while i > 128: |
|
i = i / 128 |
|
ss = chr(128 + i % 128) + ss |
|
s += ss |
|
return s |
|
|
|
def asn1_get_children(der, i): |
|
nodes = [] |
|
ii = asn1_node_first_child(der,i) |
|
nodes.append(ii) |
|
while ii[2]<i[2]: |
|
ii = asn1_node_next(der,ii) |
|
nodes.append(ii) |
|
return nodes |
|
|
|
def asn1_get_sequence(s): |
|
return map(lambda j: asn1_get_value(s, j), asn1_get_children(s, asn1_node_root(s))) |
|
|
|
def asn1_get_dict(der, i): |
|
p = {} |
|
for ii in asn1_get_children(der, i): |
|
for iii in asn1_get_children(der, ii): |
|
iiii = asn1_node_first_child(der, iii) |
|
oid = decode_OID(asn1_get_value_of_type(der, iiii, 'OBJECT IDENTIFIER')) |
|
iiii = asn1_node_next(der, iiii) |
|
value = asn1_get_value(der, iiii) |
|
p[oid] = value |
|
return p |
|
|
|
|
|
class X509(tlslite.X509): |
|
|
|
def parseBinary(self, b): |
|
|
|
# call tlslite method first |
|
tlslite.X509.parseBinary(self, b) |
|
|
|
der = str(b) |
|
root = asn1_node_root(der) |
|
cert = asn1_node_first_child(der, root) |
|
# data for signature |
|
self.data = asn1_get_all(der, cert) |
|
|
|
# optional version field |
|
if asn1_get_value(der, cert)[0] == chr(0xa0): |
|
version = asn1_node_first_child(der, cert) |
|
serial_number = asn1_node_next(der, version) |
|
else: |
|
serial_number = asn1_node_first_child(der, cert) |
|
self.serial_number = bytestr_to_int(asn1_get_value_of_type(der, serial_number, 'INTEGER')) |
|
|
|
# signature algorithm |
|
sig_algo = asn1_node_next(der, serial_number) |
|
ii = asn1_node_first_child(der, sig_algo) |
|
self.sig_algo = decode_OID(asn1_get_value_of_type(der, ii, 'OBJECT IDENTIFIER')) |
|
|
|
# issuer |
|
issuer = asn1_node_next(der, sig_algo) |
|
self.issuer = asn1_get_dict(der, issuer) |
|
|
|
# validity |
|
validity = asn1_node_next(der, issuer) |
|
ii = asn1_node_first_child(der, validity) |
|
self.notBefore = asn1_get_value_of_type(der, ii, 'UTCTime') |
|
ii = asn1_node_next(der,ii) |
|
self.notAfter = asn1_get_value_of_type(der, ii, 'UTCTime') |
|
|
|
# subject |
|
subject = asn1_node_next(der, validity) |
|
self.subject = asn1_get_dict(der, subject) |
|
subject_pki = asn1_node_next(der, subject) |
|
|
|
# extensions |
|
self.CA = False |
|
self.AKI = None |
|
self.SKI = None |
|
i = subject_pki |
|
while i[2] < cert[2]: |
|
i = asn1_node_next(der, i) |
|
d = asn1_get_dict(der, i) |
|
for oid, value in d.items(): |
|
if oid == '2.5.29.19': |
|
# Basic Constraints |
|
self.CA = bool(value) |
|
elif oid == '2.5.29.14': |
|
# Subject Key Identifier |
|
r = asn1_node_root(value) |
|
value = asn1_get_value_of_type(value, r, 'OCTET STRING') |
|
self.SKI = value.encode('hex') |
|
elif oid == '2.5.29.35': |
|
# Authority Key Identifier |
|
self.AKI = asn1_get_sequence(value)[0].encode('hex') |
|
else: |
|
pass |
|
|
|
# cert signature |
|
cert_sig_algo = asn1_node_next(der, cert) |
|
ii = asn1_node_first_child(der, cert_sig_algo) |
|
self.cert_sig_algo = decode_OID(asn1_get_value_of_type(der, ii, 'OBJECT IDENTIFIER')) |
|
cert_sig = asn1_node_next(der, cert_sig_algo) |
|
self.signature = asn1_get_value(der, cert_sig)[1:] |
|
|
|
def get_keyID(self): |
|
# http://security.stackexchange.com/questions/72077/validating-an-ssl-certificate-chain-according-to-rfc-5280-am-i-understanding-th |
|
return self.SKI if self.SKI else repr(self.subject) |
|
|
|
def get_issuer_keyID(self): |
|
return self.AKI if self.AKI else repr(self.issuer) |
|
|
|
def get_common_name(self): |
|
return self.subject.get('2.5.4.3', 'unknown') |
|
|
|
def get_signature(self): |
|
return self.cert_sig_algo, self.signature, self.data |
|
|
|
def check_ca(self): |
|
return self.CA |
|
|
|
def check_date(self): |
|
import time |
|
now = time.time() |
|
TIMESTAMP_FMT = '%y%m%d%H%M%SZ' |
|
not_before = time.mktime(time.strptime(self.notBefore, TIMESTAMP_FMT)) |
|
not_after = time.mktime(time.strptime(self.notAfter, TIMESTAMP_FMT)) |
|
if not_before > now: |
|
raise CertificateError('Certificate has not entered its valid date range.') |
|
if not_after <= now: |
|
raise CertificateError('Certificate has expired.') |
|
|
|
|
|
|
|
class X509CertChain(tlslite.X509CertChain): |
|
pass |
|
|
|
|
|
|
|
|
|
|
|
@profiler |
|
def load_certificates(ca_path): |
|
ca_list = {} |
|
ca_keyID = {} |
|
with open(ca_path, 'r') as f: |
|
s = f.read() |
|
bList = tlslite.utils.pem.dePemList(s, "CERTIFICATE") |
|
for b in bList: |
|
x = X509() |
|
try: |
|
x.parseBinary(b) |
|
x.check_date() |
|
except BaseException as e: |
|
util.print_error("cert error:", e) |
|
continue |
|
|
|
fp = x.getFingerprint() |
|
ca_list[fp] = x |
|
ca_keyID[x.get_keyID()] = fp |
|
|
|
return ca_list, ca_keyID |
|
|
|
|
|
def int_to_bytestr(i): |
|
s = chr(i % 256) |
|
while i > 256: |
|
i >>= 8 |
|
s = chr(i % 256) + s |
|
return s |
|
|
|
def create_csr(commonName, challenge, secexp): |
|
|
|
import ecdsa, hashlib |
|
from bitcoin import point_to_ser |
|
private_key = ecdsa.SigningKey.from_secret_exponent(secexp, curve = ecdsa.SECP256k1) |
|
public_key = private_key.get_verifying_key() |
|
pubkey = point_to_ser(public_key.pubkey.point, False) |
|
asn1_type_table = { |
|
'BOOLEAN': 0x01, 'INTEGER': 0x02, |
|
'BIT STRING': 0x03, 'OCTET STRING': 0x04, |
|
'NULL': 0x05, 'OBJECT IDENTIFIER': 0x06, |
|
'SEQUENCE': 0x30, 'SET': 0x31, |
|
'PrintableString': 0x13, 'IA5String': 0x16, |
|
'UTCTime': 0x17, 'ENUMERATED': 0x0A, |
|
'UTF8String': 0x0C, 'PrintableString': 0x13, |
|
} |
|
def x(t, s): |
|
c = asn1_type_table[t] & 0x3f if type(t) == str else t |
|
l = len(s) |
|
if l < 128: |
|
ls = chr(l) |
|
else: |
|
n = int_to_bytestr(l) |
|
ls = chr(len(n) + 128) + n |
|
return chr(c) + ls + s |
|
x_int = lambda i: x('INTEGER', int_to_bytestr(i)) |
|
x_seq = lambda *items: x('SEQUENCE', ''.join(items)) |
|
x_bitstring = lambda s: x('BIT STRING', s) |
|
x_utf8 = lambda s: x('UTF8String', s) |
|
x_set = lambda *items: x('SET', ''.join(items)) |
|
x_printable = lambda s: x('PrintableString', s) |
|
x_obj = lambda oid: x('OBJECT IDENTIFIER', encode_OID(oid)) |
|
body = x_seq( |
|
x_int(0), |
|
x_seq( |
|
x_set(x_seq(x_obj('2.5.4.3'), x_utf8(commonName))) |
|
), |
|
x_seq( |
|
x_seq( |
|
x_obj('1.2.840.10045.2.1'), |
|
x_obj('1.3.132.0.10') |
|
), |
|
x_bitstring(chr(0) + pubkey) |
|
), |
|
x(0xa0, x_seq(x_obj('1.2.840.113549.1.9.7'), x_set(x_utf8(challenge))) |
|
) |
|
) |
|
signature = private_key.sign_deterministic(body, hashfunc=hashlib.sha256, sigencode = ecdsa.util.sigencode_der) |
|
assert public_key.verify(signature, body, hashfunc=hashlib.sha256, sigdecode = ecdsa.util.sigdecode_der) |
|
csr = x_seq( |
|
body, |
|
x_seq(x_obj(ALGO_ECDSA_SHA256)), |
|
x_bitstring(chr(0) + signature) |
|
) |
|
return csr |
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
import os |
|
csr = create_csr('test@electrum.org', 'blah', 123456) |
|
with open('test.csr', 'w') as f: |
|
o = tlslite.utils.pem.pem(bytearray(csr), "CERTIFICATE REQUEST") |
|
f.write(o) |
|
os.system('openssl asn1parse -i -in test.csr')
|
|
|