[CalendarServer-changes] [1289] CalendarServer/branches/users/dreid/new-twisted/lib-patches/Twisted

source_changes at macosforge.org source_changes at macosforge.org
Wed Feb 28 10:49:24 PST 2007


Revision: 1289
          http://trac.macosforge.org/projects/calendarserver/changeset/1289
Author:   dreid at apple.com
Date:     2007-02-28 10:49:24 -0800 (Wed, 28 Feb 2007)

Log Message:
-----------
Make web2.auth.digest and web2.test.test_httpauth apply cleanly

Modified Paths:
--------------
    CalendarServer/branches/users/dreid/new-twisted/lib-patches/Twisted/twisted.web2.auth.digest.patch
    CalendarServer/branches/users/dreid/new-twisted/lib-patches/Twisted/twisted.web2.test.test_httpauth.patch

Modified: CalendarServer/branches/users/dreid/new-twisted/lib-patches/Twisted/twisted.web2.auth.digest.patch
===================================================================
--- CalendarServer/branches/users/dreid/new-twisted/lib-patches/Twisted/twisted.web2.auth.digest.patch	2007-02-28 18:41:36 UTC (rev 1288)
+++ CalendarServer/branches/users/dreid/new-twisted/lib-patches/Twisted/twisted.web2.auth.digest.patch	2007-02-28 18:49:24 UTC (rev 1289)
@@ -1,17 +1,153 @@
 Index: twisted/web2/auth/digest.py
 ===================================================================
---- twisted/web2/auth/digest.py	(revision 18545)
+--- twisted/web2/auth/digest.py	(revision 19737)
 +++ twisted/web2/auth/digest.py	(working copy)
