2 files changed, 90 insertions, 0 deletions

diff --git a/bitbake/lib/bb/fetch2/__init__.py b/bitbake/lib/bb/fetch2/__init__.py
index 36c955473a..68f65a97e0 100644
--- a/bitbake/lib/bb/fetch2/__init__.py
+++ b/bitbake/lib/bb/fetch2/__init__.py
@@ -61,6 +61,17 @@ class BBFetchException(Exception):
     def __str__(self):
          return self.msg
 
+class UntrustedUrl(BBFetchException):
+    """Exception raised when encountering a host not listed in BB_ALLOWED_NETWORKS"""
+    def __init__(self, url, message=''):
+        if message:
+            msg = message
+        else:
+            msg = "The URL: '%s' is not trusted and cannot be used" % url
+        self.url = url
+        BBFetchException.__init__(self, msg)
+        self.args = (url,)
+
 class MalformedUrl(BBFetchException):
     """Exception raised when encountering an invalid url"""
     def __init__(self, url, message=''):
@@ -852,6 +863,11 @@ def build_mirroruris(origud, mirrors, ld):
             newuri = uri_replace(ud, find, replace, replacements, ld)
             if not newuri or newuri in uris or newuri == origud.url:
                 continue
+
+            if not trusted_network(ld, newuri):
+                logger.debug(1, "Mirror %s not in the list of trusted networks, skipping" %  (newuri))
+                continue
+
             try:
                 newud = FetchData(newuri, ld)
                 newud.setup_localpath(ld)
@@ -972,6 +988,41 @@ def try_mirrors(d, origud, mirrors, check = False):
             return ret
     return None
 
+def trusted_network(d, url):
+    """
+    Use a trusted url during download if networking is enabled and
+    BB_ALLOWED_NETWORKS is set globally or for a specific recipe.
+    Note: modifies SRC_URI & mirrors.
+    """
+    if d.getVar('BB_NO_NETWORK', True) == "1":
+        return True
+
+    pkgname = d.expand(d.getVar('PN'))
+    trusted_hosts = d.getVarFlag('BB_ALLOWED_NETWORKS', pkgname)
+
+    if not trusted_hosts:
+        trusted_hosts = d.getVar('BB_ALLOWED_NETWORKS', True)
+
+    # Not enabled.
+    if not trusted_hosts:
+        return True
+
+    scheme, network, path, user, passwd, param = decodeurl(url)
+
+    if not network:
+        return True
+
+    network = network.lower()
+
+    for host in trusted_hosts.split(" "):
+        host = host.lower()
+        if host.startswith("*.") and ("." + network).endswith(host[1:]):
+            return True
+        if host == network:
+            return True
+
+    return False
+
 def srcrev_internal_helper(ud, d, name):
     """
     Return:
@@ -1530,6 +1581,8 @@ class Fetch(object):
                 firsterr = None
                 if not localpath and ((not verify_donestamp(ud, self.d)) or m.need_update(ud, self.d)):
                     try:
+                        if not trusted_network(self.d, ud.url):
+                            raise UntrustedUrl(ud.url)
                         logger.debug(1, "Trying Upstream")
                         m.download(ud, self.d)
                         if hasattr(m, "build_mirror_data"):
diff --git a/bitbake/lib/bb/tests/fetch.py b/bitbake/lib/bb/tests/fetch.py
index d56ef49948..d3f7b6ac6c 100644
--- a/bitbake/lib/bb/tests/fetch.py
+++ b/bitbake/lib/bb/tests/fetch.py
@@ -547,6 +547,43 @@ class FetcherNetworkTest(FetcherTest):
             os.chdir(os.path.dirname(self.unpackdir))
             fetcher.unpack(self.unpackdir)
 
+        def test_trusted_network(self):
+            # Ensure trusted_network returns False when the host IS in the list.
+            url = "git://Someserver.org/foo;rev=1"
+            self.d.setVar("BB_ALLOWED_NETWORKS", "server1.org someserver.org server2.org server3.org")
+            self.assertTrue(bb.fetch.trusted_network(self.d, url))
+
+        def test_wild_trusted_network(self):
+            # Ensure trusted_network returns true when the *.host IS in the list.
+            url = "git://Someserver.org/foo;rev=1"
+            self.d.setVar("BB_ALLOWED_NETWORKS", "server1.org *.someserver.org server2.org server3.org")
+            self.assertTrue(bb.fetch.trusted_network(self.d, url))
+
+        def test_prefix_wild_trusted_network(self):
+            # Ensure trusted_network returns true when the prefix matches *.host.
+            url = "git://git.Someserver.org/foo;rev=1"
+            self.d.setVar("BB_ALLOWED_NETWORKS", "server1.org *.someserver.org server2.org server3.org")
+            self.assertTrue(bb.fetch.trusted_network(self.d, url))
+
+        def test_two_prefix_wild_trusted_network(self):
+            # Ensure trusted_network returns true when the prefix matches *.host.
+            url = "git://something.git.Someserver.org/foo;rev=1"
+            self.d.setVar("BB_ALLOWED_NETWORKS", "server1.org *.someserver.org server2.org server3.org")
+            self.assertTrue(bb.fetch.trusted_network(self.d, url))
+
+        def test_untrusted_network(self):
+            # Ensure trusted_network returns False when the host is NOT in the list.
+            url = "git://someserver.org/foo;rev=1"
+            self.d.setVar("BB_ALLOWED_NETWORKS", "server1.org server2.org server3.org")
+            self.assertFalse(bb.fetch.trusted_network(self.d, url))
+
+        def test_wild_untrusted_network(self):
+            # Ensure trusted_network returns False when the host is NOT in the list.
+            url = "git://*.someserver.org/foo;rev=1"
+            self.d.setVar("BB_ALLOWED_NETWORKS", "server1.org server2.org server3.org")
+            self.assertFalse(bb.fetch.trusted_network(self.d, url))
+
+
 class URLHandle(unittest.TestCase):
 
     datatable = {