Implement a whitelist mechanism to validate fetch entries

By default only allow HTTP(S) and (S)FTP URL

Author: kba

bagit.py

@@ -16,6 +16,7 @@
 import tempfile
 import unicodedata
 import warnings
+from fnmatch import fnmatch
 from collections import defaultdict
 from datetime import date
 from functools import partial
@@ -127,6 +128,7 @@ def find_locale_dir():
 
 CHECKSUM_ALGOS = hashlib.algorithms_guaranteed
 DEFAULT_CHECKSUMS = ["sha256", "sha512"]
+DEFAULT_FETCH_URL_WHITELIST = ["https://*", "http://*", "ftp://*", "sftp://"]
 
 #: Block size used when reading files for hashing:
 HASH_BLOCK_SIZE = 512 * 1024
@@ -140,7 +142,7 @@ def find_locale_dir():
 
 
 def make_bag(
-    bag_dir, bag_info=None, processes=1, checksums=None, checksum=None, encoding="utf-8"
+    bag_dir, bag_info=None, processes=1, checksums=None, checksum=None, encoding="utf-8", fetch_url_whitelist=None
 ):
     """
     Convert a given directory into a bag. You can pass in arbitrary
@@ -278,7 +280,7 @@ class Bag(object):
     valid_files = ["bagit.txt", "fetch.txt"]
     valid_directories = ["data"]
 
-    def __init__(self, path=None):
+    def __init__(self, path=None, fetch_url_whitelist=None):
         super(Bag, self).__init__()
         self.tags = {}
         self.info = {}
@@ -299,6 +301,7 @@ def __init__(self, path=None):
         self.normalized_manifest_names = {}
 
         self.algorithms = []
+        self.fetch_url_whitelist = DEFAULT_FETCH_URL_WHITELIST if fetch_url_whitelist is None else fetch_url_whitelist
         self.tag_file_name = None
         self.path = abspath(path)
         if path:
@@ -582,7 +585,7 @@ def files_to_be_fetched(self):
         local filename
         """
 
-        for url, file_size, filename in self.fetch_entries():
+        for _, _, filename in self.fetch_entries():
             yield filename
 
     def fetch_files_to_be_fetched(self):
@@ -593,20 +596,23 @@ def fetch_files_to_be_fetched(self):
         opener = build_opener(proxy_handler)
         user_agent = "bagit.py/%s (Python/%s)" % (VERSION, sys.version_info)
         for url, expected_size, filename in self.fetch_entries():
-            expected_size = int(expected_size)  # FIXME should be int in the first place
+            if not fnmatch_any(url, self.fetch_url_whitelist):
+                raise BagError(_("Malformed URL in fetch.txt: %s, matches none of the whitelisted URL patterns %s") % (url, self.fetch_url_whitelist))
+            expected_size = -1 if expected_size == '-' else int(expected_size)
             if filename in self.payload_files():
                 LOGGER.info(_("File already fetched: %s"), filename)
                 continue
             req = Request(url)
             req.add_header('User-Agent', user_agent)
             resp = opener.open(req)
             headers = resp.info()
-            if "content-length" not in headers:
-                LOGGER.warning(_("Server sent no content-length for <%s>"), url)
-            else:
-                content_length = int(headers['content-length'])
-                if content_length != expected_size:
-                    raise BagError(_("Inconsistent size of %s: Expected %s but Content-Length is %s") % (filename, expected_size, content_length))
+            if expected_size >= 0:
+                if "content-length" not in headers:
+                    LOGGER.warning(_("Server sent no content-length for <%s>"), url)
+                else:
+                    content_length = int(headers['content-length'])
+                    if content_length != expected_size:
+                        raise BagError(_("Inconsistent size of %s: Expected %s but Content-Length is %s") % (filename, expected_size, content_length))
             with open(join(self.path, filename), 'wb') as out:
                 read = 0
                 while True:
@@ -615,7 +621,7 @@ def fetch_files_to_be_fetched(self):
                         break
                     read += len(block)
                     out.write(block)
-                if read != expected_size:
+                if expected_size >= 0 and read != expected_size:
                     raise BagError(_("Inconsistent size of %s: Expected %s but received %s") % (filename, expected_size, read))
             LOGGER.info(_("Fetched %s from %s"), filename, url)
 
