source: TI12-security/trunk/python/ndg_security_test/ndg/security/test/unit/x509/test_x509.py @ 5953

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/TI12-security/trunk/python/ndg_security_test/ndg/security/test/unit/x509/test_x509.py@5953
Revision 5953, 7.0 KB checked in by pjkersha, 10 years ago (diff)

Important fix for parsing Distinguished Names - handle the case where the CN contains a slash as used for MyProxy/Globus? host certificates e.g. /.../CN=host/myhost.mydomain

  • Property svn:executable set to *
  • Property svn:keywords set to Id
Line 
1#!/usr/bin/env python
2"""NDG X509 Module unit tests
3
4NERC Data Grid Project
5"""
6__author__ = "P J Kershaw"
7__date__ = "03/01/07"
8__copyright__ = "(C) 2009 Science and Technology Facilities Council"
9__license__ = "BSD - see LICENSE file in top-level directory"
10__contact__ = "Philip.Kershaw@stfc.ac.uk"
11__revision__ = '$Id:test_x509.py 4335 2008-10-14 12:44:22Z pjkersha $'
12import logging
13logging.basicConfig(level=logging.DEBUG)
14log = logging.getLogger(__name__)
15
16import unittest
17import os
18import sys
19import getpass
20from StringIO import StringIO
21
22from os.path import expandvars as xpdVars
23from os.path import join as jnPath
24mkPath = lambda file: jnPath(os.environ['NDGSEC_X509_UNITTEST_DIR'], file)
25
26from ConfigParser import SafeConfigParser
27from ndg.security.test.unit import BaseTestCase
28
29import warnings
30_warningMsg = None
31_origWarn = warnings.warn
32def _warnWrapper(*arg, **kw):
33    global _warningMsg
34    _warningMsg = arg[0]
35    _origWarn(*arg, **kw)
36
37warnings.warn = _warnWrapper
38
39from ndg.security.common.X509 import X509CertRead, X509CertParse, X500DN, \
40    X509Stack, X509StackEmptyError, SelfSignedCert, X509CertIssuerNotFound
41
42class X509TestCase(BaseTestCase):
43   
44    def __del__(self):
45        warnings.warn = _origWarn
46        if getattr(super(X509TestCase, self), "__del__", None):
47            super(X509TestCase, self).__del__()
48       
49    def setUp(self):
50        super(X509TestCase, self).setUp()
51       
52        if 'NDGSEC_INT_DEBUG' in os.environ:
53            import pdb
54            pdb.set_trace()
55       
56        if 'NDGSEC_X509_UNITTEST_DIR' not in os.environ:
57            os.environ['NDGSEC_X509_UNITTEST_DIR'] = os.path.abspath(
58                                                    os.path.dirname(__file__))
59       
60        configParser = SafeConfigParser()
61        configFilePath = jnPath(os.environ['NDGSEC_X509_UNITTEST_DIR'],
62                                "x509Test.cfg")
63        configParser.read(configFilePath)
64       
65        self.cfg = {}
66        for section in configParser.sections():
67            self.cfg[section] = dict(configParser.items(section))
68                   
69    def test01X509CertRead(self):
70        # test01X509CertRead: read in a cert from file
71        self.x509Cert = X509CertRead(
72                            xpdVars(self.cfg['test01X509CertRead']['certfile']))
73        self.assert_(self.x509Cert)
74
75    def test02X509CertAsPEM(self):
76        # test02X509CertAsPEM: display as a PEM format string
77        self.test01X509CertRead()
78        self.pemString = self.x509Cert.asPEM()
79        print(self.pemString)
80
81
82    def test03X509CertParse(self):
83        # test03X509CertParse: parse from a PEM format string
84        self.test02X509CertAsPEM()
85        self.assert_(X509CertParse(self.pemString))
86
87
88    def test04GetDN(self):
89        # test04GetDN: extract distinguished name
90        self.test01X509CertRead()
91        self.dn = self.x509Cert.dn
92        print(self.dn)
93       
94    def test05DN(self):
95        # test05DN: test X.500 Distinguished Name attributes
96        self.test04GetDN()
97        for item in self.dn.items():
98            print("%s=%s" % item)
99       
100    def test06DNCmp(self):
101        # test06DNCmp: test X.500 Distinguished Name comparison operators
102        self.test04GetDN()
103        testDN = X500DN(dn="/O=a/OU=b/CN=c")
104
105        self.assert_(not(testDN == self.dn))
106        self.assert_(testDN != self.dn)
107        self.assert_(self.dn == self.dn)
108        self.assert_(not(self.dn != self.dn))
109           
110    def test07x509Stack(self):
111        # test07X509Stack: test X509Stack functionality
112
113        self.test01X509CertRead()
114        stack = X509Stack()
115        self.assert_(len(stack)==0)
116        self.assert_(stack.push(self.x509Cert))
117        self.assert_(len(stack)==1)
118        print("stack[0] = %s" % stack[0])
119        for i in stack:
120            print("stack iterator i = %s" % i)
121        print("stack.pop() = %s" % stack.pop())
122        self.assert_(len(stack)==0)
123           
124    def test08x509StackVerifyCertChain(self):
125        # test08X509StackVerifyCertChain: testVerifyCertChain method
126
127        self.test01X509CertRead()
128        proxyCert=X509CertRead(xpdVars(
129                   self.cfg['test08X509StackVerifyCertChain']['proxycertfile']))
130
131        stack1 = X509Stack()
132        stack1.push(self.x509Cert)
133       
134        caCert=X509CertRead(xpdVars(\
135                   self.cfg['test08X509StackVerifyCertChain']['cacertfile']))
136        caStack = X509Stack()
137        caStack.push(caCert)
138       
139        print("Verification of external cert with external CA stack...")
140        stack1.verifyCertChain(x509Cert2Verify=proxyCert, 
141                               caX509Stack=caStack)
142       
143        print("Verification of stack content using CA stack...")
144        stack1.push(proxyCert)
145        stack1.verifyCertChain(caX509Stack=caStack)
146       
147        print("Verification of stack alone...")
148        stack1.push(caCert)
149        stack1.verifyCertChain()
150       
151        print("Reject self-signed cert. ...")
152        stack2 = X509Stack()
153        try:
154            stack2.verifyCertChain()
155            self.fail("Empty stack error expected")
156        except X509StackEmptyError:
157            pass
158
159        stack2.push(caCert)
160        try:
161            stack2.verifyCertChain()
162            self.fail("Reject of self-signed cert. expected")
163        except SelfSignedCert:
164            pass
165       
166        print("Accept self-signed cert. ...")
167        stack2.verifyCertChain(rejectSelfSignedCert=False)
168       
169        self.assert_(stack2.pop())
170        print("Test no cert. issuer found ...")
171        stack2.push(proxyCert)
172        try:
173            stack2.verifyCertChain()
174            self.fail("No cert. issuer error expected")
175        except X509CertIssuerNotFound:
176            pass
177       
178        print("Test no cert. issuer found again with incomplete chain ...")
179        stack2.push(self.x509Cert)
180        try:
181            stack2.verifyCertChain()
182            self.fail("No cert. issuer error expected")
183        except X509CertIssuerNotFound:
184            pass
185
186    def test09ExpiryTime(self):
187        self.test01X509CertRead()
188       
189        # Set ridiculous bounds for expiry warning to ensure a warning message
190        # is output
191        self.assert_(self.x509Cert.isValidTime(nDaysBeforeExpiryLimit=36500), 
192                                               "Certificate has expired")
193        if not _warningMsg:
194            self.fail("No warning message was set")
195        else:
196            print("PASSED - Got warning message from X509Cert."
197                  "isValidTime: %s" % _warningMsg)
198
199
200class X500DNTestCase(BaseTestCase):
201    def test01VerifyParsingForFieldsContainingSlash(self):
202        # Slash is the delimiter but fields can contain a slash too - ensure
203        # correct parsing based on a regular expression which handles this
204        # scenario
205        dnStr = ("/C=UK/O=eScience/OU=CLRC/L=RAL/CN=host/localhost/"
206                 "emailAddress=somebody@somewhere.ac.uk")
207        dn = X500DN.fromString(dnStr)
208        self.assert_(str(dn))
209        print(dn)
210                                     
211if __name__ == "__main__":
212    unittest.main()
Note: See TracBrowser for help on using the repository browser.