Source Code for Module Anomos.Crypto._Certificate

  1  # This program is free software: you can redistribute it and/or modify 
  2  # it under the terms of the GNU General Public License as published by 
  3  # the Free Software Foundation, either version 3 of the License, or 
  4  # (at your option) any later version. 
  5  # 
  6  # This program is distributed in the hope that it will be useful, 
  7  # but WITHOUT ANY WARRANTY; without even the implied warranty of 
  8  # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the 
  9  # GNU General Public License for more details. 
 10  # 
 11  # You should have received a copy of the GNU General Public License 
 12  # along with this program.  If not, see <http://www.gnu.org/licenses/>. 
 13   
 14  import hashlib 
 15  import os 
 16  import sys 
 17  import Anomos.Crypto 
 18   
 19  from Anomos import bttime, LOG as log 
 20  from Anomos.Protocol import toint 
 21  from Anomos.Crypto import global_cryptodir, global_randfile, global_certpath 
 22  from Anomos.Crypto import CryptoError 
 23  from M2Crypto import m2, RSA, EVP, X509, SSL, util as m2util 
 24   
 25  # Cipher Set: 
 26  CIPHER_SET = 'HIGH:!ADH:!MD5:@STRENGTH' 
 27  # Translation: Use high grade encryption, no Anonymous Diffie Hellman, 
 28  # no MD5, sort by strength. 
 29  # On my system, this results in the following cipher set: 
 30  # [john:~]$ openssl ciphers 'HIGH:!ADH:!MD5:@STRENGTH' 
 31  # DHE-RSA-AES256-SHA:DHE-DSS-AES256-SHA:AES256-SHA: 
 32  # EDH-RSA-DES-CBC3-SHA:EDH-DSS-DES-CBC3-SHA: 
 33  # DES-CBC3-SHA:DHE-RSA-AES128-SHA:DHE-DSS-AES128-SHA:AES128-SHA 
 34   
 35  # CTX_OPTIONS: Only allow TLSv1 
 36  CTX_OPTIONS = m2.SSL_OP_NO_SSLv2 | m2.SSL_OP_NO_SSLv3 
 37   
 38  ## X509 Verification Callbacks ## 
 39  SELF_SIGNED_ERR = [ 
 40      m2.X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT, 
 41      m2.X509_V_ERR_SELF_SIGNED_CERT_IN_CHAIN 
 42      ] 
 43   
 44  UNKNOWN_ISSUER_ERR = [ 
 45      m2.X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT, 
 46      m2.X509_V_ERR_UNABLE_TO_GET_ISSUER_CERT_LOCALLY, 
 47      m2.X509_V_ERR_UNABLE_TO_VERIFY_LEAF_SIGNATURE, 
 48      m2.X509_V_ERR_CERT_UNTRUSTED, 
 49      ] 
 50   
 51 -def _verify_callback(ok, store, **kwargs): 
 52      errnum = store.get_error() 
 53      errdepth = store.get_error_depth() 
 54      cert = store.get_current_cert() 
 55      # [DEBUGGING] Localhost exemption for self-signed certificates 
 56      if errnum in SELF_SIGNED_ERR: 
 57          if cert.get_subject().CN == 'localhost': 
 58              ok = 1 
 59      # Allow unknown CA if a context requests it 
 60      if errnum in UNKNOWN_ISSUER_ERR: 
 61          if kwargs.get('allow_unknown_ca', False): 
 62              ok = 1 
 63      # TODO: Certificate Revokation Lists? 
 64      return ok 
 65   
 66 -def mk_verify_cb(**kwargs): 
 67      def vcb(ok, store): 
 68          return _verify_callback(ok, store, **kwargs) 
 69      return vcb 
 70   
 71  ## Certificate class ## 
 72 -class Certificate: 
 73 -    def __init__(self, loc=None, secure=False, tracker=False): 
 74          self.secure = secure 
 75          self.tracker = tracker 
 76          if None in (global_cryptodir, global_randfile): 
 77              raise CryptoError('Crypto not initialized, call initCrypto first') 
 78          self.keyfile = os.path.join(global_cryptodir, '%s-key.pem' % (loc)) 
 79          self.certfile = os.path.join(global_cryptodir, '%s-cert.pem' % (loc)) 
 80   
 81          if not (os.path.exists(self.certfile) and os.path.exists(self.keyfile)): 
 82              if self.tracker: 
 83                  hostname = self._gethostname() 
 84              else: 
 85                  hostname = 'localhost' 
 86              self._create(hostname=hostname) 
 87          else: 
 88              self._load() 
 89   
 90 -    def _gethostname(self): 
 91          from socket import gethostname 
 92          hostname = gethostname() 
 93          tmpname = raw_input("Please enter the tracker's hostname " \ 
 94                          "for the SSL certificate (default: %s): " % hostname) 
 95          if tmpname.strip(" "): 
 96              hostname = tmpname 
 97          return hostname 
 98   
 99 -    def _load(self): 