@@ -799,15 +805,10 @@ def validate_fetch(self):
         Raises `BagError` for errors and otherwise returns no value
         """
 
-        for url, file_size, filename in self.fetch_entries():
-            # fetch_entries will raise a BagError for unsafe filenames
-            # so at this point we will check only that the URL is minimally
-            # well formed:
-            parsed_url = urlparse(url)
-
-            # ensure url is a remote URL, not file://
-            if not all((parsed_url.scheme, parsed_url.netloc)):
-                raise BagError(_("Malformed URL in fetch.txt: %s") % url)
+        for url, expected_size, filename in self.fetch_entries():
+            # ensure url matches one of the allowed patterns
+            if not fnmatch_any(url, self.fetch_url_whitelist):
+                raise BagError(_("Malformed URL in fetch.txt: %s, matches none of the whitelisted URL patterns %s") % (url, self.fetch_url_whitelist))
 
     def _validate_contents(self, processes=1, fast=False, completeness_only=False):
         if fast and not self.has_oxum():
@@ -1450,6 +1451,12 @@ def generate_manifest_lines(filename, algorithms=DEFAULT_CHECKSUMS):
 
     return results
 
+# Return true if any of the pattern fnmatches a string
+def fnmatch_any(s, pats):
+    for pat in pats:
+        if fnmatch(s, pat):
+            return True
+    return False
 
 def _encode_filename(s):
     s = s.replace("\r", "%0D")

locale/bagit-python.pot

@@ -8,7 +8,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: PACKAGE VERSION\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2018-11-27 15:21+0100\n"
+"POT-Creation-Date: 2018-12-03 12:06+0100\n"
 "PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
 "Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
 "Language-Team: LANGUAGE <[email protected]>\n"
@@ -168,6 +168,10 @@ msgstr ""
 msgid "Path \"%(payload_file)s\" in \"%(source_file)s\" is unsafe"
 msgstr ""
 
+#, python-format
+msgid "Malformed URL in fetch.txt: %s, matches none of the whitelisted URL patterns %s"
+msgstr ""
+
 #, python-format
 msgid "File already fetched: %s"
 msgstr ""
@@ -225,10 +229,6 @@ msgstr ""
 msgid "Expected %s to contain \"bagit.txt\""
 msgstr ""
 
-#, python-format
-msgid "Malformed URL in fetch.txt: %s"
-msgstr ""
-
 msgid "Fast validation requires bag-info.txt to include Payload-Oxum"
 msgstr ""
 

test.py

@@ -1081,33 +1081,41 @@ def test_fetch_unsafe_payloads(self):
 
         self.assertEqual(expected_msg, str(cm.exception))
 
-    def test_fetch_malformed_url(self):
-        with open(j(self.tmpdir, "fetch.txt"), "w") as fetch_txt:
-            print(
-                "//photojournal.jpl.nasa.gov/jpeg/PIA21390.jpg - data/nasa/PIA21390.jpg",
-                file=fetch_txt,
-            )
-
-        self.bag.save(manifests=True)
-
-        expected_msg = (
-            "Malformed URL in fetch.txt: //photojournal.jpl.nasa.gov/jpeg/PIA21390.jpg"
-        )
-
-        with self.assertRaises(bagit.BagError) as cm:
+    def test_invalid_urls(self):
+        invalid_urls = [
+                '//photojournal.jpl.nasa.gov/jpeg/PIA21390.jpg',
+                'file://%s' % j(self.tmpdir, "mock_data"),
+                '../../../../../etc/passwd',
+        ]
+        for url in invalid_urls:
+            with open(j(self.tmpdir, "fetch.txt"), "w") as fetch_txt:
+                print("%s - data/mock_data" % url, file=fetch_txt)
+            with self.assertRaisesRegexp(bagit.BagError, "^Malformed URL in fetch.txt: %s" % url):
+                self.bag.validate_fetch()
+
+    def test_invalid_urls_whitelist(self):
+        self.bag.fetch_url_whitelist = [
+            'https://my.inst.edu/data/*.png'
+        ]
+        valid_urls = [
+            'https://my.inst.edu/data/foo.png'
+        ]
+        invalid_urls = [
+            'https://my.inst.edu/data/foo'
+            'https://my.inst.edu/robots.txt',
+            'http://my.inst.edu/data/foo',
+            'https://example.org',
+        ]
+        for url in invalid_urls:
+            with open(j(self.tmpdir, "fetch.txt"), "w") as fetch_txt:
+                print("%s - data/mock_data" % url, file=fetch_txt)
+            with self.assertRaisesRegexp(bagit.BagError, "^Malformed URL in fetch.txt: %s" % url):
+                self.bag.validate_fetch()
+        for url in valid_urls:
+            with open(j(self.tmpdir, "fetch.txt"), "w") as fetch_txt:
+                print("%s - data/mock_data" % url, file=fetch_txt)
             self.bag.validate_fetch()
 
-        self.assertEqual(expected_msg, str(cm.exception))
-
-    # FIXME: Won't work since file:// URLs are rejected
-    #  def test_fetching_payload_file(self):
-    #      with open(j(self.tmpdir, "mock_data"), "w") as mock_data:
-    #          print("Lorem ipsum dolor sit", file=mock_data)
-    #      with open(j(self.tmpdir, "fetch.txt"), "w") as fetch_txt:
-    #          print("file://%s 21 data/mock_data" % j(self.tmpdir, "mock_data"), file=fetch_txt)
-    #      self.bag.save(manifests=True)
-    #      self.bag.validate_fetch()
-
     def test_fetching_payload_file(self):
         test_payload = 'loc/2478433644_2839c5e8b8_o_d.jpg'
         with open(j(self.tmpdir, "fetch.txt"), "w") as fetch_txt: