libs/openid/test/test_negotiation.py

   1
   2 import unittest
   3 from support import CatchLogs
   4
   5 from openid.message import Message, OPENID2_NS, OPENID1_NS, OPENID_NS
   6 from openid import association
   7 from openid.consumer.consumer import GenericConsumer, ServerError
   8 from openid.consumer.discover import OpenIDServiceEndpoint, OPENID_2_0_TYPE
   9
  10 class ErrorRaisingConsumer(GenericConsumer):
  11     """
  12     A consumer whose _requestAssocation will return predefined results
  13     instead of trying to actually perform association requests.
  14     """
  15
  16     # The list of objects to be returned by successive calls to
  17     # _requestAssocation.  Each call will pop the first element from
  18     # this list and return it to _negotiateAssociation.  If the
  19     # element is a Message object, it will be wrapped in a ServerError
  20     # exception.  Otherwise it will be returned as-is.
  21     return_messages = []
  22
  23     def _requestAssociation(self, endpoint, assoc_type, session_type):
  24         m = self.return_messages.pop(0)
  25         if isinstance(m, Message):
  26             raise ServerError.fromMessage(m)
  27         else:
  28             return m
  29
  30 class TestOpenID2SessionNegotiation(unittest.TestCase, CatchLogs):
  31     """
  32     Test the session type negotiation behavior of an OpenID 2
  33     consumer.
  34     """
  35     def setUp(self):
  36         CatchLogs.setUp(self)
  37         self.consumer = ErrorRaisingConsumer(store=None)
  38
  39         self.endpoint = OpenIDServiceEndpoint()
  40         self.endpoint.type_uris = [OPENID_2_0_TYPE]
  41         self.endpoint.server_url = 'bogus'
  42
  43     def testBadResponse(self):
  44         """
  45         Test the case where the response to an associate request is a
  46         server error or is otherwise undecipherable.
  47         """
  48         self.consumer.return_messages = [Message(self.endpoint.preferredNamespace())]
  49         self.assertEqual(self.consumer._negotiateAssociation(self.endpoint), None)
  50         self.failUnlessLogMatches('Server error when requesting an association')
  51
  52     def testEmptyAssocType(self):
  53         """
  54         Test the case where the association type (assoc_type) returned
  55         in an unsupported-type response is absent.
  56         """
  57         msg = Message(self.endpoint.preferredNamespace())
  58         msg.setArg(OPENID_NS, 'error', 'Unsupported type')
  59         msg.setArg(OPENID_NS, 'error_code', 'unsupported-type')
  60         # not set: msg.delArg(OPENID_NS, 'assoc_type')
  61         msg.setArg(OPENID_NS, 'session_type', 'new-session-type')
  62
  63         self.consumer.return_messages = [msg]
  64         self.assertEqual(self.consumer._negotiateAssociation(self.endpoint), None)
  65
  66         self.failUnlessLogMatches('Unsupported association type',
  67                                   'Server responded with unsupported association ' +
  68                                   'session but did not supply a fallback.')
  69
  70     def testEmptySessionType(self):
  71         """
  72         Test the case where the session type (session_type) returned
  73         in an unsupported-type response is absent.
  74         """
  75         msg = Message(self.endpoint.preferredNamespace())
  76         msg.setArg(OPENID_NS, 'error', 'Unsupported type')
  77         msg.setArg(OPENID_NS, 'error_code', 'unsupported-type')
  78         msg.setArg(OPENID_NS, 'assoc_type', 'new-assoc-type')
  79         # not set: msg.setArg(OPENID_NS, 'session_type', None)
  80
  81         self.consumer.return_messages = [msg]
  82         self.assertEqual(self.consumer._negotiateAssociation(self.endpoint), None)
  83
  84         self.failUnlessLogMatches('Unsupported association type',
  85                                   'Server responded with unsupported association ' +
  86                                   'session but did not supply a fallback.')
  87
  88     def testNotAllowed(self):
  89         """
  90         Test the case where an unsupported-type response specifies a
  91         preferred (assoc_type, session_type) combination that is not
  92         allowed by the consumer's SessionNegotiator.
  93         """
  94         allowed_types = []
  95
  96         negotiator = association.SessionNegotiator(allowed_types)
  97         self.consumer.negotiator = negotiator
  98
  99         msg = Message(self.endpoint.preferredNamespace())
 100         msg.setArg(OPENID_NS, 'error', 'Unsupported type')
 101         msg.setArg(OPENID_NS, 'error_code', 'unsupported-type')
 102         msg.setArg(OPENID_NS, 'assoc_type', 'not-allowed')
 103         msg.setArg(OPENID_NS, 'session_type', 'not-allowed')
 104
 105         self.consumer.return_messages = [msg]
 106         self.assertEqual(self.consumer._negotiateAssociation(self.endpoint), None)
 107
 108         self.failUnlessLogMatches('Unsupported association type',
 109                                   'Server sent unsupported session/association type:')
 110
 111     def testUnsupportedWithRetry(self):
 112         """
 113         Test the case where an unsupported-type response triggers a
 114         retry to get an association with the new preferred type.
 115         """
 116         msg = Message(self.endpoint.preferredNamespace())
 117         msg.setArg(OPENID_NS, 'error', 'Unsupported type')
 118         msg.setArg(OPENID_NS, 'error_code', 'unsupported-type')
 119         msg.setArg(OPENID_NS, 'assoc_type', 'HMAC-SHA1')
 120         msg.setArg(OPENID_NS, 'session_type', 'DH-SHA1')
 121
 122         assoc = association.Association(
 123             'handle', 'secret', 'issued', 10000, 'HMAC-SHA1')
 124
 125         self.consumer.return_messages = [msg, assoc]
 126         self.failUnless(self.consumer._negotiateAssociation(self.endpoint) is assoc)
 127
 128         self.failUnlessLogMatches('Unsupported association type')
 129
 130     def testUnsupportedWithRetryAndFail(self):
 131         """
 132         Test the case where an unsupported-typ response triggers a
 133         retry, but the retry fails and None is returned instead.
 134         """
 135         msg = Message(self.endpoint.preferredNamespace())
 136         msg.setArg(OPENID_NS, 'error', 'Unsupported type')
 137         msg.setArg(OPENID_NS, 'error_code', 'unsupported-type')
 138         msg.setArg(OPENID_NS, 'assoc_type', 'HMAC-SHA1')
 139         msg.setArg(OPENID_NS, 'session_type', 'DH-SHA1')
 140
 141         self.consumer.return_messages = [msg,
 142              Message(self.endpoint.preferredNamespace())]
 143
 144         self.failUnlessEqual(self.consumer._negotiateAssociation(self.endpoint), None)
 145
 146         self.failUnlessLogMatches('Unsupported association type',
 147                                   'Server %s refused' % (self.endpoint.server_url))
 148
 149     def testValid(self):
 150         """
 151         Test the valid case, wherein an association is returned on the
 152         first attempt to get one.
 153         """
 154         assoc = association.Association(
 155             'handle', 'secret', 'issued', 10000, 'HMAC-SHA1')
 156
 157         self.consumer.return_messages = [assoc]
 158         self.failUnless(self.consumer._negotiateAssociation(self.endpoint) is assoc)
 159         self.failUnlessLogEmpty()
 160
 161 class TestOpenID1SessionNegotiation(unittest.TestCase, CatchLogs):
 162     """
 163     Tests for the OpenID 1 consumer association session behavior.  See
 164     the docs for TestOpenID2SessionNegotiation.  Notice that this
 165     class is not a subclass of the OpenID 2 tests.  Instead, it uses
 166     many of the same inputs but inspects the log messages logged with
 167     oidutil.log.  See the calls to self.failUnlessLogMatches.  Some of
 168     these tests pass openid2-style messages to the openid 1
 169     association processing logic to be sure it ignores the extra data.
 170     """
 171     def setUp(self):
 172         CatchLogs.setUp(self)
 173         self.consumer = ErrorRaisingConsumer(store=None)
 174
 175         self.endpoint = OpenIDServiceEndpoint()
 176         self.endpoint.type_uris = [OPENID1_NS]
 177         self.endpoint.server_url = 'bogus'
 178
 179     def testBadResponse(self):
 180         self.consumer.return_messages = [Message(self.endpoint.preferredNamespace())]
 181         self.assertEqual(self.consumer._negotiateAssociation(self.endpoint), None)
 182         self.failUnlessLogMatches('Server error when requesting an association')
 183
 184     def testEmptyAssocType(self):
 185         msg = Message(self.endpoint.preferredNamespace())
 186         msg.setArg(OPENID_NS, 'error', 'Unsupported type')
 187         msg.setArg(OPENID_NS, 'error_code', 'unsupported-type')
 188         # not set: msg.setArg(OPENID_NS, 'assoc_type', None)
 189         msg.setArg(OPENID_NS, 'session_type', 'new-session-type')
 190
 191         self.consumer.return_messages = [msg]
 192         self.assertEqual(self.consumer._negotiateAssociation(self.endpoint), None)
 193
 194         self.failUnlessLogMatches('Server error when requesting an association')
 195
 196     def testEmptySessionType(self):
 197         msg = Message(self.endpoint.preferredNamespace())
 198         msg.setArg(OPENID_NS, 'error', 'Unsupported type')
 199         msg.setArg(OPENID_NS, 'error_code', 'unsupported-type')
 200         msg.setArg(OPENID_NS, 'assoc_type', 'new-assoc-type')
 201         # not set: msg.setArg(OPENID_NS, 'session_type', None)
 202
 203         self.consumer.return_messages = [msg]
 204         self.assertEqual(self.consumer._negotiateAssociation(self.endpoint), None)
 205
 206         self.failUnlessLogMatches('Server error when requesting an association')
 207
 208     def testNotAllowed(self):
 209         allowed_types = []
 210
 211         negotiator = association.SessionNegotiator(allowed_types)
 212         self.consumer.negotiator = negotiator
 213
 214         msg = Message(self.endpoint.preferredNamespace())
 215         msg.setArg(OPENID_NS, 'error', 'Unsupported type')
 216         msg.setArg(OPENID_NS, 'error_code', 'unsupported-type')
 217         msg.setArg(OPENID_NS, 'assoc_type', 'not-allowed')
 218         msg.setArg(OPENID_NS, 'session_type', 'not-allowed')
 219
 220         self.consumer.return_messages = [msg]
 221         self.assertEqual(self.consumer._negotiateAssociation(self.endpoint), None)
 222
 223         self.failUnlessLogMatches('Server error when requesting an association')
 224
 225     def testUnsupportedWithRetry(self):
 226         msg = Message(self.endpoint.preferredNamespace())
 227         msg.setArg(OPENID_NS, 'error', 'Unsupported type')
 228         msg.setArg(OPENID_NS, 'error_code', 'unsupported-type')
 229         msg.setArg(OPENID_NS, 'assoc_type', 'HMAC-SHA1')
 230         msg.setArg(OPENID_NS, 'session_type', 'DH-SHA1')
 231
 232         assoc = association.Association(
 233             'handle', 'secret', 'issued', 10000, 'HMAC-SHA1')
 234
 235         self.consumer.return_messages = [msg, assoc]
 236         self.failUnless(self.consumer._negotiateAssociation(self.endpoint) is None)
 237
 238         self.failUnlessLogMatches('Server error when requesting an association')
 239
 240     def testValid(self):
 241         assoc = association.Association(
 242             'handle', 'secret', 'issued', 10000, 'HMAC-SHA1')
 243
 244         self.consumer.return_messages = [assoc]
 245         self.failUnless(self.consumer._negotiateAssociation(self.endpoint) is assoc)
 246         self.failUnlessLogEmpty()
 247
 248 class TestNegotiatorBehaviors(unittest.TestCase, CatchLogs):
 249     def setUp(self):
 250         self.allowed_types = [
 251             ('HMAC-SHA1', 'no-encryption'),
 252             ('HMAC-SHA256', 'no-encryption'),
 253             ]
 254
 255         self.n = association.SessionNegotiator(self.allowed_types)
 256
 257     def testAddAllowedTypeNoSessionTypes(self):
 258         self.assertRaises(ValueError, self.n.addAllowedType, 'invalid')
 259
 260     def testAddAllowedTypeBadSessionType(self):
 261         self.assertRaises(ValueError, self.n.addAllowedType, 'assoc1', 'invalid')
 262
 263     def testAddAllowedTypeContents(self):
 264         assoc_type = 'HMAC-SHA1'
 265         self.failUnless(self.n.addAllowedType(assoc_type) is None)
 266
 267         for typ in association.getSessionTypes(assoc_type):
 268             self.failUnless((assoc_type, typ) in self.n.allowed_types)
 269
 270 if __name__ == '__main__':
 271     unittest.main()