100          """Attempts to load the certificate and key from self.certfile and self.keyfile, 
101             Generates the certificate and key if they don't exist""" 
102          if not self.secure: 
103              self.rsakey = RSA.load_key(self.keyfile, m2util.no_passphrase_callback) 
104          else: 
105              # Allow 3 attempts before quitting 
106              i = 0 
107              while i < 3: 
108                  try: 
109                      self.rsakey = RSA.load_key(self.keyfile) 
110                      break 
111                  except RSA.RSAError: 
112                      i += 1 
113              else: 
114                  log.warning("\nInvalid password entered, exiting.") 
115                  sys.exit() 
116          self.cert = X509.load_cert(self.certfile) 
117   
118      @Anomos.Crypto.use_rand_file 
119 -    def _create(self, hostname='localhost'): 
120          # Make the RSA key 
121          self.rsakey = RSA.gen_key(2048, m2.RSA_F4) 
122          if self.secure: 
123              # Save the key, AES256-CBC encrypted 
124              self.rsakey.save_key(self.keyfile, 'aes_256_cbc') 
125          else: 
126              # Save the key unencrypted. 
127              self.rsakey.save_key(self.keyfile, None, callback=m2util.no_passphrase_callback) 
128          # Make the public key 
129          pkey = EVP.PKey() 
130          pkey.assign_rsa(self.rsakey, 0) 
131          # Generate the certificate 
132          self.cert = X509.X509() 
133          self.cert.set_serial_number(long(bttime())) 
134          self.cert.set_version(0x2) 
135          self.cert.set_pubkey(pkey) 
136          # Set the name on the certificate 
137          name = X509.X509_Name() 
138          name.CN = hostname 
139          self.cert.set_subject(name) 
140          self.cert.set_issuer(name) 
141          # Set the period of time the cert is valid for (5 years from issue) 
142          notBefore = m2.x509_get_not_before(self.cert.x509) 
143          notAfter = m2.x509_get_not_after(self.cert.x509) 
144          m2.x509_gmtime_adj(notBefore, 0) 
145          m2.x509_gmtime_adj(notAfter, 60*60*24*365*5) 
146          # Sign the certificate 
147          self.cert.sign(pkey, 'ripemd160') 
148          # Save it 
149          self.cert.save_pem(self.certfile) 
150   
151 -    def get_ctx(self, allow_unknown_ca=False, req_peer_cert=True, session=None): 
152          ctx = SSL.Context("tlsv1") 
153          # Set certificate and private key 
154          m2.ssl_ctx_use_x509(ctx.ctx, self.cert.x509) 
155          m2.ssl_ctx_use_rsa_privkey(ctx.ctx, self.rsakey.rsa) 
156          if not m2.ssl_ctx_check_privkey(ctx.ctx): 
157              raise CryptoError('public/private key mismatch') 
158          # Ciphers/Options 
159          ctx.set_cipher_list(CIPHER_SET) 
160          ctx.set_options(CTX_OPTIONS) 
161          # CA settings 
162          cloc = os.path.join(global_certpath, 'cacert.root.pem') 
163          if ctx.load_verify_locations(cafile=cloc) != 1: 
164              log.error("Problem loading CA certificates") 
165              raise CryptoError('CA certificates not loaded') 
166          # Verification 
167          cb = mk_verify_cb(allow_unknown_ca=allow_unknown_ca) 
168          CTX_V_FLAGS = SSL.verify_peer 
169          if req_peer_cert: 
170              CTX_V_FLAGS |= SSL.verify_fail_if_no_peer_cert 
171          ctx.set_verify(CTX_V_FLAGS,3,cb) 
172          # Session 
173          if session: 
174              ctx.set_session_id_ctx(session) 
175          return ctx 
176   
177 -    def get_pub(self): 
178          return self.rsakey.pub()[1] 
179   
180 -    def fingerprint(self): 
181          return hashlib.sha1(self.get_pub()).hexdigest() 
182   
183 -    def decrypt(self, data, returnpad=False): 
184          """ 
185          Decrypts data encrypted with this public key 
186   
187          @param data: The data, padding and all, to be decrypted 
188          @type data: string 
189          @param returnpad: return "junk" decrypted padding as well as message. Default: False 
190          @type returnpad: boolean 
191   
192          @raise CryptoError: Priv. decrypt fail or Bad Checksum 
193   
194          @return: tuple (decrypted message, padding) if returnpad is true, string otherwise 
195          @rtype: tuple 
196          """ 
197          byte_key_size = len(self.rsakey)/8 
198          # Decrypt the session key and IV with our private key 
199          try: 
200              tmpsk = self.rsakey.private_decrypt(data[:byte_key_size], RSA.pkcs1_oaep_padding) 
201          except RSA.RSAError, e: 
202              raise CryptoError("A decryption error occurred: %s" % str(e)) 
203          sk = tmpsk[:32] # Session Key 
204          iv = tmpsk[32:] # IV 
205          sessionkey = Anomos.Crypto.AESKey(sk, iv) 
206          # Decrypt the rest of the message with the session key 
207          content = sessionkey.decrypt(data[byte_key_size:]) 
208          #pos = sha.digestsize 
209          pos = 20 
210          givenchksum = content[:pos] # first 20 bytes 
211          smsglen = content[pos:pos+4] # next 4 bytes 
212          imsglen = toint(smsglen) 
213          pos += 4 
214          message = content[pos:pos+imsglen] 
215          pos += imsglen 
216          mychksum = hashlib.sha1(sk+smsglen+message).digest() 
217          if givenchksum != mychksum: 
218              raise CryptoError("Bad Checksum - Data may have been tampered with") 
219          if returnpad: 
220              return (message, content[pos:]) 
221          else: 
222              return message 
223