From 59fcfa5dec7240b0ca7bf21a3a1ed95a53a9cab3 Mon Sep 17 00:00:00 2001 From: Hans-Christoph Steiner Date: Mon, 4 Mar 2024 12:44:20 +0100 Subject: [PATCH] index: download_repo_index_v2() uses mirrors test_download_repo_index_v2_url_parsing is no longer needed, since all the things it tested are now handled in test_download_repo_index_v2 --- fdroidserver/common.py | 19 ++++++++++- fdroidserver/index.py | 32 +++++++++--------- tests/api.TestCase | 53 ++++++++++++++++++++--------- tests/common.TestCase | 17 ++++++++++ tests/index.TestCase | 76 ++++++++++++++++-------------------------- tests/testcommon.py | 3 ++ 6 files changed, 119 insertions(+), 81 deletions(-) diff --git a/fdroidserver/common.py b/fdroidserver/common.py index 5d54ac09..4bc70a2e 100644 --- a/fdroidserver/common.py +++ b/fdroidserver/common.py @@ -61,7 +61,7 @@ from base64 import urlsafe_b64encode from binascii import hexlify from datetime import datetime, timedelta, timezone from queue import Queue -from urllib.parse import urlparse, urlunparse +from urllib.parse import urlparse, urlsplit, urlunparse from zipfile import ZipFile import fdroidserver.metadata @@ -619,6 +619,23 @@ def parse_mirrors_config(mirrors): raise TypeError(_('only accepts strings, lists, and tuples')) +def get_mirrors(url, filename=None): + """Get list of dict entries for mirrors, appending filename if provided.""" + # TODO use cached index if it exists + if isinstance(url, str): + url = urlsplit(url) + + if url.netloc == 'f-droid.org': + mirrors = FDROIDORG_MIRRORS + else: + mirrors = parse_mirrors_config(url.geturl()) + + if filename: + return append_filename_to_mirrors(filename, mirrors) + else: + return mirrors + + def append_filename_to_mirrors(filename, mirrors): """Append the filename to all "url" entries in the mirrors dict.""" appended = copy.deepcopy(mirrors) diff --git a/fdroidserver/index.py b/fdroidserver/index.py index 5ca59662..d3f9d44e 100644 --- a/fdroidserver/index.py +++ b/fdroidserver/index.py @@ -1633,7 +1633,7 @@ def download_repo_index_v1(url_str, etag=None, verify_fingerprint=True, timeout= return index, new_etag -def download_repo_index_v2(url_str, etag=None, verify_fingerprint=True, timeout=600): +def download_repo_index_v2(url_str, etag=None, verify_fingerprint=True, timeout=None): """Download and verifies index v2 file, then returns its data. Downloads the repository index from the given :param url_str and @@ -1652,8 +1652,13 @@ def download_repo_index_v2(url_str, etag=None, verify_fingerprint=True, timeout= - The new eTag as returned by the HTTP request """ + etag # etag is unused but needs to be there to keep the same API as the earlier functions. + url = urllib.parse.urlsplit(url_str) + if timeout is not None: + logging.warning('"timeout" argument of download_repo_index_v2() is deprecated!') + fingerprint = None if verify_fingerprint: query = urllib.parse.parse_qs(url.query) @@ -1665,29 +1670,22 @@ def download_repo_index_v2(url_str, etag=None, verify_fingerprint=True, timeout= path = url.path.rsplit('/', 1)[0] else: path = url.path.rstrip('/') + url = urllib.parse.SplitResult(url.scheme, url.netloc, path, '', '') - url = urllib.parse.SplitResult(url.scheme, url.netloc, path + '/entry.jar', '', '') - download, new_etag = net.http_get(url.geturl(), etag, timeout) + mirrors = common.get_mirrors(url, 'entry.jar') + f = net.download_using_mirrors(mirrors) + entry, public_key, fingerprint = get_index_from_jar(f, fingerprint) - if download is None: - return None, new_etag - - # jarsigner is used to verify the JAR, it requires a file for input - with tempfile.TemporaryDirectory() as dirname: - with (Path(dirname) / 'entry.jar').open('wb') as fp: - fp.write(download) - fp.flush() - entry, public_key, fingerprint = get_index_from_jar(fp.name, fingerprint) - - name = entry['index']['name'] sha256 = entry['index']['sha256'] - url = urllib.parse.SplitResult(url.scheme, url.netloc, path + name, '', '') - index, _ignored = net.http_get(url.geturl(), None, timeout) + mirrors = common.get_mirrors(url, entry['index']['name'][1:]) + f = net.download_using_mirrors(mirrors) + with open(f, 'rb') as fp: + index = fp.read() if sha256 != hashlib.sha256(index).hexdigest(): raise VerificationException( _("SHA-256 of {url} does not match entry!").format(url=url) ) - return json.loads(index), new_etag + return json.loads(index), None def get_index_from_jar(jarfile, fingerprint=None, allow_deprecated=False): diff --git a/tests/api.TestCase b/tests/api.TestCase index 0dbaefd8..e3e66765 100755 --- a/tests/api.TestCase +++ b/tests/api.TestCase @@ -2,6 +2,7 @@ import inspect import os +import shutil import sys import unittest from unittest import mock @@ -14,6 +15,8 @@ if localmodule not in sys.path: sys.path.insert(0, localmodule) import fdroidserver +from fdroidserver import common, signindex +from testcommon import GP_FINGERPRINT, mkdtemp class ApiTest(unittest.TestCase): @@ -29,6 +32,18 @@ class ApiTest(unittest.TestCase): self.basedir = os.path.join(localmodule, 'tests') os.chdir(self.basedir) + self._td = mkdtemp() + self.testdir = self._td.name + + common.config = None + config = common.read_config() + config['jarsigner'] = common.find_sdk_tools_cmd('jarsigner') + common.config = config + signindex.config = config + + def tearDown(self): + self._td.cleanup() + def test_download_repo_index_no_fingerprint(self): with self.assertRaises(fdroidserver.VerificationException): fdroidserver.download_repo_index("http://example.org") @@ -67,23 +82,31 @@ class ApiTest(unittest.TestCase): ) self.assertEqual(index_url, etag_set_to_url) - @mock.patch('fdroidserver.net.http_get') - def test_download_repo_index_v2_url_parsing(self, mock_http_get): - """Test whether it is trying to download the right file - - This passes the URL back via the etag return value just as a - hack to check which URL was actually attempted. - - """ - mock_http_get.side_effect = lambda url, etag, timeout: (None, url) - repo_url = 'https://example.org/fdroid/repo' - entry_url = 'https://example.org/fdroid/repo/entry.jar' - index_url = 'https://example.org/fdroid/repo/index-v2.json' - for url in (repo_url, entry_url, index_url): - _ignored, etag_set_to_url = fdroidserver.download_repo_index_v2( + @mock.patch('fdroidserver.net.download_using_mirrors') + def test_download_repo_index_v2(self, mock_download_using_mirrors): + """Basically a copy of IndexTest.test_download_repo_index_v2""" + mock_download_using_mirrors.side_effect = lambda mirrors: os.path.join( + self.testdir, 'repo', os.path.basename(mirrors[0]['url']) + ) + os.chdir(self.testdir) + signindex.config['keystore'] = os.path.join(self.basedir, 'keystore.jks') + os.mkdir('repo') + shutil.copy(os.path.join(self.basedir, 'repo', 'entry.json'), 'repo') + shutil.copy(os.path.join(self.basedir, 'repo', 'index-v2.json'), 'repo') + signindex.sign_index('repo', 'entry.json') + repo_url = 'https://fake.url/fdroid/repo' + entry_url = 'https://fake.url/fdroid/repo/entry.jar' + index_url = 'https://fake.url/fdroid/repo/index-v2.json' + fingerprint_url = 'https://fake.url/fdroid/repo?fingerprint=' + GP_FINGERPRINT + slash_url = 'https://fake.url/fdroid/repo//?fingerprint=' + GP_FINGERPRINT + for url in (repo_url, entry_url, index_url, fingerprint_url, slash_url): + data, _ignored = fdroidserver.download_repo_index_v2( url, verify_fingerprint=False ) - self.assertEqual(entry_url, etag_set_to_url) + self.assertEqual(['repo', 'packages'], list(data)) + self.assertEqual( + 'My First F-Droid Repo Demo', data['repo']['name']['en-US'] + ) if __name__ == "__main__": diff --git a/tests/common.TestCase b/tests/common.TestCase index c4959cb5..c2e03243 100755 --- a/tests/common.TestCase +++ b/tests/common.TestCase @@ -2968,6 +2968,23 @@ class CommonTest(unittest.TestCase): knownapks.recordapk(fake_apk, default_date=datetime.now(timezone.utc)) self.assertEqual(knownapks.apks[fake_apk], now) + def test_get_mirrors_fdroidorg(self): + mirrors = fdroidserver.common.get_mirrors( + 'https://f-droid.org/repo', 'entry.jar' + ) + self.assertEqual( + 'https://f-droid.org/repo/entry.jar', + mirrors[0]['url'], + ) + + def test_get_mirrors_other(self): + self.assertEqual( + [{'url': 'https://example.com/fdroid/repo/index-v2.json'}], + fdroidserver.common.get_mirrors( + 'https://example.com/fdroid/repo', 'index-v2.json' + ), + ) + def test_append_filename_to_mirrors(self): filename = 'test.apk' url = 'https://example.com/fdroid/repo' diff --git a/tests/index.TestCase b/tests/index.TestCase index 0e23c71b..facc9e77 100755 --- a/tests/index.TestCase +++ b/tests/index.TestCase @@ -25,13 +25,10 @@ if localmodule not in sys.path: import fdroidserver from fdroidserver import common, index, publish, signindex, update -from testcommon import TmpCwd, mkdtemp, parse_args_for_test +from testcommon import GP_FINGERPRINT, TmpCwd, mkdtemp, parse_args_for_test from pathlib import Path -GP_FINGERPRINT = 'B7C2EEFD8DAC7806AF67DFCD92EB18126BC08312A7F2D6F3862E46013C7A6135' - - class Options: nosign = True pretty = False @@ -183,32 +180,11 @@ class IndexTest(unittest.TestCase): ilist = index.download_repo_index(url, verify_fingerprint=False) self.assertEqual(index_url, ilist[1]) # etag item used to return URL - @patch('fdroidserver.net.http_get') - def test_download_repo_index_v2_url_parsing(self, mock_http_get): - """Test whether it is trying to download the right file - - This passes the URL back via the etag return value just as a - hack to check which URL was actually attempted. - - """ - mock_http_get.side_effect = lambda url, etag, timeout: (None, url) - repo_url = 'https://fake.url/fdroid/repo' - entry_url = 'https://fake.url/fdroid/repo/entry.jar' - index_url = 'https://fake.url/fdroid/repo/index-v2.json' - fingerprint_url = 'https://fake.url/fdroid/repo?fingerprint=' + GP_FINGERPRINT - slash_url = 'https://fake.url/fdroid/repo//?fingerprint=' + GP_FINGERPRINT - for url in (repo_url, entry_url, index_url, fingerprint_url, slash_url): - ilist = index.download_repo_index_v2(url, verify_fingerprint=False) - self.assertEqual(entry_url, ilist[1]) # etag item used to return URL - - @patch('fdroidserver.net.http_get') - def test_download_repo_index_v2(self, mock_http_get): - def http_get_def(url, etag, timeout): # pylint: disable=unused-argument - f = os.path.basename(url) - with open(os.path.join(self.testdir, 'repo', f), 'rb') as fp: - return (fp.read(), 'fakeetag') - - mock_http_get.side_effect = http_get_def + @patch('fdroidserver.net.download_using_mirrors') + def test_download_repo_index_v2(self, mock_download_using_mirrors): + mock_download_using_mirrors.side_effect = lambda mirrors: os.path.join( + self.testdir, 'repo', os.path.basename(mirrors[0]['url']) + ) os.chdir(self.testdir) signindex.config['keystore'] = os.path.join(self.basedir, 'keystore.jks') os.mkdir('repo') @@ -223,15 +199,15 @@ class IndexTest(unittest.TestCase): for url in (repo_url, entry_url, index_url, fingerprint_url, slash_url): data, _ignored = index.download_repo_index_v2(url, verify_fingerprint=False) self.assertEqual(['repo', 'packages'], list(data.keys())) + self.assertEqual( + 'My First F-Droid Repo Demo', data['repo']['name']['en-US'] + ) - @patch('fdroidserver.net.http_get') - def test_download_repo_index_v2_bad_fingerprint(self, mock_http_get): - def http_get_def(url, etag, timeout): # pylint: disable=unused-argument - f = os.path.basename(url) - with open(os.path.join(self.testdir, 'repo', f), 'rb') as fp: - return (fp.read(), 'fakeetag') - - mock_http_get.side_effect = http_get_def + @patch('fdroidserver.net.download_using_mirrors') + def test_download_repo_index_v2_bad_fingerprint(self, mock_download_using_mirrors): + mock_download_using_mirrors.side_effect = lambda mirrors: os.path.join( + self.testdir, 'repo', os.path.basename(mirrors[0]['url']) + ) os.chdir(self.testdir) signindex.config['keystore'] = os.path.join(self.basedir, 'keystore.jks') os.mkdir('repo') @@ -243,22 +219,26 @@ class IndexTest(unittest.TestCase): with self.assertRaises(fdroidserver.exception.VerificationException): data, _ignored = index.download_repo_index_v2(bad_fp_url) - @patch('fdroidserver.net.http_get') - def test_download_repo_index_v2_entry_verify(self, mock_http_get): - def http_get_def(url, etag, timeout): # pylint: disable=unused-argument - return (b'not the entry.jar file contents', 'fakeetag') + @patch('fdroidserver.net.download_using_mirrors') + def test_download_repo_index_v2_entry_verify(self, mock_download_using_mirrors): + def download_using_mirrors_def(mirrors): + f = os.path.join(tempfile.mkdtemp(), os.path.basename(mirrors[0]['url'])) + Path(f).write_text('not the entry.jar file contents') + return f - mock_http_get.side_effect = http_get_def + mock_download_using_mirrors.side_effect = download_using_mirrors_def url = 'https://fake.url/fdroid/repo?fingerprint=' + GP_FINGERPRINT with self.assertRaises(fdroidserver.exception.VerificationException): data, _ignored = index.download_repo_index_v2(url) - @patch('fdroidserver.net.http_get') - def test_download_repo_index_v2_index_verify(self, mock_http_get): - def http_get_def(url, etag, timeout): # pylint: disable=unused-argument - return (b'not the index-v2.json file contents', 'fakeetag') + @patch('fdroidserver.net.download_using_mirrors') + def test_download_repo_index_v2_index_verify(self, mock_download_using_mirrors): + def download_using_mirrors_def(mirrors): + f = os.path.join(tempfile.mkdtemp(), os.path.basename(mirrors[0]['url'])) + Path(f).write_text('not the index-v2.json file contents') + return f - mock_http_get.side_effect = http_get_def + mock_download_using_mirrors.side_effect = download_using_mirrors_def os.chdir(self.testdir) signindex.config['keystore'] = os.path.join(self.basedir, 'keystore.jks') os.mkdir('repo') diff --git a/tests/testcommon.py b/tests/testcommon.py index f0fd11bd..edb54fb0 100644 --- a/tests/testcommon.py +++ b/tests/testcommon.py @@ -24,6 +24,9 @@ import unittest.mock from pathlib import Path +GP_FINGERPRINT = 'B7C2EEFD8DAC7806AF67DFCD92EB18126BC08312A7F2D6F3862E46013C7A6135' + + class TmpCwd: """Context-manager for temporarily changing the current working directory."""