-@@ -5,6 +5,7 @@
+@@ -5,9 +5,10 @@
  
  http://www.faqs.org/rfcs/rfc2617.html
  """
 +import time
  
  from twisted.cred import credentials, error
- from zope.interface import implements
-@@ -118,10 +119,12 @@
- class DigestCredentialFactory:
+-from zope.interface import implements
++from zope.interface import implements, Interface
+ 
+ from twisted.web2.auth.interfaces import ICredentialFactory
+ 
+@@ -17,9 +18,9 @@
+ # The digest math
+ 
+ algorithms = {
+-    'md5': md5.md5,
+-    'md5-sess': md5.md5,
+-    'sha': sha.sha,
++    'md5': md5.new,
++    'md5-sess': md5.new,
++    'sha': sha.new,
+ }
+ 
+ # DigestCalcHA1
+@@ -30,15 +31,40 @@
+     pszPassword,
+     pszNonce,
+     pszCNonce,
++    preHA1=None
+ ):
++    """
++    @param pszAlg: The name of the algorithm to use to calculate the digest.
++        Currently supported are md5 md5-sess and sha.
+ 
+-    m = algorithms[pszAlg]()
+-    m.update(pszUserName)
+-    m.update(":")
+-    m.update(pszRealm)
+-    m.update(":")
+-    m.update(pszPassword)
+-    HA1 = m.digest()
++    @param pszUserName: The username
++    @param pszRealm: The realm
++    @param pszPassword: The password
++    @param pszNonce: The nonce
++    @param pszCNonce: The cnonce
++
++    @param preHA1: If available this is a str containing a previously
++       calculated HA1 as a hex string. If this is given then the values for
++       pszUserName, pszRealm, and pszPassword are ignored.
++    """
++
++    if (preHA1 and (pszUserName or pszRealm or pszPassword)):
++        raise TypeError(("preHA1 is incompatible with the pszUserName, "
++                         "pszRealm, and pszPassword arguments"))
++
++    if not preHA1:
++        # We need to calculate the HA1 from the username:realm:password
++        m = algorithms[pszAlg]()
++        m.update(pszUserName)
++        m.update(":")
++        m.update(pszRealm)
++        m.update(":")
++        m.update(pszPassword)
++        HA1 = m.digest()
++    else:
++        # We were given a username:realm:password
++        HA1 = preHA1.decode('hex')
++
+     if pszAlg == "md5-sess":
+         m = algorithms[pszAlg]()
+         m.update(HA1)
+@@ -47,6 +73,7 @@
+         m.update(":")
+         m.update(pszCNonce)
+         HA1 = m.digest()
++
+     return HA1.encode('hex')
+ 
+ # DigestCalcResponse
+@@ -83,14 +110,29 @@
+         m.update(pszQop)
+         m.update(":")
+     m.update(HA2)
+-    hash = m.digest().encode('hex')
+-    return hash
++    respHash = m.digest().encode('hex')
++    return respHash
+ 
+ 
++class IUsernameDigestHash(Interface):
++    """
++    This credential is used when a CredentialChecker has access to the hash
++    of the username:realm:password as in an Apache .htdigest file.
++    """
++    def checkHash(self, digestHash):
++        """
++        @param digestHash: The hashed username:realm:password to check against.
++
++        @return: a deferred which becomes, or a boolean indicating if the
++            hash matches.
++        """
++
++
+ class DigestedCredentials:
+     """Yet Another Simple HTTP Digest authentication scheme"""
+ 
+-    implements(credentials.IUsernameHashedPassword)
++    implements(credentials.IUsernameHashedPassword,
++               IUsernameDigestHash)
+ 
+     def __init__(self, username, method, realm, fields):
+         self.username = username
+@@ -114,40 +156,143 @@
+ 
+         return expected == response
+ 
++    def checkHash(self, digestHash):
++        response = self.fields.get('response')
++        uri = self.fields.get('uri')
++        nonce = self.fields.get('nonce')
++        cnonce = self.fields.get('cnonce')
++        nc = self.fields.get('nc')
++        algo = self.fields.get('algorithm', 'md5').lower()
++        qop = self.fields.get('qop', 'auth')
+ 
+-class DigestCredentialFactory:
++        expected = calcResponse(
++            calcHA1(algo, None, None, None, nonce, cnonce, preHA1=digestHash),
++            algo, nonce, nc, cnonce, qop, self.method, uri, None
++        )
++
++        return expected == response
++
++
++class DigestCredentialFactory(object):
++    """
++    Support for RFC2617 HTTP Digest Authentication
++
++    @cvar CHALLENGE_LIFETIME_SECS: The number of seconds for which an
++        opaque should be valid.
++
++    @ivar privateKey: A random string used for generating the secure opaque.
++    """
++
      implements(ICredentialFactory)
  
 -    CHALLENGE_LIFETIME = 15
@@ -19,57 +155,117 @@
  
      scheme = "digest"
  
-+    pkey = '%d%d%d' %  tuple([random.randrange(sys.maxint) for _ in range(3)])
-+
      def __init__(self, algorithm, realm):
-         """@type algorithm: C{str}
-            @param algorithm: case insensitive string that specifies
-@@ -132,7 +135,6 @@
-            @param realm: case sensitive string that specifies the realm
-                          portion of the challenge
+-        """@type algorithm: C{str}
+-           @param algorithm: case insensitive string that specifies
+-              the hash algorithm used, should be either, md5, md5-sess
+-              or sha
++        """
++        @type algorithm: C{str}
++        @param algorithm: case insensitive string that specifies
++            the hash algorithm used, should be either, md5, md5-sess
++            or sha
+ 
+-           @type realm: C{str}
+-           @param realm: case sensitive string that specifies the realm
+-                         portion of the challenge
++        @type realm: C{str}
++        @param realm: case sensitive string that specifies the realm
++            portion of the challenge
          """
 -        self.outstanding = {}
          self.algorithm = algorithm
          self.realm = realm
  
-@@ -141,13 +143,41 @@
++        c = tuple([random.randrange(sys.maxint) for _ in range(3)])
++
++        self.privateKey = '%d%d%d' % c
++
+     def generateNonce(self):
+         c = tuple([random.randrange(sys.maxint) for _ in range(3)])
          c = '%d%d%d' % c
          return c
  
 -    def generateOpaque(self):
 -        return str(random.randrange(sys.maxint))
++    def _getTime(self):
++        """
++        Parameterize the time based seed used in generateOpaque
++        so we can deterministically unittest it's behavior.
++        """
++        return time.time()
+ 
 +    def generateOpaque(self, nonce, clientip):
-+        # Now, what we do is encode the nonce, client ip and a timestamp in the opaque value
-+        # with a suitable digest
-+        key = "%s,%s,%s" % (nonce, clientip, str(int(time.time())))
-+        digest = md5.new(key + DigestCredentialFactory.pkey).hexdigest()
++        """
++        Generate an opaque to be returned to the client.
++        This should be a unique string that can be returned to us and verified.
++        """
++
++        # Now, what we do is encode the nonce, client ip and a timestamp
++        # in the opaque value with a suitable digest
++        key = "%s,%s,%s" % (nonce, clientip, str(int(self._getTime())))
++        digest = md5.new(key + self.privateKey).hexdigest()
 +        ekey = key.encode('base64')
-+        return "%s-%s" % (digest, ekey.replace('\n', ''),)
- 
++        return "%s-%s" % (digest, ekey.strip('\n'))
++
 +    def verifyOpaque(self, opaque, nonce, clientip):
++        """
++        Given the opaque and nonce from the request, as well as the clientip
++        that made the request, verify that the opaque was generated by us.
++        And that it's not too old."
++
++        @param opaque: The opaque value from the Digest response
++        @param nonce: The nonce value from the Digest response
++        @param clientip: The remote IP address of the client making the request
++
++        @return: Return True if the opaque was successfully verified
++
++        @raise L{twisted.cred.error.LoginFailed}
++        """
++
 +        # First split the digest from the key
 +        opaque_parts = opaque.split('-')
 +        if len(opaque_parts) != 2:
 +            raise error.LoginFailed('Invalid response, invalid opaque value')
-+        
++
 +        # Verify the key
 +        key = opaque_parts[1].decode('base64')
 +        key_parts = key.split(',')
++
 +        if len(key_parts) != 3:
 +            raise error.LoginFailed('Invalid response, invalid opaque value')
++
 +        if key_parts[0] != nonce:
-+            raise error.LoginFailed('Invalid response, incompatible opaque/nonce values')
++            raise error.LoginFailed(
++                'Invalid response, incompatible opaque/nonce values')
++
 +        if key_parts[1] != clientip:
-+            raise error.LoginFailed('Invalid response, incompatible opaque/client values')
-+        if int(time.time()) - int(key_parts[2]) > DigestCredentialFactory.CHALLENGE_LIFETIME_SECS:
-+            raise error.LoginFailed('Invalid response, incompatible opaque/nonce too old')
++            raise error.LoginFailed(
++                'Invalid response, incompatible opaque/client values')
 +
++        if (int(self._getTime()) - int(key_parts[2]) >
++            DigestCredentialFactory.CHALLENGE_LIFETIME_SECS):
++
++            raise error.LoginFailed(
++                'Invalid response, incompatible opaque/nonce too old')
++
 +        # Verify the digest
-+        digest = md5.new(key + DigestCredentialFactory.pkey).hexdigest()
++        digest = md5.new(key + self.privateKey).hexdigest()
 +        if digest != opaque_parts[0]:
 +            raise error.LoginFailed('Invalid response, invalid opaque value')
-+        
++
++        return True
++
      def getChallenge(self, peer):
++        """
++        Generate the challenge for use in the WWW-Authenticate header
++
++        @param peer: The L{IAddress} of the requesting client.
++
++        @return: The C{dict} that can be used to generate a WWW-Authenticate
++            header.
++        """
++
          c = self.generateNonce()
 -        o = self.generateOpaque()
 -        self.outstanding[o] = c
@@ -78,15 +274,32 @@
          return {'nonce': c,
                  'opaque': o,
                  'qop': 'auth',
-@@ -167,9 +197,7 @@
+@@ -161,15 +306,22 @@
+             return s
+         response = ' '.join(response.splitlines())
+         parts = response.split(',')
+-        auth = dict([(k.strip(), unq(v.strip())) for (k, v) in [p.split('=', 1) for p in parts]])
+ 
++        auth = {}
++
++        for (k, v) in [p.split('=', 1) for p in parts]:
++            auth[k.strip()] = unq(v.strip())
++
+         username = auth.get('username')
          if not username:
              raise error.LoginFailed('Invalid response, no username given')
  
 -        if auth.get('opaque') not in self.outstanding:
 -            raise error.LoginFailed('Invalid response, opaque not outstanding')
++        # Now verify the nonce/opaque values for this client
++        if self.verifyOpaque(auth.get('opaque'),
++                             auth.get('nonce'),
++                             request.remoteAddr.host):
+ 
+-        del self.outstanding[auth['opaque']]
 -
--        del self.outstanding[auth['opaque']]
-+        # Now verify the nonce/opaque values for this client
-+        self.verifyOpaque(auth.get('opaque'), auth.get('nonce'), request.remoteAddr.host)
-             
-         return DigestedCredentials(username, request.method, self.realm, auth)
+-        return DigestedCredentials(username, request.method, self.realm, auth)
++            return DigestedCredentials(username,
++                                       request.method,
++                                       self.realm,
++                                       auth)

Modified: CalendarServer/branches/users/dreid/new-twisted/lib-patches/Twisted/twisted.web2.test.test_httpauth.patch
===================================================================
--- CalendarServer/branches/users/dreid/new-twisted/lib-patches/Twisted/twisted.web2.test.test_httpauth.patch	2007-02-28 18:41:36 UTC (rev 1288)
+++ CalendarServer/branches/users/dreid/new-twisted/lib-patches/Twisted/twisted.web2.test.test_httpauth.patch	2007-02-28 18:49:24 UTC (rev 1289)
@@ -1,106 +1,552 @@
 Index: twisted/web2/test/test_httpauth.py
 ===================================================================
---- twisted/web2/test/test_httpauth.py	(revision 18545)
+--- twisted/web2/test/test_httpauth.py	(revision 19737)
 +++ twisted/web2/test/test_httpauth.py	(working copy)
-@@ -1,3 +1,6 @@
+@@ -1,3 +1,5 @@
 +import md5
-+import time
 +from twisted.internet import address
  from twisted.trial import unittest
- from twisted.internet import defer
  from twisted.cred import error
-@@ -12,8 +15,13 @@
+ from twisted.web2.auth import basic, digest, wrapper
+@@ -9,12 +11,29 @@
+ import base64
+ 
+ class FakeDigestCredentialFactory(digest.DigestCredentialFactory):
++    """
++    A Fake Digest Credential Factory that generates a predictable
++    nonce and opaque
++    """
++
++    def __init__(self, *args, **kwargs):
++        super(FakeDigestCredentialFactory, self).__init__(*args, **kwargs)
++
++        self.privateKey = "0"
++
      def generateNonce(self):
++        """
++        Generate a static nonce
++        """
          return '178288758716122392881254770685'
  
 -    def generateOpaque(self):
 -        return '1041524039'
-+    def generateOpaque(self, nonce, clientip):
-+        # Now, what we do is encode the nonce, client ip and a timestamp in the opaque value
-+        # with a suitable digest
-+        key = "%s,%s,%s" % (nonce, clientip, str(int(0)))
-+        digest = md5.new(key + "0").hexdigest()
-+        ekey = key.encode('base64')
-+        return "%s-%s" % (digest, ekey.replace('\n', ''),)
++    def _getTime(self):
++        """
++        Return a stable time
++        """
++        return 0
  
++
  class BasicAuthTestCase(unittest.TestCase):
      def setUp(self):
-@@ -56,32 +64,63 @@
+         self.credentialFactory = basic.BasicCredentialFactory('foo')
+@@ -54,39 +73,315 @@
+                           self.credentialFactory.decode,
+                           response, _trivial_GET)
  
- challengeResponse = ('digest', {'nonce': '178288758716122392881254770685', 
-                                 'qop': 'auth', 'realm': 'test realm', 
+-challengeResponse = ('digest', {'nonce': '178288758716122392881254770685',
+-                                'qop': 'auth', 'realm': 'test realm',
 -                                'algorithm': 'md5', 'opaque': '1041524039'})
-+                                'algorithm': 'md5',
-+                                'opaque': '75c4bd95b96b7b7341c646c6502f0833-MTc4Mjg4NzU4NzE2MTIyMzkyODgxMjU0NzcwNjg1LHJlbW90ZWhvc3QsMA=='})
  
 -authRequest = 'username="username", realm="test realm", nonce="178288758716122392881254770685", uri="/write/", response="62f388be1cf678fbdfce87910871bcc5", opaque="1041524039", algorithm="md5", cnonce="29fc54aa1641c6fa0e151419361c8f23", nc=00000001, qop="auth"'
-+cnonce = "29fc54aa1641c6fa0e151419361c8f23"
++clientAddress = address.IPv4Address('TCP', '127.0.0.1', 80)
  
-+authRequest1 = 'username="username", realm="test realm", nonce="%s", uri="/write/", response="%s", opaque="%s", algorithm="md5", cnonce="29fc54aa1641c6fa0e151419361c8f23", nc=00000001, qop="auth"'
-+authRequest2 = 'username="username", realm="test realm", nonce="%s", uri="/write/", response="%s", opaque="%s", algorithm="md5", cnonce="29fc54aa1641c6fa0e151419361c8f23", nc=00000002, qop="auth"'
++challengeOpaque = ('75c4bd95b96b7b7341c646c6502f0833-MTc4Mjg4NzU'
++                   '4NzE2MTIyMzkyODgxMjU0NzcwNjg1LHJlbW90ZWhvc3Q'
++                   'sMA==')
 +
++challengeNonce = '178288758716122392881254770685'
++
++challengeResponse = ('digest',
++                     {'nonce': challengeNonce,
++                      'qop': 'auth', 'realm': 'test realm',
++                      'algorithm': 'md5',
++                      'opaque': challengeOpaque})
++
++cnonce = "29fc54aa1641c6fa0e151419361c8f23"
++
++authRequest1 = ('username="username", realm="test realm", nonce="%s", '
++                'uri="/write/", response="%s", opaque="%s", algorithm="md5", '
++                'cnonce="29fc54aa1641c6fa0e151419361c8f23", nc=00000001, '
++                'qop="auth"')
++
++authRequest2 = ('username="username", realm="test realm", nonce="%s", '
++                'uri="/write/", response="%s", opaque="%s", algorithm="md5", '
++                'cnonce="29fc54aa1641c6fa0e151419361c8f23", nc=00000002, '
++                'qop="auth"')
++
  namelessAuthRequest = 'realm="test realm",nonce="doesn\'t matter"'
  
++
  class DigestAuthTestCase(unittest.TestCase):
++    """
++    Test the behavior of DigestCredentialFactory
++    """
++
      def setUp(self):
--        self.credentialFactory = FakeDigestCredentialFactory('md5', 
-+        self.credentialFactory = digest.DigestCredentialFactory('md5', 
-                                                              'test realm')
+-        self.credentialFactory = FakeDigestCredentialFactory('md5',
+-                                                             'test realm')
++        """
++        Create a DigestCredentialFactory for testing
++        """
++        self.credentialFactory = digest.DigestCredentialFactory('md5',
++                                                                'test realm')
  
+-    def testGetChallenge(self):
+-        self.assertEquals(
+-            self.credentialFactory.getChallenge(None),
+-            challengeResponse[1])
 +    def getDigestResponse(self, challenge, ncount):
++        """
++        Calculate the response for the given challenge
++        """
 +        nonce = challenge.get('nonce')
 +        algo = challenge.get('algorithm').lower()
 +        qop = challenge.get('qop')
-+        
+ 
+-    def testResponse(self):
+-        challenge = self.credentialFactory.getChallenge(None)
 +        expected = digest.calcResponse(
-+            digest.calcHA1(algo, "username", "test realm", "password", nonce, cnonce),
++            digest.calcHA1(algo,
++                           "username",
++                           "test realm",
++                           "password",
++                           nonce,
++                           cnonce),
 +            algo, nonce, ncount, cnonce, qop, "GET", "/write/", None
-+        )
++            )
 +        return expected
+ 
+-        creds = self.credentialFactory.decode(authRequest, _trivial_GET)
++    def test_getChallenge(self):
++        """
++        Test that all the required fields exist in the challenge,
++        and that the information matches what we put into our
++        DigestCredentialFactory
++        """
 +
-     def testGetChallenge(self):
--        self.assertEquals(
--            self.credentialFactory.getChallenge(None),
--            challengeResponse[1])
-+        challenge = self.credentialFactory.getChallenge(address.IPv4Address('TCP', "127.0.0.1", 80))
++        challenge = self.credentialFactory.getChallenge(clientAddress)
 +        self.assertEquals(challenge['qop'], 'auth')
 +        self.assertEquals(challenge['realm'], 'test realm')
 +        self.assertEquals(challenge['algorithm'], 'md5')
 +        self.assertTrue(challenge.has_key("nonce"))
 +        self.assertTrue(challenge.has_key("opaque"))
++
++    def test_response(self):
++        """
++        Test that we can decode a valid response to our challenge
++        """
++
++        challenge = self.credentialFactory.getChallenge(clientAddress)
++
++        clientResponse = authRequest1 % (
++            challenge['nonce'],
++            self.getDigestResponse(challenge, "00000001"),
++            challenge['opaque'])
++
++        creds = self.credentialFactory.decode(clientResponse, _trivial_GET)
+         self.failUnless(creds.checkPassword('password'))
  
-     def testResponse(self):
+-    def testFailsWithDifferentMethod(self):
 -        challenge = self.credentialFactory.getChallenge(None)
-+        challenge = self.credentialFactory.getChallenge(address.IPv4Address('TCP', "127.0.0.1", 80))
++    def test_multiResponse(self):
++        """
++        Test that multiple responses to to a single challenge are handled
++        successfully.
++        """
  
--        creds = self.credentialFactory.decode(authRequest, _trivial_GET)
-+        clientResponse = authRequest1 % (challenge['nonce'], self.getDigestResponse(challenge, "00000001"), challenge['opaque'])
-+        creds = self.credentialFactory.decode(clientResponse, _trivial_GET)
-         self.failUnless(creds.checkPassword('password'))
- 
-+    def testMultiResponse(self):
-+        challenge = self.credentialFactory.getChallenge(address.IPv4Address('TCP', "127.0.0.1", 80))
+-        creds = self.credentialFactory.decode(authRequest, SimpleRequest(None, 'POST', '/'))
++        challenge = self.credentialFactory.getChallenge(clientAddress)
 +
-+        clientResponse = authRequest1 % (challenge['nonce'], self.getDigestResponse(challenge, "00000001"), challenge['opaque'])
++        clientResponse = authRequest1 % (
++            challenge['nonce'],
++            self.getDigestResponse(challenge, "00000001"),
++            challenge['opaque'])
++
 +        creds = self.credentialFactory.decode(clientResponse, _trivial_GET)
 +        self.failUnless(creds.checkPassword('password'))
 +
-+        clientResponse = authRequest2 % (challenge['nonce'], self.getDigestResponse(challenge, "00000002"), challenge['opaque'])
++        clientResponse = authRequest2 % (
++            challenge['nonce'],
++            self.getDigestResponse(challenge, "00000002"),
++            challenge['opaque'])
++
 +        creds = self.credentialFactory.decode(clientResponse, _trivial_GET)
 +        self.failUnless(creds.checkPassword('password'))
 +
-     def testFailsWithDifferentMethod(self):
--        challenge = self.credentialFactory.getChallenge(None)
-+        challenge = self.credentialFactory.getChallenge(address.IPv4Address('TCP', "127.0.0.1", 80))
-         
--        creds = self.credentialFactory.decode(authRequest, SimpleRequest(None, 'POST', '/'))
-+        clientResponse = authRequest1 % (challenge['nonce'], self.getDigestResponse(challenge, "00000001"), challenge['opaque'])
-+        creds = self.credentialFactory.decode(clientResponse, SimpleRequest(None, 'POST', '/'))
++    def test_failsWithDifferentMethod(self):
++        """
++        Test that the response fails if made for a different request method
++        than it is being issued for.
++        """
++
++        challenge = self.credentialFactory.getChallenge(clientAddress)
++
++        clientResponse = authRequest1 % (
++            challenge['nonce'],
++            self.getDigestResponse(challenge, "00000001"),
++            challenge['opaque'])
++
++        creds = self.credentialFactory.decode(clientResponse,
++                                              SimpleRequest(None, 'POST', '/'))
          self.failIf(creds.checkPassword('password'))
  
-     def testNoUsername(self):
-@@ -221,7 +260,7 @@
+-    def testNoUsername(self):
+-        self.assertRaises(error.LoginFailed, self.credentialFactory.decode, namelessAuthRequest, _trivial_GET)
++    def test_noUsername(self):
++        """
++        Test that login fails when our response does not contain a username
++        """
  
++        self.assertRaises(error.LoginFailed,
++                          self.credentialFactory.decode,
++                          namelessAuthRequest,
++                          _trivial_GET)
++
++    def test_checkHash(self):
++        """
++        Check that given a hash of the form 'username:realm:password'
++        we can verify the digest challenge
++        """
++
++        challenge = self.credentialFactory.getChallenge(clientAddress)
++
++        clientResponse = authRequest1 % (
++            challenge['nonce'],
++            self.getDigestResponse(challenge, "00000001"),
++            challenge['opaque'])
++
++        creds = self.credentialFactory.decode(clientResponse, _trivial_GET)
++
++        self.failUnless(creds.checkHash(
++                md5.md5('username:test realm:password').hexdigest()))
++
++        self.failIf(creds.checkHash(
++                md5.md5('username:test realm:bogus').hexdigest()))
++
++    def test_invalidOpaque(self):
++        """
++        Test that login fails when the opaque does not contain all the required
++        parts.
++        """
++
++        credentialFactory = FakeDigestCredentialFactory('md5', 'test realm')
++
++        challenge = credentialFactory.getChallenge(clientAddress)
++
++        self.assertRaises(
++            error.LoginFailed,
++            credentialFactory.verifyOpaque,
++            'badOpaque',
++            challenge['nonce'],
++            clientAddress.host)
++
++        badOpaque = ('foo-%s' % (
++                'nonce,clientip'.encode('base64').strip('\n'),))
++
++        self.assertRaises(
++            error.LoginFailed,
++            credentialFactory.verifyOpaque,
++            badOpaque,
++            challenge['nonce'],
++            clientAddress.host)
++
++    def test_incompatibleNonce(self):
++        """
++        Test that login fails when the given nonce from the response, does not
++        match the nonce encoded in the opaque.
++        """
++
++        credentialFactory = FakeDigestCredentialFactory('md5', 'test realm')
++
++        challenge = credentialFactory.getChallenge(clientAddress)
++
++        badNonceOpaque = credentialFactory.generateOpaque(
++            '1234567890',
++            clientAddress.host)
++
++        self.assertRaises(
++            error.LoginFailed,
++            credentialFactory.verifyOpaque,
++            badNonceOpaque,
++            challenge['nonce'],
++            clientAddress.host)
++
++    def test_incompatibleClientIp(self):
++        """
++        Test that the login fails when the request comes from a client ip
++        other than what is encoded in the opaque.
++        """
++
++        credentialFactory = FakeDigestCredentialFactory('md5', 'test realm')
++
++        challenge = credentialFactory.getChallenge(clientAddress)
++
++        badNonceOpaque = credentialFactory.generateOpaque(
++            challenge['nonce'],
++            '10.0.0.1')
++
++        self.assertRaises(
++            error.LoginFailed,
++            credentialFactory.verifyOpaque,
++            badNonceOpaque,
++            challenge['nonce'],
++            clientAddress.host)
++
++    def test_oldNonce(self):
++        """
++        Test that the login fails when the given opaque is older than
++        DigestCredentialFactory.CHALLENGE_LIFETIME_SECS
++        """
++
++        credentialFactory = FakeDigestCredentialFactory('md5', 'test realm')
++
++        challenge = credentialFactory.getChallenge(clientAddress)
++
++        key = '%s,%s,%s' % (challenge['nonce'],
++                            clientAddress.host,
++                            '-137876876')
++        digest = md5.new(key + credentialFactory.privateKey).hexdigest()
++        ekey = key.encode('base64')
++
++        oldNonceOpaque = '%s-%s' % (digest, ekey.strip('\n'))
++
++        self.assertRaises(
++            error.LoginFailed,
++            credentialFactory.verifyOpaque,
++            oldNonceOpaque,
++            challenge['nonce'],
++            clientAddress.host)
++
++    def test_mismatchedOpaqueChecksum(self):
++        """
++        Test that login fails when the opaque checksum fails verification
++        """
++
++        credentialFactory = FakeDigestCredentialFactory('md5', 'test realm')
++
++        challenge = credentialFactory.getChallenge(clientAddress)
++
++
++        key = '%s,%s,%s' % (challenge['nonce'],
++                            clientAddress.host,
++                            '0')
++
++        digest = md5.new(key + 'this is not the right pkey').hexdigest()
++
++        badChecksum = '%s-%s' % (digest,
++                                 key.encode('base64').strip('\n'))
++
++        self.assertRaises(
++            error.LoginFailed,
++            credentialFactory.verifyOpaque,
++            badChecksum,
++            challenge['nonce'],
++            clientAddress.host)
++
++    def test_incompatibleCalcHA1Options(self):
++        """
++        Test that the appropriate error is raised when any of the
++        pszUsername, pszRealm, or pszPassword arguments are specified with
++        the preHA1 keyword argument.
++        """
++
++        arguments = (
++            ("user", "realm", "password", "preHA1"),
++            (None, "realm", None, "preHA1"),
++            (None, None, "password", "preHA1"),
++            )
++
++        for pszUsername, pszRealm, pszPassword, preHA1 in arguments:
++            self.assertRaises(
++                TypeError,
++                digest.calcHA1,
++                "md5",
++                pszUsername,
++                pszRealm,
++                pszPassword,
++                "nonce",
++                "cnonce",
++                preHA1=preHA1
++                )
++
++
+ from zope.interface import implements
+ from twisted.cred import portal, checkers
+ 
+@@ -105,6 +400,7 @@
+         """
+         self.username = username
+ 
++
+ class TestAuthRealm(object):
+     """
+     Test realm that supports the IHTTPUser interface
+@@ -169,7 +465,7 @@
+         del self.credFactory
+         del self.protectedResource
+ 
+-    def test_AuthenticatedRequest(self):
++    def test_authenticatedRequest(self):
+         """
+         Test that after successful authentication the request provides
+         IAuthenticatedRequest and that the request.avatar implements
+@@ -207,7 +503,7 @@
+         d.addCallback(checkRequest)
+         return d
+ 
+-    def test_AllowedMethods(self):
++    def test_allowedMethods(self):
+         """
+         Test that unknown methods result in a 401 instead of a 405 when
+         authentication hasn't been completed.
+@@ -219,17 +515,18 @@
+                                         [self.credFactory],
+                                         self.portal,
+                                         interfaces=(IHTTPUser,))
+-        d = self.assertResponse((root, 'http://localhost/'),
+-                                (401,
+-                                 {'WWW-Authenticate': [('basic',
+-                                                        {'realm': "test realm"})]},
+-                                None))
++        d = self.assertResponse(
++            (root, 'http://localhost/'),
++            (401,
++             {'WWW-Authenticate': [('basic',
++                                    {'realm': "test realm"})]},
++             None))
+ 
+         self.method = 'GET'
+ 
+         return d
+ 
+-    def test_UnauthorizedResponse(self):
++    def test_unauthorizedResponse(self):
+         """
+         Test that a request with no credentials results in a
+         valid Unauthorized response.
+@@ -240,22 +537,24 @@
+                                         interfaces=(IHTTPUser,))
+ 
+         def makeDeepRequest(res):
+-            return self.assertResponse((root,
+-                                        'http://localhost/foo/bar/baz/bax'),
+-                                       (401,
+-                                        {'WWW-Authenticate': [('basic',
+-                                                               {'realm': "test realm"})]},
+-                                        None))
++            return self.assertResponse(
++                (root,
++                 'http://localhost/foo/bar/baz/bax'),
++                (401,
++                 {'WWW-Authenticate': [('basic',
++                                        {'realm': "test realm"})]},
++                 None))
+ 
+-        d = self.assertResponse((root, 'http://localhost/'),
+-                                (401,
+-                                 {'WWW-Authenticate': [('basic',
+-                                                        {'realm': "test realm"})]},
+-                                 None))
++        d = self.assertResponse(
++            (root, 'http://localhost/'),
++            (401,
++             {'WWW-Authenticate': [('basic',
++                                    {'realm': "test realm"})]},
++             None))
+ 
+         return d.addCallback(makeDeepRequest)
+ 
+-    def test_BadCredentials(self):
++    def test_badCredentials(self):
+         """
+         Test that a request with bad credentials results in a valid
+         Unauthorized response
+@@ -267,16 +566,17 @@
+ 
+         credentials = base64.encodestring('bad:credentials')
+ 
+-        d = self.assertResponse((root, 'http://localhost/',
+-                                 {'authorization': ('basic', credentials)}),
+-                                (401,
+-                                 {'WWW-Authenticate': [('basic',
+-                                                        {'realm': "test realm"})]},
+-                                 None))
++        d = self.assertResponse(
++            (root, 'http://localhost/',
++             {'authorization': [('basic', credentials)]}),
++            (401,
++             {'WWW-Authenticate': [('basic',
++                                    {'realm': "test realm"})]},
++             None))
+ 
+         return d
+ 
+-    def test_SuccessfulLogin(self):
++    def test_successfulLogin(self):
+         """
+         Test that a request with good credentials results in the
+         appropriate response from the protected resource
+@@ -297,7 +597,7 @@
+ 
+         return d
+ 
+-    def test_WrongScheme(self):
++    def test_wrongScheme(self):
+         """
+         Test that a request with credentials for a scheme that is not
+         advertised by this resource results in the appropriate
+@@ -319,16 +619,17 @@
+ 
+         return d
+ 
+-    def test_MultipleWWWAuthenticateSchemes(self):
++    def test_multipleWWWAuthenticateSchemes(self):
+         """
+         Test that our unauthorized response can contain challenges for
+         multiple authentication schemes.
+         """
+-        root = wrapper.HTTPAuthResource(self.protectedResource,
+-                                        (basic.BasicCredentialFactory('test realm'),
+-                                         FakeDigestCredentialFactory('md5', 'test realm')),
+-                                        self.portal,
+-                                        interfaces=(IHTTPUser,))
++        root = wrapper.HTTPAuthResource(
++            self.protectedResource,
++            (basic.BasicCredentialFactory('test realm'),
++             FakeDigestCredentialFactory('md5', 'test realm')),
++            self.portal,
++            interfaces=(IHTTPUser,))
+ 
+         d = self.assertResponse((root, 'http://localhost/', {}),
+                                 (401,
+@@ -339,15 +640,18 @@
+ 
+         return d
+ 
+-    def test_AuthorizationAgainstMultipleSchemes(self):
++    def test_authorizationAgainstMultipleSchemes(self):
+         """
+-        Test that we can authenticate to either authentication scheme.
++        Test that we can successfully authenticate when presented
++        with multiple WWW-Authenticate headers
+         """
+-        root = wrapper.HTTPAuthResource(self.protectedResource,
+-                                        (basic.BasicCredentialFactory('test realm'),
+-                                         FakeDigestCredentialFactory('md5', 'test realm')),
++
++        root = wrapper.HTTPAuthResource(
++            self.protectedResource,
++            (basic.BasicCredentialFactory('test realm'),
++             FakeDigestCredentialFactory('md5', 'test realm')),
+                                         self.portal,
+-                                        interfaces=(IHTTPUser,))
++            interfaces=(IHTTPUser,))
+ 
+         def respondBasic(ign):
+             credentials = base64.encodestring('username:password')
+@@ -362,7 +666,7 @@
+ 
          def respond(ign):
              d = self.assertResponse((root, 'http://localhost/',
 -                                     {'authorization': authRequest}),
@@ -108,3 +554,21 @@
                                      (200,
                                       {},
                                       None))
+@@ -377,7 +681,7 @@
+ 
+         return d
+ 
+-    def test_WrappedResourceGetsFullSegments(self):
++    def test_wrappedResourceGetsFullSegments(self):
+         """
+         Test that the wrapped resource gets all the URL segments in it's
+         locateChild.
+@@ -405,7 +709,7 @@
+ 
+         return d
+ 
+-    def test_InvalidCredentials(self):
++    def test_invalidCredentials(self):
+         """
+         Malformed or otherwise invalid credentials (as determined by
+         the credential factory) should result in an Unauthorized response

-------------- next part --------------
An HTML attachment was scrubbed...
URL: http://lists.macosforge.org/pipermail/calendarserver-changes/attachments/20070228/81d8058e/attachment.html


More information about the calendarserver-changes mailing list