Support delegated primaries

TODO.txt

@@ -1,3 +1,3 @@
 
- * delegation of authority for a domain
+ * replace secure_urlopen with the "requests" library
 

browserid/tests/support.py

@@ -57,6 +57,13 @@ def fetch_wellknown_file(hostname, verify=None):
     a hash of the hostname.  This lets us exercise all the crypto code
     while using predictable local values.
     """
+
+    if hostname == "redirect.org":
+        return {"authority": "delegated.org"}
+
+    if hostname == "infinite.org":
+        return {"authority": "infinite.org"}
+
     return {"public-key": get_keypair(hostname)[0]}
 
 

browserid/tests/test_verifiers.py

@@ -140,6 +140,12 @@ def test_well_known_doc_with_public_key(self):
         assertion = make_assertion("t@m.com", "http://e.com")
         self.assertTrue(self.verifier.verify(assertion))
 
+    @patch('browserid.wellknown.fetch_wellknown_file', fetch_wellknown_file)
+    def test_delegated_primary(self):
+        assertion = make_assertion("t@redirect.org", "http://persona.org",
+                issuer="delegated.org")
+        self.assertTrue(self.verifier.verify(assertion))
+
     @patch('browserid.wellknown.fetch_wellknown_file', fetch_wellknown_file)
     def test_audience_verification(self):
 

browserid/tests/test_wellknown.py

@@ -3,9 +3,11 @@
 from mock import Mock, patch
 from requests.exceptions import RequestException
 
-from browserid.wellknown import fetch_wellknown_file
+from browserid.wellknown import fetch_wellknown_file, WellKnownManager
 from browserid.errors import ConnectionError, InvalidIssuerError
 from browserid.tests.support import unittest
+from browserid.tests.support import (fetch_wellknown_file as
+        patched_wellknown_file)
 
 
 # Retrieved from browserid.org on April 3rd 2012
@@ -87,3 +89,35 @@ def test_well_known_doc_with_no_public_key(self):
     def test_successful_fetch(self):
         key = self._fetch('test.com', response_text=BROWSERID_PK)
         self.assertEquals(key, BROWSERID_PK_PY['public-key'])
+
+
+class TestIssuerValidation(unittest.TestCase):
+    def setUp(self):
+        self.wellknown = WellKnownManager()
+
+    def test_trusted_secondaries(self):
+        self.assertTrue(self.wellknown.is_issuer_valid('test.com', 'browserid.org'))
+        self.assertFalse(self.wellknown.is_issuer_valid('test.com',
+            'browserid.org', trusted_secondaries=[], max_hops=0))
+
+    def test_hostname_issuer(self):
+        self.assertTrue(self.wellknown.is_issuer_valid('test.com', 'test.com'))
+        self.assertFalse(self.wellknown.is_issuer_valid('abc.com', 'test.com',
+            max_hops=0))
+
+    @patch('browserid.wellknown.fetch_wellknown_file', patched_wellknown_file)
+    def test_delegated_primary(self):
+        self.assertTrue(self.wellknown.is_issuer_valid('redirect.org',
+            'delegated.org'))
+
+    def test_disabled_delegated_primary(self):
+        self.assertFalse(self.wellknown.is_issuer_valid('redirect.org',          
+            'delegated.org', max_hops=0))
+
+    @patch('browserid.wellknown.fetch_wellknown_file', patched_wellknown_file)
+    def test_infinite_delegated_primary_recursion(self):
+        self.assertFalse(self.wellknown.is_issuer_valid('infinite.org', None))
+        self.assertRaises(RuntimeError, self.wellknown.is_issuer_valid,
+            'infinite.org', None, max_hops=None)
+
+

browserid/verifiers/local.py

@@ -7,17 +7,13 @@
 
 from browserid import jwt
 from browserid.verifiers import Verifier
-from browserid.wellknown import WellKnownManager
+from browserid.wellknown import WellKnownManager, DEFAULT_TRUSTED_SECONDARIES
 from browserid.utils import  unbundle_certs_and_assertion
 from browserid.errors import (InvalidSignatureError,
                               ExpiredSignatureError,
                               UnsupportedCertChainError)
 
 
-DEFAULT_TRUSTED_SECONDARIES = ("browserid.org", "diresworb.org",
-                               "dev.diresworb.org")
-
-
 class LocalVerifier(Verifier):
     """Class for local verification of BrowserID identity assertions.
 
@@ -84,10 +80,11 @@ def verify(self, assertion, audience=None, now=None):
             # No point doing all that crypto if we're going to fail out anyway.
             email = certificates[-1].payload["principal"]["email"]
             root_issuer = certificates[0].payload["iss"]
-            if root_issuer not in self.trusted_secondaries:
-                if not email.endswith("@" + root_issuer):
-                    msg = "untrusted root issuer: %s" % (root_issuer,)
-                    raise InvalidSignatureError(msg)
+            provider = email.split('@')[-1]
+            if not self.wellknown_manager.is_issuer_valid(provider,
+                    root_issuer, self.trusted_secondaries):
+                msg = "untrusted root issuer: %s" % (root_issuer,)
+                raise InvalidSignatureError(msg)
 
             # Verify the entire chain of certificates.
             cert = self.verify_certificate_chain(certificates, now=now)

browserid/wellknown.py

@@ -13,7 +13,10 @@
 from browserid.errors import (ConnectionError,
                               InvalidIssuerError)
 
+DEFAULT_TRUSTED_SECONDARIES = ("browserid.org", "diresworb.org",
+                               "dev.diresworb.org")
 WELL_KNOWN_URL = "/.well-known/browserid"
+DEFAULT_MAX_HOPS = 5
 
 
 class WellKnownManager(object):
@@ -57,6 +60,46 @@ def get_key(self, hostname):
     def fetch_wellknown_file(self, hostname):
         return fetch_wellknown_file(hostname, verify=self.verify)
 
+    def is_issuer_valid(self, hostname, issuer, trusted_secondaries=None,
+            max_hops=DEFAULT_MAX_HOPS):
+        """
+        This method allows you to check if a hostname is valid for an issuer.
+
+        There are three methods where this will return True:
+        * The hostname is the issuer
+        * The hostname is in the list of trusted secondaries
+        * When the hostname delegates to the issuer
+
+        You can disable the check for delegated primaries by setting max_hops
+        to 0.
+        """
+        if hostname == issuer:
+            return True
+
+        if trusted_secondaries is None:
+            trusted_secondaries = DEFAULT_TRUSTED_SECONDARIES
+
+        if issuer in trusted_secondaries:
+            return True
+
+        if max_hops is not 0:
+            def _is_issuer_valid(hostname, issuer, current_hop=1,
+                    max_hops=DEFAULT_MAX_HOPS):
+                support = self[hostname]
+
+                if 'authority' in support and support['authority'] == issuer:
+                    return True
+
+                if max_hops is not None and current_hop >= max_hops:
+                    return False
+
+                return _is_issuer_valid(hostname, issuer,
+                        current_hop + 1, max_hops)
+
+            return _is_issuer_valid(hostname, issuer, max_hops=max_hops)
+
+        return False
+
 
 class FIFOCache(object):
     """A simple in-memory FIFO cache for BrowserID public keys.