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