ctypescrypto/x509.py

   1 from ctypes import c_void_p,create_string_buffer,c_long,c_int,POINTER,c_char_p
   2 from ctypescrypto.bio import Membio
   3 from ctypescrypto.pkey import PKey
   4 from ctypescrypto.oid import Oid
   5 from ctypescrypto.exception import LibCryptoError
   6 from ctypescrypto import libcrypto
   7 class X509Error(LibCryptoError):
   8         """
   9         Exception, generated when some openssl function fail
  10         during X509 operation
  11         """
  12         pass
  13
  14
  15 class X509Name:
  16         """
  17         Class which represents X.509 distinguished name - typically
  18         a certificate subject name or an issuer name.
  19         """
  20         # XN_FLAG_SEP_COMMA_PLUS & ASN1_STRFLG_UTF8_CONVERT
  21         PRINT_FLAG=0x10010
  22         ESC_MSB=4
  23         def __init__(self,ptr=None,copy=False):
  24                 """
  25                 Creates a X509Name object
  26                 @param ptr - pointer to X509_NAME C structure (as returned by some  OpenSSL functions
  27                 @param copy - indicates that this structure have to be freed upon object destruction
  28                 """
  29                 if ptr is not None:
  30                         self.ptr=ptr
  31                         self.need_free=copy
  32                         self.writable=False
  33                 else:
  34                         self.ptr=libcrypto.X509_NAME_new()
  35                         self.need_free=True
  36                         self.writable=True
  37         def __del__(self):
  38                 """
  39                 Frees if neccessary
  40                 """
  41                 if self.need_free:
  42                         libcrypto.X509_NAME_free(self.ptr)
  43         def __str__(self):
  44                 """
  45                 Produces an ascii representation of the name, escaping all symbols > 0x80
  46                 Probably it is not what you want, unless your native language is English
  47                 """
  48                 b=Membio()
  49                 libcrypto.X509_NAME_print_ex(b.bio,self.ptr,0,self.PRINT_FLAG | self.ESC_MSB)
  50                 return str(b)
  51         def __unicode__(self):
  52                 """
  53                 Produces unicode representation of the name.
  54                 """
  55                 b=Membio()
  56                 libcrypto.X509_NAME_print_ex(b.bio,self.ptr,0,self.PRINT_FLAG)
  57                 return unicode(b)
  58         def __len__(self):
  59                 """
  60                 return number of components in the name
  61                 """
  62                 return libcrypto.X509_NAME_entry_count(self.ptr)
  63         def __cmp__(self,other):
  64                 """
  65                 Compares X509 names
  66                 """
  67                 return libcrypto.X509_NAME_cmp(self.ptr,other.ptr)
  68         def __eq__(self,other):
  69                 return libcrypto.X509_NAME_cmp(self.ptr,other.ptr)==0
  70
  71         def __getitem__(self,key):
  72                 if isinstance(key,Oid):
  73                         # Return first matching field
  74                         idx=libcrypto.X509_NAME_get_index_by_NID(self.ptr,key.nid,-1)
  75                         if idx<0:
  76                                 raise KeyError("Key not found "+repr(Oid))
  77                         entry=libcrypto.X509_NAME_get_entry(self.ptr,idx)
  78                         s=libcrypto.X509_NAME_ENTRY_get_data(entry)
  79                         b=Membio()
  80                         libcrypto.ASN1_STRING_print_ex(b.bio,s,self.PRINT_FLAG)
  81                         return unicode(b)
  82                 elif isinstance(key,int):
  83                         # Return OID, string tuple
  84                         entry=libcrypto.X509_NAME_get_entry(self.ptr,key)
  85                         if entry is None:
  86                                 raise IndexError("name entry index out of range")
  87                         obj=libcrypto.X509_NAME_ENTRY_get_object(entry)
  88                         nid=libcrypto.OBJ_obj2nid(obj)
  89                         if nid==0:
  90                                 buf=create_string_buffer(80)
  91                                 len=libcrypto.OBJ_obj2txt(buf,80,obj,1)
  92                                 oid=Oid(buf[0:len])
  93                         else:
  94                                 oid=Oid(nid)
  95                         s=libcrypto.X509_NAME_ENTRY_get_data(entry)
  96                         b=Membio()
  97                         libcrypto.ASN1_STRING_print_ex(b.bio,s,self.PRINT_FLAG)
  98                         return (oid,unicode(b))
  99
 100         def __setitem__(self,key,val):
 101                 if not self.writable:
 102                         raise ValueError("Attempt to modify constant X509 object")
 103 class X509_extlist:
 104         def __init__(self,ptr):
 105                 self.ptr=ptr
 106         def __del__(self):
 107                 libcrypto.X509_NAME_free(self.ptr)
 108         def __str__(self):
 109                 raise NotImplementedError
 110         def __len__(self):
 111                 return libcrypto.X509_NAME_entry_count(self.ptr)
 112
 113         def __getattr__(self,key):
 114                 raise NotImplementedError
 115         def __setattr__(self,key,val):
 116                 raise NotImplementedError
 117
 118
 119
 120
 121 class X509:
 122         """
 123         Represents X.509 certificate.
 124         """
 125         def __init__(self,data=None,ptr=None,format="PEM"):
 126                 """
 127                 Initializes certificate
 128                 @param data - serialized certificate in PEM or DER format.
 129                 @param ptr - pointer to X509, returned by some openssl function.
 130                         mutually exclusive with data
 131                 @param format - specifies data format. "PEM" or "DER", default PEM
 132                 """
 133                 if ptr is not None:
 134                         if data is not None:
 135                                 raise TypeError("Cannot use data and ptr simultaneously")
 136                         self.cert = ptr
 137                 elif data is None:
 138                         raise TypeError("data argument is required")
 139                 else:
 140                         b=Membio(data)
 141                         if format == "PEM":
 142                                 self.cert=libcrypto.PEM_read_bio_X509(b.bio,None,None,None)
 143                         else:
 144                                 self.cert=libcrypto.d2i_X509_bio(b.bio,None)
 145                         if self.cert is None:
 146                                 raise X509Error("error reading certificate")
 147         def __del__(self):
 148                 """
 149                 Frees certificate object
 150                 """
 151                 libcrypto.X509_free(self.cert)
 152         def __str__(self):
 153                 """ Returns der string of the certificate """
 154                 b=Membio()
 155                 if libcrypto.i2d_X509_bio(b.bio,self.cert)==0:
 156                         raise X509Error("error serializing certificate")
 157                 return str(b)
 158         def __repr__(self):
 159                 """ Returns valid call to the constructor """
 160                 return "X509(data="+repr(str(self))+",format='DER')"
 161         @property
 162         def pubkey(self):
 163                 """EVP PKEy object of certificate public key"""
 164                 return PKey(ptr=libcrypto.X509_get_pubkey(self.cert,False))
 165         def verify(self,store=None,key=None):
 166                 """
 167                 Verify self. Supports verification on both X509 store object
 168                 or just public issuer key
 169                 @param store X509Store object.
 170                 @param key - PKey object
 171                 parameters are mutually exclusive. If neither is specified, attempts to verify
 172                 itself as self-signed certificate
 173                 """
 174                 if store is not None and key is not None:
 175                         raise X509Error("key and store cannot be specified simultaneously")
 176                 if store is not None:
 177                         ctx=libcrypto.X509_STORE_CTX_new()
 178                         if ctx is None:
 179                                 raise X509Error("Error allocating X509_STORE_CTX")
 180                         if libcrypto.X509_STORE_CTX_init(ctx,store.store,self.cert,None) < 0:
 181                                 raise X509Error("Error allocating X509_STORE_CTX")
 182                         res= libcrypto.X509_verify_cert(ctx)
 183                         libcrypto.X509_STORE_CTX_free(ctx)
 184                         return res>0
 185                 else:
 186                         if key is None:
 187                                 if self.issuer != self.subject:
 188                                         # Not a self-signed certificate
 189                                         return False
 190                                 key = self.pubkey
 191                         res = libcrypto.X509_verify(self.cert,key.key)
 192                         if res < 0:
 193                                 raise X509Error("X509_verify failed")
 194                         return res>0
 195
 196         @property
 197         def subject(self):
 198                 """ X509Name for certificate subject name """
 199                 return X509Name(libcrypto.X509_get_subject_name(self.cert))
 200         @property
 201         def issuer(self):
 202                 """ X509Name for certificate issuer name """
 203                 return X509Name(libcrypto.X509_get_issuer_name(self.cert))
 204         @property
 205         def serial(self):
 206                 """ Serial number of certificate as integer """
 207                 asnint=libcrypto.X509_get_serialNumber(self.cert)
 208                 b=Membio()
 209                 libcrypto.i2a_ASN1_INTEGER(b.bio,asnint)
 210                 return int(str(b),16)
 211         @property
 212         def startDate(self):
 213                 """ Certificate validity period start date """
 214                 # Need deep poke into certificate structure (x)->cert_info->validity->notBefore
 215                 raise NotImplementedError
 216         @property
 217         def endDate(self):
 218                 """ Certificate validity period end date """
 219                 # Need deep poke into certificate structure (x)->cert_info->validity->notAfter
 220                 raise NotImplementedError
 221         def extensions(self):
 222                 """ Returns list of extensions """
 223                 raise NotImplementedError
 224         def check_ca(self):
 225                 """ Returns True if certificate is CA certificate """
 226                 return libcrypto.X509_check_ca(self.cert)>0
 227 class X509Store:
 228         """
 229                 Represents trusted certificate store. Can be used to lookup CA certificates to verify
 230
 231                 @param file - file with several certificates and crls to load into store
 232                 @param dir - hashed directory with certificates and crls
 233                 @param default - if true, default verify location (directory) is installed
 234
 235         """
 236         def __init__(self,file=None,dir=None,default=False):
 237                 """
 238                 Creates X509 store and installs lookup method. Optionally initializes
 239                 by certificates from given file or directory.
 240                 """
 241                 #
 242                 # Todo - set verification flags
 243                 #
 244                 self.store=libcrypto.X509_STORE_new()
 245                 if self.store is None:
 246                         raise X509Error("allocating store")
 247                 lookup=libcrypto.X509_STORE_add_lookup(self.store,libcrypto.X509_LOOKUP_file())
 248                 if lookup is None:
 249                         raise X509Error("error installing file lookup method")
 250                 if (file is not None):
 251                         if not libcrypto.X509_LOOKUP_ctrl(lookup,1,file,1,None)>0:
 252                                 raise X509Error("error loading trusted certs from file "+file)
 253                 lookup=libcrypto.X509_STORE_add_lookup(self.store,libcrypto.X509_LOOKUP_hash_dir())
 254                 if lookup is None:
 255                         raise X509Error("error installing hashed lookup method")
 256                 if dir is not None:
 257                         if not libcrypto.X509_LOOKUP_ctrl(lookup,2,dir,1,None)>0:
 258                                 raise X509Error("error adding hashed  trusted certs dir "+dir)
 259                 if default:
 260                         if not libcrypto.X509_LOOKUP_ctrl(lookup,2,None,3,None)>0:
 261                                 raise X509Error("error adding default trusted certs dir ")
 262         def add_cert(self,cert):
 263                 """
 264                 Explicitely adds certificate to set of trusted in the store
 265                 @param cert - X509 object to add
 266                 """
 267                 if not isinstance(cert,X509):
 268                         raise TypeError("cert should be X509")
 269                 libcrypto.X509_STORE_add_cert(self.store,cert.cert)
 270         def add_callback(self,callback):
 271                 """
 272                 Installs callbac function, which would receive detailed information
 273                 about verified ceritificates
 274                 """
 275                 raise NotImplementedError
 276         def setflags(self,flags):
 277                 """
 278                 Set certificate verification flags.
 279                 @param flags - integer bit mask. See OpenSSL X509_V_FLAG_* constants
 280                 """
 281                 libcrypto.X509_STORE_set_flags(self.store,flags)
 282         def setpurpose(self,purpose):
 283                 """
 284                 Sets certificate purpose which verified certificate should match
 285                 @param purpose - number from 1 to 9 or standard strind defined in Openssl
 286                 possible strings - sslcient,sslserver, nssslserver, smimesign,smimeencrypt, crlsign, any,ocsphelper
 287                 """
 288                 if isinstance(purpose,str):
 289                         purp_no=X509_PURPOSE_get_by_sname(purpose)
 290                         if purp_no <=0:
 291                                 raise X509Error("Invalid certificate purpose '"+purpose+"'")
 292                 elif isinstance(purpose,int):
 293                         purp_no = purpose
 294                 if libcrypto.X509_STORE_set_purpose(self.store,purp_no)<=0:
 295                         raise X509Error("cannot set purpose")
 296 libcrypto.i2a_ASN1_INTEGER.argtypes=(c_void_p,c_void_p)
 297 libcrypto.ASN1_STRING_print_ex.argtypes=(c_void_p,c_void_p,c_long)
 298 libcrypto.X509_get_serialNumber.argtypes=(c_void_p,)
 299 libcrypto.X509_get_serialNumber.restype=c_void_p
 300 libcrypto.X509_NAME_ENTRY_get_object.restype=c_void_p
 301 libcrypto.X509_NAME_ENTRY_get_object.argtypes=(c_void_p,)
 302 libcrypto.OBJ_obj2nid.argtypes=(c_void_p,)
 303 libcrypto.X509_NAME_get_entry.restype=c_void_p
 304 libcrypto.X509_NAME_get_entry.argtypes=(c_void_p,c_int)
 305 libcrypto.X509_STORE_new.restype=c_void_p
 306 libcrypto.X509_STORE_add_lookup.restype=c_void_p
 307 libcrypto.X509_STORE_add_lookup.argtypes=(c_void_p,c_void_p)
 308 libcrypto.X509_LOOKUP_file.restype=c_void_p
 309 libcrypto.X509_LOOKUP_hash_dir.restype=c_void_p
 310 libcrypto.X509_LOOKUP_ctrl.restype=c_int
 311 libcrypto.X509_LOOKUP_ctrl.argtypes=(c_void_p,c_int,c_char_p,c_long,POINTER(c_char_p))