source: TI12-security/trunk/NDGSecurity/python/ndg_security_common/ndg/security/common/X509.py @ 7155

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/TI12-security/trunk/NDGSecurity/python/ndg_security_common/ndg/security/common/X509.py@7155
Revision 7155, 33.8 KB checked in by pjkersha, 9 years ago (diff)

Incomplete - task 2: XACML-Security Integration

  • migrating to ndg.saml and ndg.soap imports now that SAML WSGI middleware has moved to ndg.saml egg.
  • Property svn:eol-style set to native
  • Property svn:keywords set to Id
Line 
1"""X.509 certificate handling class encapsulates M2Crypto.X509
2
3NERC Data Grid Project
4"""
5__author__ = "P J Kershaw"
6__date__ = "05/04/05"
7__copyright__ = "(C) 2009 Science and Technology Facilities Council"
8__license__ = "BSD - See LICENSE file in the top-level directory"
9__contact__ = "Philip.Kershaw@stfc.ac.uk"
10__revision__ = '$Id$'
11import logging
12log = logging.getLogger(__name__)
13from warnings import warn # warn of impending certificate expiry
14
15import types
16import re
17import os
18import errno
19
20# Handle not before and not after strings
21from time import strptime
22from datetime import datetime
23
24import M2Crypto
25
26
27class X509CertError(Exception):
28    """Exception handling for NDG X.509 Certificate handling class."""
29
30class X509CertReadError(X509CertError):
31    """Error reading in certificate from file"""
32
33class X509CertParseError(X509CertError):
34    """Error parsing a certificate"""
35 
36class X509CertInvalidNotBeforeTime(X509CertError):
37    """Call from X509Cert.isValidTime if certificates not before time is
38    BEFORE the current system time"""
39   
40class X509CertExpired(X509CertError):
41    """Call from X509Cert.isValidTime if certificate has expired"""
42
43   
44class X509Cert(object):
45    "NDG X509 Certificate Handling"
46
47    formatPEM = M2Crypto.X509.FORMAT_PEM
48    formatDER = M2Crypto.X509.FORMAT_DER
49   
50    def __init__(self, filePath=None, m2CryptoX509=None):
51
52        # Set certificate file path
53        if filePath is not None:
54            if not isinstance(filePath, basestring):
55                raise X509CertError("Certificate File Path input must be a "
56                                    "valid string")
57           
58        self.__filePath = filePath           
59        self.__dn = None
60        self.__dtNotBefore = None
61        self.__dtNotAfter = None
62       
63        if m2CryptoX509:
64            self.__setM2CryptoX509(m2CryptoX509)
65        else:
66            self.__m2CryptoX509 = None
67
68    def read(self, 
69             filePath=None, 
70             format=None, 
71             warningStackLevel=3,
72             **isValidTimeKw):
73        """Read a certificate from PEM encoded DER format file
74       
75        @type filePath: basestring
76        @param filePath: file path of PEM format file to be read
77       
78        @type format: int
79        @param format: format of input file - PEM is the default.  Set to
80        X509Cert.formatDER for DER format
81       
82        @type isValidTimeKw: dict
83        @param isValidTimeKw: keywords to isValidTime() call"""
84
85        if format is None:
86            format = X509Cert.formatPEM
87       
88        # Check for optional input certificate file path
89        if filePath is not None:
90            if not isinstance(filePath, basestring):
91                raise X509CertError("Certificate File Path input must be a "
92                                    "valid string")
93           
94            self.__filePath = filePath
95       
96        try:
97            self.__m2CryptoX509 = M2Crypto.X509.load_cert(self.__filePath,
98                                                          format=format)
99        except Exception, e:
100            raise X509CertReadError("Error loading certificate \"%s\": %s" %
101                                    (self.__filePath, e))
102
103        # Update DN and validity times from M2Crypto X509 object just
104        # created
105        self.__setM2CryptoX509()
106       
107        self.isValidTime(warningStackLevel=warningStackLevel, **isValidTimeKw)
108
109    def parse(self, 
110              certTxt, 
111              format=None, 
112              warningStackLevel=3,
113              **isValidTimeKw):
114        """Read a certificate input as a string
115       
116        @type certTxt: basestring
117        @param certTxt: PEM encoded certificate to parse
118       
119        @type format: int
120        @param format: format of input file - PEM is the default.  Set to
121        X509Cert.formatDER for DER format
122       
123        @type isValidTimeKw: dict
124        @param isValidTimeKw: keywords to isValidTime() call"""
125
126        if format is None:
127            format = X509Cert.formatPEM
128           
129        try:
130            # Create M2Crypto memory buffer and pass to load certificate
131            # method
132            #
133            # Nb. input converted to standard string - buffer method won't
134            # accept unicode type strings
135#            certBIO = M2Crypto.BIO.MemoryBuffer(str(certTxt))
136#            self.__m2CryptoX509 = M2Crypto.X509.load_cert_bio(certBIO)
137            self.__m2CryptoX509 = M2Crypto.X509.load_cert_string(str(certTxt),
138                                                                 format=format)
139        except Exception, e:
140            raise X509CertParseError("Error loading certificate: %s" % e)
141
142        # Update DN and validity times from M2Crypto X509 object just
143        # created
144        self.__setM2CryptoX509()
145       
146        self.isValidTime(warningStackLevel=warningStackLevel, **isValidTimeKw)
147     
148    def __setM2CryptoX509(self, m2CryptoX509=None):
149        """Private method allows class members to be updated from the
150        current M2Crypto object.  __m2CryptoX509 must have been set."""
151       
152        if m2CryptoX509 is not None:
153            if not isinstance(m2CryptoX509, M2Crypto.X509.X509):
154                raise TypeError("Incorrect type for input M2Crypto.X509.X509 "
155                                "object")
156                   
157            self.__m2CryptoX509 = m2CryptoX509
158                   
159        # Get distinguished name
160        m2CryptoX509Name = self.__m2CryptoX509.get_subject()
161
162        # Instantiate X500 Distinguished name
163        self.__dn = X500DN(m2CryptoX509Name=m2CryptoX509Name)
164       
165        # Get not before and not after validity times
166        #
167        # Only option for M2Crypto seems to be to return the times as
168        # formatted strings and then parse them in order to create a datetime
169        # type
170       
171        try:
172            m2CryptoNotBefore = self.__m2CryptoX509.get_not_before()
173            self.__dtNotBefore=self.__m2CryptoUTC2datetime(m2CryptoNotBefore)
174                                       
175        except Exception, e:
176            raise X509CertError("Not Before time: %s" % e)
177
178        try:
179            m2CryptoNotAfter = self.__m2CryptoX509.get_not_after()
180            self.__dtNotAfter = self.__m2CryptoUTC2datetime(m2CryptoNotAfter)
181                                   
182        except Exception, e:
183            raise X509CertError("Not After time: %s" % e)
184
185    def __getM2CryptoX509(self, m2CryptoX509=None):
186        "Return M2Crypto X.509 cert object"
187        return self.__m2CryptoX509
188   
189    m2CryptoX509 = property(fset=__setM2CryptoX509,
190                            fget=__getM2CryptoX509,
191                            doc="M2Crypto.X509.X509 type")
192       
193    def toString(self, **kw):
194        """Return certificate file content as a PEM format
195        string"""
196        return self.asPEM(**kw)
197       
198    def asPEM(self, filePath=None):
199        """Return certificate file content as a PEM format
200        string"""
201       
202        # Check M2Crypto.X509 object has been instantiated - if not call
203        # read method
204        if self.__m2CryptoX509 is None:
205            self.read(filePath)
206           
207        return self.__m2CryptoX509.as_pem()
208
209    def asDER(self):
210        """Return certificate file content in DER format"""
211       
212        # Check M2Crypto.X509 object has been instantiated
213        assert(self.__m2CryptoX509)
214        return self.__m2CryptoX509.as_der()
215
216    # Make some attributes accessible as read-only
217    def __getDN(self):
218        """Get X500 Distinguished Name."""
219        return self.__dn
220
221    dn = property(fget=__getDN, doc="X.509 Distinguished Name")
222   
223    def __getVersion(self):
224        """Get X.509 Certificate version"""
225        if self.__m2CryptoX509 is None:
226            return None
227       
228        return self.__m2CryptoX509.get_version()
229
230    version = property(fget=__getVersion, doc="X.509 Certificate version")
231       
232    def __getSerialNumber(self):
233        """Get Serial Number"""
234        if self.__m2CryptoX509 is None:
235            return None
236       
237        return self.__m2CryptoX509.get_serial_number()
238   
239    serialNumber = property(fget=__getSerialNumber, 
240                            doc="X.509 Certificate Serial Number")
241
242    def __getNotBefore(self):
243        """Get not before validity time as datetime type"""
244        if self.__m2CryptoX509 is None:
245            return None
246       
247        return self.__dtNotBefore
248
249    notBefore = property(fget=__getNotBefore, 
250                         doc="Not before validity time as datetime type")
251       
252    def __getNotAfter(self):
253        """Get not after validity time as datetime type"""
254        if self.__m2CryptoX509 is None:
255            return None
256       
257        return self.__dtNotAfter
258
259    notAfter = property(fget=__getNotAfter, 
260                         doc="Not after validity time as datetime type")
261       
262    def __getPubKey(self):
263        """Get public key
264       
265        @return: RSA public key for certificate
266        @rtype: M2Crypto.RSA.RSA_pub"""
267        if self.__m2CryptoX509 is None:
268            return None
269       
270        return self.__m2CryptoX509.get_pubkey()
271
272    pubKey = property(fget=__getPubKey, doc="Public Key")
273       
274    def __getIssuer(self):
275        """Get Certificate issuer"""
276        if self.__m2CryptoX509 is None:
277            return None
278       
279        # Return as X500DN type
280        return X500DN(m2CryptoX509Name=self.__m2CryptoX509.get_issuer())
281
282    issuer = property(fget=__getIssuer, doc="Certificate Issuer")
283   
284    def __getSubject(self):
285        """Get Certificate subject"""
286        if self.__m2CryptoX509 is None:
287            return None
288
289        # Return as X500DN type
290        return X500DN(m2CryptoX509Name=self.__m2CryptoX509.get_subject())
291   
292    subject = property(fget=__getSubject, doc="Certificate subject")
293
294    def isValidTime(self, 
295                    raiseExcep=False, 
296                    expiryWarning=True, 
297                    nDaysBeforeExpiryLimit=30,
298                    warningStackLevel=2):
299        """Check Certificate for expiry
300
301        @type raiseExcep: bool
302        @param raiseExcep: set True to raise an exception if certificate is
303        invalid
304       
305        @type expiryWarning: bool
306        @param expiryWarning: set to True to output a warning message if the
307        certificate is due to expire in less than nDaysBeforeExpiryLimit days.
308        Message is sent using warnings.warn and through logging.warning.  No
309        message is set if the certificate has an otherwise invalid time
310       
311        @type nDaysBeforeExpiryLimit: int
312        @param nDaysBeforeExpiryLimit: used in conjunction with the
313        expiryWarning flag.  Set the number of days in advance of certificate
314        expiry from which to start outputing warnings
315       
316        @type warningStackLevel: int
317        @param warningStackLevel: set where in the stack to flag the warning
318        from.  Level 2 will flag it at the level of the caller of this
319        method.  Level 3 would flag at the level of the caller of the caller
320        and so on.
321       
322        @raise X509CertInvalidNotBeforeTime: current time is before the
323        certificate's notBefore time
324        @raise X509CertExpired: current time is after the certificate's
325        notAfter time"""
326
327        if not isinstance(self.__dtNotBefore, datetime):
328            raise X509CertError("Not Before datetime is not set")
329
330        if not isinstance(self.__dtNotAfter, datetime):
331            raise X509CertError("Not After datetime is not set")
332       
333        dtNow = datetime.utcnow()
334        isValidTime = dtNow > self.__dtNotBefore and dtNow < self.__dtNotAfter
335
336        # Helper string for message output
337        if self.__filePath:
338            fileInfo = ' "%s"' % self.__filePath
339        else:
340            fileInfo = ''
341             
342       
343        # Set a warning message for impending expiry of certificate but only
344        # if the certificate is not any other way invalid - see below
345        if isValidTime and expiryWarning:
346            dtTime2Expiry = self.__dtNotAfter - dtNow
347            if dtTime2Expiry.days < nDaysBeforeExpiryLimit:
348                msg = ('Certificate%s with DN "%s" will expire in %d days on: '
349                       '%s' % (fileInfo, 
350                               self.dn, 
351                               dtTime2Expiry.days, 
352                               self.__dtNotAfter))
353                warn(msg, stacklevel=warningStackLevel)
354                log.warning(msg)
355       
356                     
357        if dtNow < self.__dtNotBefore:
358            msg = ("Current time %s is before the certificate's Not Before "
359                   'Time %s for certificate%s with DN "%s"' % 
360                   (dtNow, self.__dtNotBefore, fileInfo, self.dn))
361            log.error(msg)
362            if raiseExcep:
363                raise X509CertInvalidNotBeforeTime(msg)
364           
365        elif dtNow > self.__dtNotAfter:
366            msg = ('Certificate%s with DN "%s" has expired: the time now is '
367                   '%s and the certificate expiry is %s.' %(fileInfo,
368                                                            self.dn, 
369                                                            dtNow, 
370                                                            self.__dtNotAfter))
371            log.error(msg)
372            if raiseExcep:
373                raise X509CertExpired(msg)
374
375        # If exception flag is not set return validity as bool
376        return isValidTime
377
378
379
380
381    def __m2CryptoUTC2datetime(self, m2CryptoUTC):
382        """Convert M2Crypto UTC time string as returned by get_not_before/
383        get_not_after methods into datetime type"""
384       
385        datetimeRE = "([a-zA-Z]{3} {1,2}\d{1,2} \d{2}:\d{2}:\d{2} \d{4}).*"
386        sM2CryptoUTC = None
387       
388        try:
389            # Convert into string
390            sM2CryptoUTC = str(m2CryptoUTC)
391           
392            # Check for expected format - string may have trailing GMT - ignore
393            sTime = re.findall(datetimeRE, sM2CryptoUTC)[0]
394
395            # Convert into a tuple
396            lTime = strptime(sTime, "%b %d %H:%M:%S %Y")[0:6]
397
398            return datetime(lTime[0], lTime[1], lTime[2],
399                            lTime[3], lTime[4], lTime[5])
400                                   
401        except Exception, e:
402            msg = "Error parsing M2Crypto UTC"
403            if sM2CryptoUTC is not None:
404                msg += ": " + sM2CryptoUTC
405               
406            raise X509CertError(msg)
407       
408    def verify(self, pubKey, **kw):
409        """Verify a certificate against the public key of the
410        issuer
411       
412        @param pubKey: public key of cert that issued self
413        @type pubKey: M2Crypto.RSA.RSA_pub
414        @param **kw: keywords to pass to M2Crypto.X509.X509 -
415        'pkey'
416        @type: dict
417        @return: True if verifies OK, False otherwise
418        @rtype: bool
419        """
420        return bool(self.__m2CryptoX509.verify(pubKey, **kw))
421
422    @classmethod
423    def Read(cls, filePath, warningStackLevel=4, **isValidTimeKw):
424        """Create a new X509 certificate read in from a file"""
425        x509Cert = cls(filePath=filePath) 
426        x509Cert.read(warningStackLevel=warningStackLevel, **isValidTimeKw)
427       
428        return x509Cert
429   
430    @classmethod
431    def Parse(cls, x509CertTxt, warningStackLevel=4, **isValidTimeKw):
432        """Create a new X509 certificate from string of file content"""
433        x509Cert = cls()     
434        x509Cert.parse(x509CertTxt, 
435                       warningStackLevel=warningStackLevel,
436                       **isValidTimeKw)
437       
438        return x509Cert
439
440    @classmethod
441    def fromM2Crypto(cls, m2CryptoX509):
442        """Convenience method to instantiate a new object from an M2Crypto
443        X.509 certificate object"""
444        x509Cert = cls(m2CryptoX509=m2CryptoX509)
445        return x509Cert
446   
447# Alternative AttCert constructors
448def X509CertRead(filePath, warningStackLevel=4, **isValidTimeKw):
449    """Create a new X509 certificate read in from a file"""
450
451    x509Cert = X509Cert(filePath=filePath)   
452    x509Cert.read(warningStackLevel=warningStackLevel, **isValidTimeKw)
453   
454    return x509Cert
455
456def X509CertParse(x509CertTxt, warningStackLevel=4, **isValidTimeKw):
457    """Create a new X509 certificate from string of file content"""
458
459    x509Cert = X509Cert()
460    x509Cert.parse(x509CertTxt, 
461                   warningStackLevel=warningStackLevel, 
462                   **isValidTimeKw)
463   
464    return x509Cert
465
466
467class X509StackError(X509CertError):
468    """Error from X509Stack type"""
469
470
471class X509StackEmptyError(X509CertError):
472    """Expecting non-zero length X509Stack"""
473
474
475class X509CertIssuerNotFound(X509CertError):
476    """Raise from verifyCertChain if no certificate can be found to verify the
477    input"""
478
479
480class SelfSignedCert(X509CertError):
481    """Raise from verifyCertChain if cert. is self-signed and
482    rejectSelfSignedCert=True"""
483
484
485class X509CertInvalidSignature(X509CertError):
486    """X.509 Certificate has an invalid signature"""
487       
488       
489class X509Stack(object):
490    """Wrapper for M2Crypto X509_Stack"""
491   
492    def __init__(self, m2X509Stack=None):
493        """Initialise from an M2Crypto stack object
494       
495        @param m2X509Stack: M2Crypto X.509 stack object
496        @type m2X509Stack: M2Crypto.X509.X509_Stack"""
497       
498        self.__m2X509Stack = m2X509Stack or M2Crypto.X509.X509_Stack()
499       
500    def __len__(self):
501        """@return: length of stack
502        @rtype: int"""
503        return self.__m2X509Stack.__len__()
504
505    def __getitem__(self, idx):
506        """Index stack as an array
507        @param idx: stack index
508        @type idx: int
509        @return: X.509 cert object
510        @rtype: ndg.security.common.X509.X509Cert"""
511       
512        return X509Cert(m2CryptoX509=self.__m2X509Stack.__getitem__(idx))
513   
514    def __iter__(self):
515        """@return: stack iterator
516        @rtype: listiterator"""
517        return iter([X509Cert(m2CryptoX509=i) for i in self.__m2X509Stack])
518
519    def push(self, x509Cert):
520        """Push an X509 certificate onto the stack.
521       
522        @param x509Cert: X509 object.
523        @type x509Cert: M2Crypto.X509.X509,
524        ndg.security.common.X509.X509Cert or basestring
525        @return: The number of X509 objects currently on the stack.
526        @rtype: int"""
527        if isinstance(x509Cert, M2Crypto.X509.X509):
528            return self.__m2X509Stack.push(x509Cert)
529       
530        elif isinstance(x509Cert, X509Cert):
531            return self.__m2X509Stack.push(x509Cert.m2CryptoX509)
532       
533        elif isinstance(x509Cert, basestring):
534            return self.__m2X509Stack.push(\
535                                       X509CertParse(x509Cert).m2CryptoX509)           
536        else:
537            raise X509StackError("Expecting M2Crypto.X509.X509, ndg.security."
538                                 "common.X509.X509Cert or string type")
539               
540    def pop(self):
541        """Pop a certificate from the stack.
542       
543        @return: X509 object that was popped, or None if there is nothing
544        to pop.
545        @rtype: ndg.security.common.X509.X509Cert
546        """
547        return X509Cert(m2CryptoX509=self.__m2X509Stack.pop())
548
549    def asDER(self):
550        """Return the stack as a DER encoded string
551        @return: DER string
552        @rtype: string"""
553        return self.__m2X509Stack.as_der()
554
555    def verifyCertChain(self, 
556                        x509Cert2Verify=None, 
557                        caX509Stack=[],
558                        rejectSelfSignedCert=True):
559        """Treat stack as a list of certificates in a chain of
560        trust.  Validate the signatures through to a single root issuer. 
561
562        @param x509Cert2Verify: X.509 certificate to be verified default is
563        last in the stack
564        @type x509Cert2Verify: X509Cert
565       
566        @param caX509Stack: X.509 stack containing CA certificates that are
567        trusted.
568        @type caX509Stack: X509Stack
569       
570        @param rejectSelfSignedCert: Set to True (default) to raise an
571        SelfSignedCert exception if a certificate in self's stack is
572        self-signed. 
573        @type rejectSelfSignedCert: bool"""
574       
575        n2Validate = len(self)
576        if x509Cert2Verify:
577            # One more to validate in addition to stack content
578            n2Validate += 1
579        else:
580            # Validate starting from last on stack - but check first that it's
581            # populated
582            if n2Validate == 0:
583                raise X509StackEmptyError("Empty stack and no x509Cert2Verify "
584                                          "set: no cert.s to verify")
585
586            x509Cert2Verify = self[-1]
587             
588        # Exit loop if all certs have been validated or if find a self
589        # signed cert.
590        nValidated = 0
591        issuerX509Cert = None
592        while nValidated < n2Validate:
593            issuerX509Cert = None
594            issuerDN = x509Cert2Verify.issuer
595           
596            # Search for issuing certificate in stack
597            for x509Cert in self:
598                if x509Cert.dn == issuerDN:
599                    # Match found - the cert.'s issuer has been found in the
600                    # stack
601                    issuerX509Cert = x509Cert
602                    break
603                   
604            if issuerX509Cert:
605                # An issuing cert. has been found - use it to check the
606                # signature of the cert. to be verified
607                if not x509Cert2Verify.verify(issuerX509Cert.pubKey):
608                    X509CertInvalidSignature('Signature is invalid for cert. '
609                                             '"%s"' % x509Cert2Verify.dn)
610               
611                # In the next iteration the issuer cert. will be checked:
612                # 1) search for a cert. in the stack that issued it
613                # 2) If found use the issuing cert. to verify
614                x509Cert2Verify = issuerX509Cert
615                nValidated += 1
616            else:
617                # All certs in the stack have been searched
618                break
619
620
621        if issuerX509Cert:           
622            # Check for self-signed certificate
623            if (nValidated == 1 and rejectSelfSignedCert and 
624                issuerX509Cert.dn == issuerX509Cert.issuer):
625
626                # If only one iteration occurred then it must be a self
627                # signed certificate
628                raise SelfSignedCert("Certificate is self signed: [DN=%s]" %
629                                     issuerX509Cert.dn)
630           
631            if not caX509Stack:
632                caX509Stack = [issuerX509Cert]
633                         
634        elif not caX509Stack:
635            raise X509CertIssuerNotFound('No issuer certificate found for '
636                                         'certificate "%s"' % 
637                                         x509Cert2Verify.dn)
638           
639        for caCert in caX509Stack:
640            issuerDN = x509Cert2Verify.issuer
641            if caCert.dn == issuerDN:
642                issuerX509Cert = caCert
643                break
644       
645        if issuerX509Cert:
646            if not x509Cert2Verify.verify(issuerX509Cert.pubKey):
647                X509CertInvalidSignature('Signature is invalid for cert. "%s"' %
648                                         x509Cert2Verify.dn)
649           
650            # Chain is validated through to CA cert
651            return
652        else:
653            raise X509CertIssuerNotFound('No issuer cert. found for '
654                                         'certificate "%s"'%x509Cert2Verify.dn)
655       
656        # If this point is reached then an issuing cert is missing from the
657        # chain       
658        raise X509CertIssuerNotFound('Can\'t find issuer cert "%s" for '
659                                     'certificate "%s"' %
660                                     (x509Cert2Verify.issuer, 
661                                      x509Cert2Verify.dn))
662
663
664def X509StackParseFromDER(derString):
665    """Make a new stack from a DER string
666   
667    @param derString: DER formatted X.509 stack data
668    @type derString: string
669    @return: new stack object
670    @rtype: X509Stack""" 
671    return X509Stack(m2X509Stack=M2Crypto.X509.new_stack_from_der(derString))
672
673
674class X500DNError(Exception):
675    """Exception handling for NDG X.500 DN class."""
676
677
678# For use with parseSeparator method:
679import re
680
681
682class X500DN(dict):
683    "NDG X500 Distinguished name"
684   
685    # Class attribute - look-up mapping short name attributes to their long
686    # name equivalents
687    # * private *
688    __shortNameLUT = {
689        'commonName':               'CN',
690        'organisationalUnitName':   'OU',
691        'organisation':             'O',
692        'countryName':              'C',
693        'emailAddress':             'EMAILADDRESS',
694        'localityName':             'L',
695        'stateOrProvinceName':      'ST',
696        'streetAddress':            'STREET',
697        'domainComponent':              'DC',
698        'userid':                       'UID'
699    }
700    PARSER_RE_STR = '/(%s)=' % '|'.join(__shortNameLUT.keys() + 
701                                        __shortNameLUT.values())
702   
703    PARSER_RE = re.compile(PARSER_RE_STR)
704   
705    def __init__(self, dn=None, m2CryptoX509Name=None, separator=None):
706
707        """Create a new X500 Distinguished Name
708
709        @type m2CryptoX509Name: M2Crypto.X509.X509_Name
710        @param m2CryptoX509Name:   initialise using using an
711        M2Crypto.X509.X509_Name
712        @type dn: basestring
713        @param dn: initialise using a distinguished name string
714        @type separator: basestring
715        @param: separator: separator used to delimit dn fields - usually '/'
716        or ','.  If dn is input and separator is omitted the separator
717        character will be automatically parsed from the dn string.
718        """
719       
720        # Private key data
721        self.__dat = {}.fromkeys(X500DN.__shortNameLUT.values(), '')
722   
723        dict.__init__(self)
724   
725        self.__separator = None
726       
727        # Check for separator from input
728        if separator is not None:
729            if not isinstance(separator, basestring):
730                raise X500DNError("dn Separator must be a valid string")
731
732            # Check for single character but allow trailing space chars
733            if len(separator.lstrip()) is not 1:
734                raise X500DNError("dn separator must be a single character")
735
736            self.__separator = separator
737           
738        if m2CryptoX509Name is not None:
739            # the argument is an x509 dn in m2crypto format
740            self.deserialise(str(m2CryptoX509Name))
741           
742        elif dn is not None:
743            # Separator can be parsed from the input DN string - only attempt
744            # if no explict separator was input
745            if self.__separator is None:
746                self.__separator = self.parseSeparator(dn)
747               
748            # Split Distinguished name string into constituent fields
749            self.deserialise(dn)
750
751    @classmethod
752    def fromString(cls, dn):
753        """Convenience method for parsing DN string into a new instance
754        """
755        return cls(dn=dn)
756
757    def __repr__(self):
758        """Give representation based on underlying dict object"""
759        return repr(self.__dat)
760       
761    def __str__(self):
762        """Behaviour for print and string statements - convert DN into
763        serialised format."""
764        return self.serialise()
765       
766    def __eq__(self, x500dn):
767        """Return true if the all the fields of the two DNs are equal"""
768       
769        if not isinstance(x500dn, X500DN):
770            return False
771
772        return self.__dat.items() == x500dn.items()
773   
774    def __ne__(self, x500dn):
775        """Return true if the all the fields of the two DNs are equal"""
776       
777        if not isinstance(x500dn, X500DN):
778            return False
779
780        return self.__dat.items() != x500dn.items()
781 
782    def __delitem__(self, key):
783        """Prevent keys from being deleted."""
784        raise X500DNError('Keys cannot be deleted from the X500DN')
785
786    def __getitem__(self, key):
787
788        # Check input key
789        if self.__dat.has_key(key):
790
791            # key recognised
792            return self.__dat[key]
793       
794        elif X500DN.__shortNameLUT.has_key(key):
795
796            # key not recognised - but a long name version of the key may
797            # have been passed
798            shortName = X500DN.__shortNameLUT[key]
799            return self.__dat[shortName]
800
801        else:
802            # key not recognised as a short or long name version
803            raise KeyError('Key "' + key + '" not recognised for X500DN')
804
805    def __setitem__(self, key, item):
806       
807        # Check input key
808        if self.__dat.has_key(key):
809
810            # key recognised
811            self.__dat[key] = item
812           
813        elif X500DN.__shortNameLUT.has_key(key):
814               
815            # key not recognised - but a long name version of the key may
816            # have been passed
817            shortName = X500DN.__shortNameLUT[key]
818            self.__dat[shortName] = item
819           
820        else:
821            # key not recognised as a short or long name version
822            raise KeyError('Key "' + key + '" not recognised for X500DN')
823
824    def clear(self):
825        raise X500DNError("Data cannot be cleared from X500DN")
826
827    def copy(self):
828
829        import copy
830        return copy.copy(self)
831
832    def keys(self):
833        return self.__dat.keys()
834
835    def items(self):
836        return self.__dat.items()
837
838    def values(self):
839        return self.__dat.values()
840
841    def has_key(self, key):
842        return self.__dat.has_key(key)
843
844    # 'in' operator
845    def __contains__(self, key):
846        return self.has_key(key)
847
848    def get(self, *arg):
849        return self.__dat.get(*arg)
850 
851    def serialise(self, separator=None):
852        """Combine fields in Distinguished Name into a single string."""
853       
854        if separator:
855            if not isinstance(separator, basestring):
856                raise X500DNError("Separator must be a valid string")
857               
858            self.__separator = separator
859           
860        else:
861            # Default to / if no separator is set
862            separator = '/'
863
864
865        # If using '/' then prepend DN with an initial '/' char
866        if separator == '/':
867            sDN = separator
868        else:
869            sDN = ''
870     
871        dnList = []
872        for (key, val) in self.__dat.items():
873            if val:
874                if isinstance(val, tuple):
875                    dnList += [separator.join(["%s=%s" % (key, valSub) \
876                                               for valSub in val])]
877                else:
878                    dnList += ["%s=%s" % (key, val)]
879               
880        sDN += separator.join(dnList)
881                               
882        return sDN
883
884    serialize = serialise
885   
886    def deserialise(self, dn, separator=None):
887        """Break up a DN string into it's constituent fields and use to
888        update the object's dictionary"""
889       
890        if separator:
891            if not isinstance(separator, basestring):
892                raise X500DNError("Separator must be a valid string")
893
894            self.__separator = separator
895
896
897        # If no separator has been set, parse if from the DN string           
898        if self.__separator is None:
899            self.__separator = self.parseSeparator(dn)
900
901        try:
902            dnFields = X500DN.PARSER_RE.split(dn)
903            if len(dnFields) < 2:
904                raise X500DNError("Error parsing DN string: \"%s\"" % dn)
905
906            items = zip(dnFields[1::2], dnFields[2::2])
907           
908            # Reset existing dictionary values
909            self.__dat.fromkeys(self.__dat, '')
910           
911            # Strip leading and trailing space chars and convert into a
912            # dictionary
913            parsedDN = {}
914            for key, val in items:
915                key = key.strip()
916                if key in parsedDN:
917                    if isinstance(parsedDN[key], tuple):
918                        parsedDN[key] = tuple(list(parsedDN[key]) + [val])
919                    else:
920                        parsedDN[key] = (parsedDN[key], val)
921                else:
922                    parsedDN[key] = val
923               
924            # Copy matching DN fields
925            for key, val in parsedDN.items():
926                if key not in self.__dat and key not in self.__shortNameLUT:
927                    raise X500DNError('Invalid field "%s" in input DN string' %
928                                      key)
929
930                self.__dat[key] = val
931
932               
933        except Exception, excep:
934            raise X500DNError("Error de-serialising DN \"%s\": %s" % \
935                              (dn, str(excep)))
936
937    deserialize = deserialise
938   
939    def parseSeparator(self, dn):
940        """Attempt to parse the separator character from a given input
941        DN string.  If not found, return None
942
943        DNs don't use standard separators e.g.
944
945        /C=UK/O=eScience/OU=CLRC/L=DL/CN=AN Other
946        CN=SUM Oneelse,L=Didcot, O=RAL,OU=SSTD
947
948        This function isolates and identifies the character.  - In the above,
949        '/' and ',' respectively"""
950
951
952        # Make a regular expression containing all the possible field
953        # identifiers with equal sign appended and 'or'ed together.  \W should
954        # match the separator which preceeds the field name. \s* allows any
955        # whitespace between field name and field separator to be taken into
956        # account.
957        #
958        # The resulting match should be a list.  The first character in each
959        # element in the list should be the field separator and should be the
960        # same
961        regExpr = '|'.join(['\W\s*'+i+'=' for i in self.__dat.keys()])
962        match = re.findall(regExpr, dn)
963           
964        # In the first example above, the resulting match is:
965        # ['/C=', '/O=', '/OU=', '/L=']
966        # In each element the first character is the separator
967        sepList = [i[0:1] for i in match]
968
969        # All separators should be the same character - return None if they
970        # don't match
971        if not [i for i in sepList if i != sepList[0]]:
972            return sepList[0]
973        else:
974            return None
975
976    @classmethod
977    def Parse(cls, dn):
978        """Convenience method to create an X500DN object from a DN string
979        @type dn: basestring
980        @param dn: Distinguished Name
981        """
982        return cls(dn=dn)
983   
984    Deserialise = Deserialize = Parse
Note: See TracBrowser for help on using the repository browser.