libs/openid/test/test_verifydisco.py

   1 import unittest
   2 from openid import message
   3 from openid.test.support import OpenIDTestMixin
   4 from openid.consumer import consumer
   5 from openid.test.test_consumer import TestIdRes
   6 from openid.consumer import discover
   7
   8 def const(result):
   9     """Return a function that ignores any arguments and just returns
  10     the specified result"""
  11     def constResult(*args, **kwargs):
  12         return result
  13
  14     return constResult
  15
  16 class DiscoveryVerificationTest(OpenIDTestMixin, TestIdRes):
  17     def failUnlessProtocolError(self, prefix, callable, *args, **kwargs):
  18         try:
  19             result = callable(*args, **kwargs)
  20         except consumer.ProtocolError, e:
  21             self.failUnless(
  22                 e[0].startswith(prefix),
  23                 'Expected message prefix %r, got message %r' % (prefix, e[0]))
  24         else:
  25             self.fail('Expected ProtocolError with prefix %r, '
  26                       'got successful return %r' % (prefix, result))
  27
  28     def test_openID1NoLocalID(self):
  29         endpoint = discover.OpenIDServiceEndpoint()
  30         endpoint.claimed_id = 'bogus'
  31
  32         msg = message.Message.fromOpenIDArgs({})
  33         self.failUnlessProtocolError(
  34             'Missing required field openid.identity',
  35             self.consumer._verifyDiscoveryResults, msg, endpoint)
  36         self.failUnlessLogEmpty()
  37
  38     def test_openID1NoEndpoint(self):
  39         msg = message.Message.fromOpenIDArgs({'identity':'snakes on a plane'})
  40         self.failUnlessRaises(RuntimeError,
  41                               self.consumer._verifyDiscoveryResults, msg)
  42         self.failUnlessLogEmpty()
  43
  44     def test_openID2NoOPEndpointArg(self):
  45         msg = message.Message.fromOpenIDArgs({'ns':message.OPENID2_NS})
  46         self.failUnlessRaises(KeyError,
  47                               self.consumer._verifyDiscoveryResults, msg)
  48         self.failUnlessLogEmpty()
  49
  50     def test_openID2LocalIDNoClaimed(self):
  51         msg = message.Message.fromOpenIDArgs({'ns':message.OPENID2_NS,
  52                                               'op_endpoint':'Phone Home',
  53                                               'identity':'Jose Lius Borges'})
  54         self.failUnlessProtocolError(
  55             'openid.identity is present without',
  56             self.consumer._verifyDiscoveryResults, msg)
  57         self.failUnlessLogEmpty()
  58
  59     def test_openID2NoLocalIDClaimed(self):
  60         msg = message.Message.fromOpenIDArgs({'ns':message.OPENID2_NS,
  61                                               'op_endpoint':'Phone Home',
  62                                               'claimed_id':'Manuel Noriega'})
  63         self.failUnlessProtocolError(
  64             'openid.claimed_id is present without',
  65             self.consumer._verifyDiscoveryResults, msg)
  66         self.failUnlessLogEmpty()
  67
  68     def test_openID2NoIdentifiers(self):
  69         op_endpoint = 'Phone Home'
  70         msg = message.Message.fromOpenIDArgs({'ns':message.OPENID2_NS,
  71                                               'op_endpoint':op_endpoint})
  72         result_endpoint = self.consumer._verifyDiscoveryResults(msg)
  73         self.failUnless(result_endpoint.isOPIdentifier())
  74         self.failUnlessEqual(op_endpoint, result_endpoint.server_url)
  75         self.failUnlessEqual(None, result_endpoint.claimed_id)
  76         self.failUnlessLogEmpty()
  77
  78     def test_openID2NoEndpointDoesDisco(self):
  79         op_endpoint = 'Phone Home'
  80         sentinel = discover.OpenIDServiceEndpoint()
  81         sentinel.claimed_id = 'monkeysoft'
  82         self.consumer._discoverAndVerify = const(sentinel)
  83         msg = message.Message.fromOpenIDArgs(
  84             {'ns':message.OPENID2_NS,
  85              'identity':'sour grapes',
  86              'claimed_id':'monkeysoft',
  87              'op_endpoint':op_endpoint})
  88         result = self.consumer._verifyDiscoveryResults(msg)
  89         self.failUnlessEqual(sentinel, result)
  90         self.failUnlessLogMatches('No pre-discovered')
  91
  92     def test_openID2MismatchedDoesDisco(self):
  93         mismatched = discover.OpenIDServiceEndpoint()
  94         mismatched.identity = 'nothing special, but different'
  95         mismatched.local_id = 'green cheese'
  96
  97         op_endpoint = 'Phone Home'
  98         sentinel = discover.OpenIDServiceEndpoint()
  99         sentinel.claimed_id = 'monkeysoft'
 100         self.consumer._discoverAndVerify = const(sentinel)
 101         msg = message.Message.fromOpenIDArgs(
 102             {'ns':message.OPENID2_NS,
 103              'identity':'sour grapes',
 104              'claimed_id':'monkeysoft',
 105              'op_endpoint':op_endpoint})
 106         result = self.consumer._verifyDiscoveryResults(msg, mismatched)
 107         self.failUnlessEqual(sentinel, result)
 108         self.failUnlessLogMatches('Error attempting to use stored',
 109                                   'Attempting discovery')
 110
 111     def test_openid2UsePreDiscovered(self):
 112         endpoint = discover.OpenIDServiceEndpoint()
 113         endpoint.local_id = 'my identity'
 114         endpoint.claimed_id = 'i am sam'
 115         endpoint.server_url = 'Phone Home'
 116         endpoint.type_uris = [discover.OPENID_2_0_TYPE]
 117
 118         msg = message.Message.fromOpenIDArgs(
 119             {'ns':message.OPENID2_NS,
 120              'identity':endpoint.local_id,
 121              'claimed_id':endpoint.claimed_id,
 122              'op_endpoint':endpoint.server_url})
 123         result = self.consumer._verifyDiscoveryResults(msg, endpoint)
 124         self.failUnless(result is endpoint)
 125         self.failUnlessLogEmpty()
 126
 127     def test_openid2UsePreDiscoveredWrongType(self):
 128         text = "verify failed"
 129
 130         endpoint = discover.OpenIDServiceEndpoint()
 131         endpoint.local_id = 'my identity'
 132         endpoint.claimed_id = 'i am sam'
 133         endpoint.server_url = 'Phone Home'
 134         endpoint.type_uris = [discover.OPENID_1_1_TYPE]
 135
 136         def discoverAndVerify(claimed_id, to_match_endpoints):
 137             self.failUnlessEqual(claimed_id, endpoint.claimed_id)
 138             for to_match in to_match_endpoints:
 139                 self.failUnlessEqual(claimed_id, to_match.claimed_id)
 140             raise consumer.ProtocolError(text)
 141
 142         self.consumer._discoverAndVerify = discoverAndVerify
 143
 144         msg = message.Message.fromOpenIDArgs(
 145             {'ns':message.OPENID2_NS,
 146              'identity':endpoint.local_id,
 147              'claimed_id':endpoint.claimed_id,
 148              'op_endpoint':endpoint.server_url})
 149
 150         try:
 151             r = self.consumer._verifyDiscoveryResults(msg, endpoint)
 152         except consumer.ProtocolError, e:
 153             # Should we make more ProtocolError subclasses?
 154             self.failUnless(str(e), text)
 155         else:
 156             self.fail("expected ProtocolError, %r returned." % (r,))
 157
 158         self.failUnlessLogMatches('Error attempting to use stored',
 159                                   'Attempting discovery')
 160
 161     def test_openid1UsePreDiscovered(self):
 162         endpoint = discover.OpenIDServiceEndpoint()
 163         endpoint.local_id = 'my identity'
 164         endpoint.claimed_id = 'i am sam'
 165         endpoint.server_url = 'Phone Home'
 166         endpoint.type_uris = [discover.OPENID_1_1_TYPE]
 167
 168         msg = message.Message.fromOpenIDArgs(
 169             {'ns':message.OPENID1_NS,
 170              'identity':endpoint.local_id})
 171         result = self.consumer._verifyDiscoveryResults(msg, endpoint)
 172         self.failUnless(result is endpoint)
 173         self.failUnlessLogEmpty()
 174
 175     def test_openid1UsePreDiscoveredWrongType(self):
 176         class VerifiedError(Exception): pass
 177
 178         def discoverAndVerify(claimed_id, _to_match):
 179             raise VerifiedError
 180
 181         self.consumer._discoverAndVerify = discoverAndVerify
 182
 183         endpoint = discover.OpenIDServiceEndpoint()
 184         endpoint.local_id = 'my identity'
 185         endpoint.claimed_id = 'i am sam'
 186         endpoint.server_url = 'Phone Home'
 187         endpoint.type_uris = [discover.OPENID_2_0_TYPE]
 188
 189         msg = message.Message.fromOpenIDArgs(
 190             {'ns':message.OPENID1_NS,
 191              'identity':endpoint.local_id})
 192
 193         self.failUnlessRaises(
 194             VerifiedError,
 195             self.consumer._verifyDiscoveryResults, msg, endpoint)
 196
 197         self.failUnlessLogMatches('Error attempting to use stored',
 198                                   'Attempting discovery')
 199
 200     def test_openid2Fragment(self):
 201         claimed_id = "http://unittest.invalid/"
 202         claimed_id_frag = claimed_id + "#fragment"
 203         endpoint = discover.OpenIDServiceEndpoint()
 204         endpoint.local_id = 'my identity'
 205         endpoint.claimed_id = claimed_id
 206         endpoint.server_url = 'Phone Home'
 207         endpoint.type_uris = [discover.OPENID_2_0_TYPE]
 208
 209         msg = message.Message.fromOpenIDArgs(
 210             {'ns':message.OPENID2_NS,
 211              'identity':endpoint.local_id,
 212              'claimed_id': claimed_id_frag,
 213              'op_endpoint': endpoint.server_url})
 214         result = self.consumer._verifyDiscoveryResults(msg, endpoint)
 215
 216         self.failUnlessEqual(result.local_id, endpoint.local_id)
 217         self.failUnlessEqual(result.server_url, endpoint.server_url)
 218         self.failUnlessEqual(result.type_uris, endpoint.type_uris)
 219
 220         self.failUnlessEqual(result.claimed_id, claimed_id_frag)
 221
 222         self.failUnlessLogEmpty()
 223
 224     def test_openid1Fallback1_0(self):
 225         claimed_id = 'http://claimed.id/'
 226         endpoint = None
 227         resp_mesg = message.Message.fromOpenIDArgs({
 228             'ns': message.OPENID1_NS,
 229             'identity': claimed_id})
 230         # Pass the OpenID 1 claimed_id this way since we're passing
 231         # None for the endpoint.
 232         resp_mesg.setArg(message.BARE_NS, 'openid1_claimed_id', claimed_id)
 233
 234         # We expect the OpenID 1 discovery verification to try
 235         # matching the discovered endpoint against the 1.1 type and
 236         # fall back to 1.0.
 237         expected_endpoint = discover.OpenIDServiceEndpoint()
 238         expected_endpoint.type_uris = [discover.OPENID_1_0_TYPE]
 239         expected_endpoint.local_id = None
 240         expected_endpoint.claimed_id = claimed_id
 241
 242         discovered_services = [expected_endpoint]
 243         self.consumer._discover = lambda *args: ('unused', discovered_services)
 244
 245         actual_endpoint = self.consumer._verifyDiscoveryResults(
 246             resp_mesg, endpoint)
 247         self.failUnless(actual_endpoint is expected_endpoint)
 248
 249 # XXX: test the implementation of _discoverAndVerify
 250
 251
 252 class TestVerifyDiscoverySingle(TestIdRes):
 253     # XXX: more test the implementation of _verifyDiscoverySingle
 254     def test_endpointWithoutLocalID(self):
 255         # An endpoint like this with no local_id is generated as a result of
 256         # e.g. Yadis discovery with no LocalID tag.
 257         endpoint = discover.OpenIDServiceEndpoint()
 258         endpoint.server_url = "http://localhost:8000/openidserver"
 259         endpoint.claimed_id = "http://localhost:8000/id/id-jo"
 260         to_match = discover.OpenIDServiceEndpoint()
 261         to_match.server_url = "http://localhost:8000/openidserver"
 262         to_match.claimed_id = "http://localhost:8000/id/id-jo"
 263         to_match.local_id = "http://localhost:8000/id/id-jo"
 264         result = self.consumer._verifyDiscoverySingle(endpoint, to_match)
 265         # result should always be None, raises exception on failure.
 266         self.failUnlessEqual(result, None)
 267         self.failUnlessLogEmpty()
 268
 269 if __name__ == '__main__':
 270     unittest.main()