mirror of
https://gitlab.com/fdroid/fdroidserver.git
synced 2024-11-14 11:00:10 +01:00
Merge branch 'fdroid-install' into 'master'
install: expand subcommand to be able to fetch F-Droid.apk and install it See merge request fdroid/fdroidserver!1546
This commit is contained in:
Hans-Christoph Steiner
commit
643d8da709
@ -109,8 +109,8 @@ __complete_gpgsign() {
|
||||
}
|
||||
|
||||
__complete_install() {
|
||||
opts="-v -q"
|
||||
lopts="--verbose --quiet --all"
|
||||
opts="-v -q -a -p -n -y"
|
||||
lopts="--verbose --quiet --all --privacy-mode --no --yes"
|
||||
case "${cur}" in
|
||||
-*)
|
||||
__complete_options
|
||||
|
||||
@ -28,6 +28,7 @@
|
||||
# common.py is imported by all modules, so do not import third-party
|
||||
# libraries here as they will become a requirement for all commands.
|
||||
|
||||
import copy
|
||||
import difflib
|
||||
from typing import List
|
||||
import git
|
||||
@ -60,6 +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, urlsplit, urlunparse
|
||||
from zipfile import ZipFile
|
||||
|
||||
import fdroidserver.metadata
|
||||
@ -435,6 +437,14 @@ def get_config():
|
||||
return config
|
||||
|
||||
|
||||
def get_cachedir():
|
||||
cachedir = config and config.get('cachedir')
|
||||
if cachedir and os.path.exists(cachedir):
|
||||
return Path(cachedir)
|
||||
else:
|
||||
return Path(tempfile.mkdtemp())
|
||||
|
||||
|
||||
def regsub_file(pattern, repl, path):
|
||||
with open(path, 'rb') as f:
|
||||
text = f.read()
|
||||
@ -609,6 +619,34 @@ 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)
|
||||
for mirror in appended:
|
||||
parsed = urlparse(mirror['url'])
|
||||
mirror['url'] = urlunparse(
|
||||
parsed._replace(path=os.path.join(parsed.path, filename))
|
||||
)
|
||||
return appended
|
||||
|
||||
|
||||
def file_entry(filename, hash_value=None):
|
||||
meta = {}
|
||||
meta["name"] = "/" + Path(filename).as_posix().split("/", 1)[1]
|
||||
@ -4620,3 +4658,75 @@ def _install_ndk(ndk):
|
||||
logging.info(
|
||||
_('Set NDK {release} ({version}) up').format(release=ndk, version=version)
|
||||
)
|
||||
|
||||
|
||||
FDROIDORG_MIRRORS = [
|
||||
{
|
||||
'isPrimary': True,
|
||||
'url': 'https://f-droid.org/repo',
|
||||
'dnsA': ['65.21.79.229', '136.243.44.143'],
|
||||
'dnsAAAA': ['2a01:4f8:212:c98::2', '2a01:4f9:3b:546d::2'],
|
||||
'worksWithoutSNI': True,
|
||||
},
|
||||
{
|
||||
'url': 'http://fdroidorg6cooksyluodepej4erfctzk7rrjpjbbr6wx24jh3lqyfwyd.onion/fdroid/repo'
|
||||
},
|
||||
{
|
||||
'url': 'http://dotsrccccbidkzg7oc7oj4ugxrlfbt64qebyunxbrgqhxiwj3nl6vcad.onion/fdroid/repo'
|
||||
},
|
||||
{
|
||||
'url': 'http://ftpfaudev4triw2vxiwzf4334e3mynz7osqgtozhbc77fixncqzbyoyd.onion/fdroid/repo'
|
||||
},
|
||||
{
|
||||
'url': 'http://lysator7eknrfl47rlyxvgeamrv7ucefgrrlhk7rouv3sna25asetwid.onion/pub/fdroid/repo'
|
||||
},
|
||||
{
|
||||
'url': 'http://mirror.ossplanetnyou5xifr6liw5vhzwc2g2fmmlohza25wwgnnaw65ytfsad.onion/fdroid/repo'
|
||||
},
|
||||
{'url': 'https://fdroid.tetaneutral.net/fdroid/repo', 'countryCode': 'FR'},
|
||||
{
|
||||
'url': 'https://ftp.agdsn.de/fdroid/repo',
|
||||
'countryCode': 'DE',
|
||||
"dnsA": ["141.30.235.39"],
|
||||
"dnsAAAA": ["2a13:dd85:b00:12::1"],
|
||||
"worksWithoutSNI": True,
|
||||
},
|
||||
{
|
||||
'url': 'https://ftp.fau.de/fdroid/repo',
|
||||
'countryCode': 'DE',
|
||||
"dnsA": ["131.188.12.211"],
|
||||
"dnsAAAA": ["2001:638:a000:1021:21::1"],
|
||||
"worksWithoutSNI": True,
|
||||
},
|
||||
{'url': 'https://ftp.gwdg.de/pub/android/fdroid/repo', 'countryCode': 'DE'},
|
||||
{
|
||||
'url': 'https://ftp.lysator.liu.se/pub/fdroid/repo',
|
||||
'countryCode': 'SE',
|
||||
"dnsA": ["130.236.254.251", "130.236.254.253"],
|
||||
"dnsAAAA": ["2001:6b0:17:f0a0::fb", "2001:6b0:17:f0a0::fd"],
|
||||
"worksWithoutSNI": True,
|
||||
},
|
||||
{'url': 'https://mirror.cyberbits.eu/fdroid/repo', 'countryCode': 'FR'},
|
||||
{
|
||||
'url': 'https://mirror.fcix.net/fdroid/repo',
|
||||
'countryCode': 'US',
|
||||
"dnsA": ["23.152.160.16"],
|
||||
"dnsAAAA": ["2620:13b:0:1000::16"],
|
||||
"worksWithoutSNI": True,
|
||||
},
|
||||
{'url': 'https://mirror.kumi.systems/fdroid/repo', 'countryCode': 'AT'},
|
||||
{'url': 'https://mirror.level66.network/fdroid/repo', 'countryCode': 'DE'},
|
||||
{'url': 'https://mirror.ossplanet.net/fdroid/repo', 'countryCode': 'TW'},
|
||||
{'url': 'https://mirrors.dotsrc.org/fdroid/repo', 'countryCode': 'DK'},
|
||||
{'url': 'https://opencolo.mm.fcix.net/fdroid/repo', 'countryCode': 'US'},
|
||||
{
|
||||
'url': 'https://plug-mirror.rcac.purdue.edu/fdroid/repo',
|
||||
'countryCode': 'US',
|
||||
"dnsA": ["128.211.151.252"],
|
||||
"dnsAAAA": ["2001:18e8:804:35::1337"],
|
||||
"worksWithoutSNI": True,
|
||||
},
|
||||
]
|
||||
FDROIDORG_FINGERPRINT = (
|
||||
'43238D512C1E5EB2D6569F4A3AFBF5523418B82E0A3ED1552770ABB9A9C9CCAB'
|
||||
)
|
||||
|
||||
@ -23,12 +23,15 @@ import urllib.parse
|
||||
|
||||
|
||||
class GithubApi:
|
||||
"""
|
||||
Warpper for some select calls to GitHub Json/REST API.
|
||||
"""Wrapper for some select calls to GitHub Json/REST API.
|
||||
|
||||
This class wraps some calls to api.github.com. This is not intended to be a
|
||||
general API wrapper. Instead it's purpose is to return pre-filtered and
|
||||
transformed data that's playing well with other fdroidserver functions.
|
||||
|
||||
With the GitHub API, the token is optional, but it has pretty
|
||||
severe rate limiting.
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, api_token, repo_path):
|
||||
@ -41,9 +44,10 @@ class GithubApi:
|
||||
def _req(self, url, data=None):
|
||||
h = {
|
||||
"Accept": "application/vnd.github+json",
|
||||
"Authorization": f"Bearer {self._api_token}",
|
||||
"X-GitHub-Api-Version": "2022-11-28",
|
||||
}
|
||||
if self._api_token:
|
||||
h["Authorization"] = f"Bearer {self._api_token}"
|
||||
return urllib.request.Request(
|
||||
url,
|
||||
headers=h,
|
||||
@ -65,6 +69,17 @@ class GithubApi:
|
||||
released_tags = self.list_released_tags()
|
||||
return [x for x in all_tags if x not in released_tags]
|
||||
|
||||
def get_latest_apk(self):
|
||||
req = self._req(
|
||||
f"https://api.github.com/repos/{self._repo_path}/releases/latest"
|
||||
)
|
||||
with urllib.request.urlopen(req) as resp: # nosec CWE-22 disable bandit warning
|
||||
assets = json.load(resp)['assets']
|
||||
for asset in assets:
|
||||
url = asset.get('browser_download_url')
|
||||
if url and url.endswith('.apk'):
|
||||
return url
|
||||
|
||||
def tag_exists(self, tag):
|
||||
"""
|
||||
Check if git tag is present on github.
|
||||
|
||||
@ -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):
|
||||
|
||||
@ -20,32 +20,275 @@
|
||||
import sys
|
||||
import os
|
||||
import glob
|
||||
from argparse import ArgumentParser
|
||||
import locale
|
||||
import logging
|
||||
import termios
|
||||
import tty
|
||||
|
||||
import defusedxml.ElementTree as XMLElementTree
|
||||
|
||||
from argparse import ArgumentParser, BooleanOptionalAction
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlencode, urlparse, urlunparse
|
||||
|
||||
from . import _
|
||||
from . import common
|
||||
from .common import SdkToolsPopen
|
||||
from . import common, github, index, net
|
||||
from .exception import FDroidException
|
||||
|
||||
config = None
|
||||
|
||||
DEFAULT_IPFS_GATEWAYS = ("https://gateway.ipfs.io/ipfs/",)
|
||||
MAVEN_CENTRAL_MIRRORS = [
|
||||
{
|
||||
"url": "https://repo1.maven.org/maven2/",
|
||||
"dnsA": ["199.232.16.209"],
|
||||
"worksWithoutSNI": True,
|
||||
},
|
||||
{
|
||||
"url": "https://repo.maven.apache.org/maven2/",
|
||||
"dnsA": ["199.232.16.215"],
|
||||
"worksWithoutSNI": True,
|
||||
},
|
||||
{
|
||||
"url": "https://maven-central-asia.storage-download.googleapis.com/maven2/",
|
||||
},
|
||||
{
|
||||
"url": "https://maven-central-eu.storage-download.googleapis.com/maven2/",
|
||||
},
|
||||
{
|
||||
"url": "https://maven-central.storage-download.googleapis.com/maven2/",
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
def download_apk(appid='org.fdroid.fdroid', privacy_mode=False):
|
||||
"""Download an APK from F-Droid via the first mirror that works."""
|
||||
url = urlunparse(
|
||||
urlparse(common.FDROIDORG_MIRRORS[0]['url'])._replace(
|
||||
query=urlencode({'fingerprint': common.FDROIDORG_FINGERPRINT})
|
||||
)
|
||||
)
|
||||
|
||||
data, _ignored = index.download_repo_index_v2(url)
|
||||
app = data.get('packages', dict()).get(appid)
|
||||
preferred_version = None
|
||||
for version in app['versions'].values():
|
||||
if not preferred_version:
|
||||
# if all else fails, use the first one
|
||||
preferred_version = version
|
||||
if not version.get('releaseChannels'):
|
||||
# prefer APK in default release channel
|
||||
preferred_version = version
|
||||
break
|
||||
print('skipping', version)
|
||||
|
||||
mirrors = common.append_filename_to_mirrors(
|
||||
preferred_version['file']['name'][1:], common.FDROIDORG_MIRRORS
|
||||
)
|
||||
ipfsCIDv1 = preferred_version['file'].get('ipfsCIDv1')
|
||||
if ipfsCIDv1:
|
||||
for gateway in DEFAULT_IPFS_GATEWAYS:
|
||||
mirrors.append({'url': os.path.join(gateway, ipfsCIDv1)})
|
||||
f = net.download_using_mirrors(mirrors)
|
||||
if f and os.path.exists(f):
|
||||
versionCode = preferred_version['manifest']['versionCode']
|
||||
f = Path(f)
|
||||
return str(f.rename(f.with_stem(f'{appid}_{versionCode}')).resolve())
|
||||
|
||||
|
||||
def download_fdroid_apk(privacy_mode=False): # pylint: disable=unused-argument
|
||||
"""Directly download the current F-Droid APK and verify it.
|
||||
|
||||
This downloads the "download button" link, which is the version
|
||||
that is best tested for new installs.
|
||||
|
||||
"""
|
||||
mirror = common.FDROIDORG_MIRRORS[0]
|
||||
mirror['url'] = urlunparse(urlparse(mirror['url'])._replace(path='F-Droid.apk'))
|
||||
return net.download_using_mirrors([mirror])
|
||||
|
||||
|
||||
def download_fdroid_apk_from_github(privacy_mode=False):
|
||||
"""Download F-Droid.apk from F-Droid's GitHub Releases."""
|
||||
if common.config and not privacy_mode:
|
||||
token = common.config.get('github_token')
|
||||
else:
|
||||
token = None
|
||||
gh = github.GithubApi(token, 'https://github.com/f-droid/fdroidclient')
|
||||
latest_apk = gh.get_latest_apk()
|
||||
return net.download_file(latest_apk)
|
||||
|
||||
|
||||
def download_fdroid_apk_from_ipns(privacy_mode=False):
|
||||
"""Download the F-Droid APK from an IPNS repo."""
|
||||
cid = 'k51qzi5uqu5dl4hbcksbdmplanu9n4hivnqsupqe6vzve1pdbeh418ssptldd3'
|
||||
mirrors = [
|
||||
{"url": f"https://ipfs.io/ipns/{cid}/F-Droid.apk"},
|
||||
]
|
||||
if not privacy_mode:
|
||||
mirrors.append({"url": f"https://{cid}.ipns.dweb.link/F-Droid.apk"})
|
||||
return net.download_using_mirrors(mirrors)
|
||||
|
||||
|
||||
def download_fdroid_apk_from_maven(privacy_mode=False):
|
||||
"""Download F-Droid.apk from Maven Central and official mirrors."""
|
||||
path = 'org/fdroid/fdroid/F-Droid'
|
||||
if privacy_mode:
|
||||
mirrors = MAVEN_CENTRAL_MIRRORS[:2] # skip the Google servers
|
||||
else:
|
||||
mirrors = MAVEN_CENTRAL_MIRRORS
|
||||
mirrors = common.append_filename_to_mirrors(
|
||||
os.path.join(path, 'maven-metadata.xml'), mirrors
|
||||
)
|
||||
metadata = net.download_using_mirrors(mirrors)
|
||||
version = XMLElementTree.parse(metadata).getroot().findall('*.//latest')[0].text
|
||||
mirrors = common.append_filename_to_mirrors(
|
||||
os.path.join(path, version, f'F-Droid-{version}.apk'), mirrors
|
||||
)
|
||||
return net.download_using_mirrors(mirrors)
|
||||
|
||||
|
||||
def install_fdroid_apk(privacy_mode=False):
|
||||
"""Download and install F-Droid.apk using all tricks we can muster.
|
||||
|
||||
By default, this first tries to fetch the official install APK
|
||||
which is offered when someone clicks the "download" button on
|
||||
https://f-droid.org/. Then it will try all the mirrors and
|
||||
methods until it gets something successful, or runs out of
|
||||
options.
|
||||
|
||||
There is privacy_mode which tries to download from mirrors first,
|
||||
so that this downloads from a mirror that has many different kinds
|
||||
of files available, thereby breaking the clear link to F-Droid.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None for success or the error message.
|
||||
|
||||
"""
|
||||
country_code = locale.getlocale()[0].split('_')[-1]
|
||||
if privacy_mode is None and country_code in ('CN', 'HK', 'IR', 'TM'):
|
||||
logging.warning(
|
||||
_('Privacy mode was enabled based on your locale ({country_code}).').format(
|
||||
country_code=country_code
|
||||
)
|
||||
)
|
||||
privacy_mode = True
|
||||
|
||||
if privacy_mode or not (common.config and common.config.get('jarsigner')):
|
||||
download_methods = [
|
||||
download_fdroid_apk_from_maven,
|
||||
download_fdroid_apk_from_ipns,
|
||||
download_fdroid_apk_from_github,
|
||||
]
|
||||
else:
|
||||
download_methods = [
|
||||
download_apk,
|
||||
download_fdroid_apk_from_maven,
|
||||
download_fdroid_apk_from_github,
|
||||
download_fdroid_apk_from_ipns,
|
||||
download_fdroid_apk,
|
||||
]
|
||||
for method in download_methods:
|
||||
try:
|
||||
f = method(privacy_mode=privacy_mode)
|
||||
break
|
||||
except Exception as e:
|
||||
logging.info(e)
|
||||
else:
|
||||
return _('F-Droid.apk could not be downloaded from any known source!')
|
||||
|
||||
if common.config and common.config.get('apksigner'):
|
||||
# TODO this should always verify, but that requires APK sig verification in Python #94
|
||||
logging.info(_('Verifying package {path} with apksigner.').format(path=f))
|
||||
common.verify_apk_signature(f)
|
||||
fingerprint = common.apk_signer_fingerprint(f)
|
||||
if fingerprint.upper() != common.FDROIDORG_FINGERPRINT:
|
||||
return _('{path} has the wrong fingerprint ({fingerprint})!').format(
|
||||
path=f, fingerprint=fingerprint
|
||||
)
|
||||
|
||||
if common.config and common.config.get('adb'):
|
||||
if devices():
|
||||
install_apks_to_devices([f])
|
||||
os.remove(f)
|
||||
else:
|
||||
os.remove(f)
|
||||
return _('No devices found for `adb install`! Please plug one in.')
|
||||
|
||||
|
||||
def devices():
|
||||
p = SdkToolsPopen(['adb', "devices"])
|
||||
"""Get the list of device serials for use with adb commands."""
|
||||
p = common.SdkToolsPopen(['adb', "devices"])
|
||||
if p.returncode != 0:
|
||||
raise FDroidException("An error occured when finding devices: %s" % p.output)
|
||||
lines = [line for line in p.output.splitlines() if not line.startswith('* ')]
|
||||
if len(lines) < 3:
|
||||
return []
|
||||
lines = lines[1:-1]
|
||||
return [line.split()[0] for line in lines]
|
||||
serials = list()
|
||||
for line in p.output.splitlines():
|
||||
columns = line.strip().split("\t", maxsplit=1)
|
||||
if len(columns) == 2:
|
||||
serial, status = columns
|
||||
if status == 'device':
|
||||
serials.append(serial)
|
||||
else:
|
||||
d = {'serial': serial, 'status': status}
|
||||
logging.warning(_('adb reports {serial} is "{status}"!'.format(**d)))
|
||||
return serials
|
||||
|
||||
|
||||
def install_apks_to_devices(apks):
|
||||
"""Install the list of APKs to all Android devices reported by `adb devices`."""
|
||||
for apk in apks:
|
||||
# Get device list each time to avoid device not found errors
|
||||
devs = devices()
|
||||
if not devs:
|
||||
raise FDroidException(_("No attached devices found"))
|
||||
logging.info(_("Installing %s...") % apk)
|
||||
for dev in devs:
|
||||
logging.info(
|
||||
_("Installing '{apkfilename}' on {dev}...").format(
|
||||
apkfilename=apk, dev=dev
|
||||
)
|
||||
)
|
||||
p = common.SdkToolsPopen(['adb', "-s", dev, "install", apk])
|
||||
fail = ""
|
||||
for line in p.output.splitlines():
|
||||
if line.startswith("Failure"):
|
||||
fail = line[9:-1]
|
||||
if not fail:
|
||||
continue
|
||||
|
||||
if fail == "INSTALL_FAILED_ALREADY_EXISTS":
|
||||
logging.warning(
|
||||
_('"{apkfilename}" is already installed on {dev}.').format(
|
||||
apkfilename=apk, dev=dev
|
||||
)
|
||||
)
|
||||
else:
|
||||
raise FDroidException(
|
||||
_("Failed to install '{apkfilename}' on {dev}: {error}").format(
|
||||
apkfilename=apk, dev=dev, error=fail
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def read_char():
|
||||
"""Read input from the terminal prompt one char at a time."""
|
||||
fd = sys.stdin.fileno()
|
||||
old_settings = termios.tcgetattr(fd)
|
||||
try:
|
||||
tty.setraw(fd)
|
||||
ch = sys.stdin.read(1)
|
||||
finally:
|
||||
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
||||
return ch
|
||||
|
||||
|
||||
def strtobool(val):
|
||||
"""Convert a localized string representation of truth to True or False."""
|
||||
return val.lower() in ('', 'y', 'yes', _('yes'), _('true')) # '' is pressing Enter
|
||||
|
||||
|
||||
def main():
|
||||
global config
|
||||
|
||||
# Parse command line...
|
||||
parser = ArgumentParser(
|
||||
usage="%(prog)s [options] [APPID[:VERCODE] [APPID[:VERCODE] ...]]"
|
||||
)
|
||||
@ -62,22 +305,61 @@ def main():
|
||||
default=False,
|
||||
help=_("Install all signed applications available"),
|
||||
)
|
||||
parser.add_argument(
|
||||
"-p",
|
||||
"--privacy-mode",
|
||||
action=BooleanOptionalAction,
|
||||
default=None,
|
||||
help=_("Download F-Droid.apk using mirrors that leak less to the network"),
|
||||
)
|
||||
parser.add_argument(
|
||||
"-y",
|
||||
"--yes",
|
||||
action="store_true",
|
||||
default=None,
|
||||
help=_("Automatic yes to all prompts."),
|
||||
)
|
||||
parser.add_argument(
|
||||
"-n",
|
||||
"--no",
|
||||
action="store_false",
|
||||
dest='yes',
|
||||
help=_("Automatic no to all prompts."),
|
||||
)
|
||||
options = common.parse_args(parser)
|
||||
|
||||
common.set_console_logging(options.verbose)
|
||||
logging.captureWarnings(True) # for SNIMissingWarning
|
||||
|
||||
common.get_config()
|
||||
|
||||
if not options.appid and not options.all:
|
||||
parser.error(
|
||||
_("option %s: If you really want to install all the signed apps, use --all")
|
||||
% "all"
|
||||
run_install = options.yes
|
||||
if options.yes is None and sys.stdout.isatty():
|
||||
print(
|
||||
_(
|
||||
'Would you like to download and install F-Droid.apk via adb? (YES/no)'
|
||||
),
|
||||
flush=True,
|
||||
)
|
||||
|
||||
config = common.read_config()
|
||||
answer = ''
|
||||
while True:
|
||||
in_char = read_char()
|
||||
if in_char == '\r': # Enter key
|
||||
break
|
||||
if not in_char.isprintable():
|
||||
sys.exit(1)
|
||||
answer += in_char
|
||||
run_install = strtobool(answer)
|
||||
if run_install:
|
||||
sys.exit(install_fdroid_apk(options.privacy_mode))
|
||||
sys.exit(1)
|
||||
|
||||
output_dir = 'repo'
|
||||
if not os.path.isdir(output_dir):
|
||||
logging.info(_("No signed output directory - nothing to do"))
|
||||
sys.exit(0)
|
||||
if (options.appid or options.all) and not os.path.isdir(output_dir):
|
||||
logging.error(_("No signed output directory - nothing to do"))
|
||||
# TODO prompt user if they want to download from f-droid.org
|
||||
sys.exit(1)
|
||||
|
||||
if options.appid:
|
||||
vercodes = common.read_pkg_args(options.appid, True)
|
||||
@ -99,45 +381,14 @@ def main():
|
||||
for appid, apk in apks.items():
|
||||
if not apk:
|
||||
raise FDroidException(_("No signed APK available for %s") % appid)
|
||||
install_apks_to_devices(apks.values())
|
||||
|
||||
else:
|
||||
elif options.all:
|
||||
apks = {
|
||||
common.publishednameinfo(apkfile)[0]: apkfile
|
||||
for apkfile in sorted(glob.glob(os.path.join(output_dir, '*.apk')))
|
||||
}
|
||||
|
||||
for appid, apk in apks.items():
|
||||
# Get device list each time to avoid device not found errors
|
||||
devs = devices()
|
||||
if not devs:
|
||||
raise FDroidException(_("No attached devices found"))
|
||||
logging.info(_("Installing %s...") % apk)
|
||||
for dev in devs:
|
||||
logging.info(
|
||||
_("Installing '{apkfilename}' on {dev}...").format(
|
||||
apkfilename=apk, dev=dev
|
||||
)
|
||||
)
|
||||
p = SdkToolsPopen(['adb', "-s", dev, "install", apk])
|
||||
fail = ""
|
||||
for line in p.output.splitlines():
|
||||
if line.startswith("Failure"):
|
||||
fail = line[9:-1]
|
||||
if not fail:
|
||||
continue
|
||||
|
||||
if fail == "INSTALL_FAILED_ALREADY_EXISTS":
|
||||
logging.warning(
|
||||
_('"{apkfilename}" is already installed on {dev}.').format(
|
||||
apkfilename=apk, dev=dev
|
||||
)
|
||||
)
|
||||
else:
|
||||
raise FDroidException(
|
||||
_("Failed to install '{apkfilename}' on {dev}: {error}").format(
|
||||
apkfilename=apk, dev=dev, error=fail
|
||||
)
|
||||
)
|
||||
install_apks_to_devices(apks.values())
|
||||
|
||||
logging.info('\n' + _('Finished'))
|
||||
|
||||
|
||||
@ -17,13 +17,20 @@
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import copy
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
import requests
|
||||
import tempfile
|
||||
import time
|
||||
import urllib
|
||||
import urllib3
|
||||
from requests.adapters import HTTPAdapter, Retry
|
||||
from requests.exceptions import ChunkedEncodingError
|
||||
|
||||
from . import _, common
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
HEADERS = {'User-Agent': 'F-Droid'}
|
||||
|
||||
@ -64,14 +71,88 @@ def download_file(url, local_filename=None, dldir='tmp', retries=3, backoff_fact
|
||||
f.write(chunk)
|
||||
f.flush()
|
||||
return local_filename
|
||||
except ChunkedEncodingError as err:
|
||||
except requests.exceptions.ChunkedEncodingError as err:
|
||||
if i == retries:
|
||||
raise err
|
||||
logging.warning('Download interrupted, retrying...')
|
||||
logger.warning('Download interrupted, retrying...')
|
||||
time.sleep(backoff_factor * 2**i)
|
||||
raise ValueError("retries must be >= 0")
|
||||
|
||||
|
||||
def download_using_mirrors(mirrors, local_filename=None):
|
||||
"""Try to download the file from any working mirror.
|
||||
|
||||
Download the file that all URLs in the mirrors list point to,
|
||||
trying all the tricks, starting with the most private methods
|
||||
first. The list of mirrors is converted into a list of mirror
|
||||
configurations to try, in order that the should be attempted.
|
||||
|
||||
This builds mirror_configs_to_try using all possible combos to
|
||||
try. If a mirror is marked with worksWithoutSNI: True, then this
|
||||
logic will try it twice: first without SNI, then again with SNI.
|
||||
|
||||
"""
|
||||
mirrors = common.parse_mirrors_config(mirrors)
|
||||
mirror_configs_to_try = []
|
||||
for mirror in mirrors:
|
||||
mirror_configs_to_try.append(mirror)
|
||||
if mirror.get('worksWithoutSNI'):
|
||||
m = copy.deepcopy(mirror)
|
||||
del m['worksWithoutSNI']
|
||||
mirror_configs_to_try.append(m)
|
||||
|
||||
if not local_filename:
|
||||
for mirror in mirrors:
|
||||
filename = urllib.parse.urlparse(mirror['url']).path.split('/')[-1]
|
||||
if filename:
|
||||
break
|
||||
if filename:
|
||||
local_filename = os.path.join(common.get_cachedir(), filename)
|
||||
else:
|
||||
local_filename = tempfile.mkstemp(prefix='fdroid-')
|
||||
|
||||
timeouts = (2, 10, 100)
|
||||
last_exception = None
|
||||
for timeout in timeouts:
|
||||
for mirror in mirror_configs_to_try:
|
||||
last_exception = None
|
||||
urllib3.util.ssl_.HAS_SNI = not mirror.get('worksWithoutSNI')
|
||||
try:
|
||||
# the stream=True parameter keeps memory usage low
|
||||
r = requests.get(
|
||||
mirror['url'],
|
||||
stream=True,
|
||||
allow_redirects=False,
|
||||
headers=HEADERS,
|
||||
# add jitter to the timeout to be less predictable
|
||||
timeout=timeout + random.randint(0, timeout), # nosec B311
|
||||
)
|
||||
if r.status_code != 200:
|
||||
raise requests.exceptions.HTTPError(r.status_code, response=r)
|
||||
with open(local_filename, 'wb') as f:
|
||||
for chunk in r.iter_content(chunk_size=1024):
|
||||
if chunk: # filter out keep-alive new chunks
|
||||
f.write(chunk)
|
||||
f.flush()
|
||||
return local_filename
|
||||
except (
|
||||
ConnectionError,
|
||||
requests.exceptions.ChunkedEncodingError,
|
||||
requests.exceptions.ConnectionError,
|
||||
requests.exceptions.ContentDecodingError,
|
||||
requests.exceptions.HTTPError,
|
||||
requests.exceptions.SSLError,
|
||||
requests.exceptions.StreamConsumedError,
|
||||
requests.exceptions.Timeout,
|
||||
requests.exceptions.UnrewindableBodyError,
|
||||
) as e:
|
||||
last_exception = e
|
||||
logger.debug(_('Retrying failed download: %s') % str(e))
|
||||
# if it hasn't succeeded by now, then give up and raise last exception
|
||||
if last_exception:
|
||||
raise last_exception
|
||||
|
||||
|
||||
def http_get(url, etag=None, timeout=600):
|
||||
"""Download the content from the given URL by making a GET request.
|
||||
|
||||
|
||||
@ -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__":
|
||||
|
||||
@ -36,6 +36,7 @@ if localmodule not in sys.path:
|
||||
sys.path.insert(0, localmodule)
|
||||
|
||||
import fdroidserver.index
|
||||
import fdroidserver.install
|
||||
import fdroidserver.signindex
|
||||
import fdroidserver.common
|
||||
import fdroidserver.metadata
|
||||
@ -2967,6 +2968,38 @@ 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'
|
||||
mirrors = [{'url': url}]
|
||||
self.assertEqual(
|
||||
[{'url': url + '/' + filename}],
|
||||
fdroidserver.common.append_filename_to_mirrors(filename, mirrors),
|
||||
)
|
||||
|
||||
def test_append_filename_to_mirrors_full(self):
|
||||
filename = 'test.apk'
|
||||
mirrors = fdroidserver.common.FDROIDORG_MIRRORS
|
||||
for mirror in fdroidserver.common.append_filename_to_mirrors(filename, mirrors):
|
||||
self.assertTrue(mirror['url'].endswith('/' + filename))
|
||||
|
||||
|
||||
APKS_WITH_JAR_SIGNATURES = (
|
||||
(
|
||||
|
||||
@ -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')
|
||||
|
||||
@ -5,8 +5,12 @@
|
||||
import inspect
|
||||
import os
|
||||
import sys
|
||||
import textwrap
|
||||
import unittest
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
localmodule = os.path.realpath(
|
||||
os.path.join(os.path.dirname(inspect.getfile(inspect.currentframe())), '..')
|
||||
)
|
||||
@ -14,18 +18,25 @@ print('localmodule: ' + localmodule)
|
||||
if localmodule not in sys.path:
|
||||
sys.path.insert(0, localmodule)
|
||||
|
||||
import fdroidserver.common
|
||||
import fdroidserver.install
|
||||
import fdroidserver
|
||||
from fdroidserver import common, install
|
||||
from fdroidserver.exception import BuildException, FDroidException
|
||||
|
||||
|
||||
class InstallTest(unittest.TestCase):
|
||||
'''fdroidserver/install.py'''
|
||||
|
||||
def tearDown(self):
|
||||
common.config = None
|
||||
|
||||
def test_devices(self):
|
||||
config = dict()
|
||||
fdroidserver.common.fill_config_defaults(config)
|
||||
fdroidserver.common.config = config
|
||||
try:
|
||||
config['adb'] = fdroidserver.common.find_sdk_tools_cmd('adb')
|
||||
except FDroidException as e:
|
||||
self.skipTest(f'Skipping test because: {e}')
|
||||
self.assertTrue(os.path.exists(config['adb']))
|
||||
self.assertTrue(os.path.isfile(config['adb']))
|
||||
devices = fdroidserver.install.devices()
|
||||
@ -33,6 +44,228 @@ class InstallTest(unittest.TestCase):
|
||||
for device in devices:
|
||||
self.assertIsInstance(device, str)
|
||||
|
||||
def test_devices_fail(self):
|
||||
common.config = dict()
|
||||
common.fill_config_defaults(common.config)
|
||||
common.config['adb'] = '/bin/false'
|
||||
with self.assertRaises(FDroidException):
|
||||
fdroidserver.install.devices()
|
||||
|
||||
def test_devices_fail_nonexistent(self):
|
||||
"""This is mostly just to document this strange difference in behavior"""
|
||||
common.config = dict()
|
||||
common.fill_config_defaults(common.config)
|
||||
common.config['adb'] = '/nonexistent'
|
||||
with self.assertRaises(BuildException):
|
||||
fdroidserver.install.devices()
|
||||
|
||||
@patch('fdroidserver.common.SdkToolsPopen')
|
||||
def test_devices_with_mock_none(self, mock_SdkToolsPopen):
|
||||
p = Mock()
|
||||
mock_SdkToolsPopen.return_value = p
|
||||
p.output = 'List of devices attached\n\n'
|
||||
p.returncode = 0
|
||||
common.config = dict()
|
||||
common.fill_config_defaults(common.config)
|
||||
self.assertEqual([], fdroidserver.install.devices())
|
||||
|
||||
@patch('fdroidserver.common.SdkToolsPopen')
|
||||
def test_devices_with_mock_one(self, mock_SdkToolsPopen):
|
||||
p = Mock()
|
||||
mock_SdkToolsPopen.return_value = p
|
||||
p.output = 'List of devices attached\n05995813\tdevice\n\n'
|
||||
p.returncode = 0
|
||||
common.config = dict()
|
||||
common.fill_config_defaults(common.config)
|
||||
self.assertEqual(['05995813'], fdroidserver.install.devices())
|
||||
|
||||
@patch('fdroidserver.common.SdkToolsPopen')
|
||||
def test_devices_with_mock_many(self, mock_SdkToolsPopen):
|
||||
p = Mock()
|
||||
mock_SdkToolsPopen.return_value = p
|
||||
p.output = textwrap.dedent(
|
||||
"""* daemon not running; starting now at tcp:5037
|
||||
* daemon started successfully
|
||||
List of devices attached
|
||||
RZCT809FTQM device
|
||||
05995813 device
|
||||
emulator-5556 device
|
||||
emulator-5554 unauthorized
|
||||
0a388e93 no permissions (missing udev rules? user is in the plugdev group); see [http://developer.android.com/tools/device.html]
|
||||
986AY133QL device
|
||||
09301JEC215064 device
|
||||
015d165c3010200e device
|
||||
4DCESKVGUC85VOTO device
|
||||
|
||||
"""
|
||||
)
|
||||
p.returncode = 0
|
||||
common.config = dict()
|
||||
common.fill_config_defaults(common.config)
|
||||
self.assertEqual(
|
||||
[
|
||||
'RZCT809FTQM',
|
||||
'05995813',
|
||||
'emulator-5556',
|
||||
'986AY133QL',
|
||||
'09301JEC215064',
|
||||
'015d165c3010200e',
|
||||
'4DCESKVGUC85VOTO',
|
||||
],
|
||||
fdroidserver.install.devices(),
|
||||
)
|
||||
|
||||
@patch('fdroidserver.common.SdkToolsPopen')
|
||||
def test_devices_with_mock_error(self, mock_SdkToolsPopen):
|
||||
p = Mock()
|
||||
mock_SdkToolsPopen.return_value = p
|
||||
p.output = textwrap.dedent(
|
||||
"""* daemon not running. starting it now on port 5037 *
|
||||
* daemon started successfully *
|
||||
** daemon still not running
|
||||
error: cannot connect to daemon
|
||||
"""
|
||||
)
|
||||
p.returncode = 0
|
||||
common.config = dict()
|
||||
common.fill_config_defaults(common.config)
|
||||
self.assertEqual([], fdroidserver.install.devices())
|
||||
|
||||
@patch('fdroidserver.common.SdkToolsPopen')
|
||||
def test_devices_with_mock_no_permissions(self, mock_SdkToolsPopen):
|
||||
p = Mock()
|
||||
mock_SdkToolsPopen.return_value = p
|
||||
p.output = textwrap.dedent(
|
||||
"""List of devices attached
|
||||
???????????????? no permissions
|
||||
"""
|
||||
)
|
||||
p.returncode = 0
|
||||
common.config = dict()
|
||||
common.fill_config_defaults(common.config)
|
||||
self.assertEqual([], fdroidserver.install.devices())
|
||||
|
||||
@patch('fdroidserver.common.SdkToolsPopen')
|
||||
def test_devices_with_mock_unauthorized(self, mock_SdkToolsPopen):
|
||||
p = Mock()
|
||||
mock_SdkToolsPopen.return_value = p
|
||||
p.output = textwrap.dedent(
|
||||
"""List of devices attached
|
||||
aeef5e4e unauthorized
|
||||
"""
|
||||
)
|
||||
p.returncode = 0
|
||||
common.config = dict()
|
||||
common.fill_config_defaults(common.config)
|
||||
self.assertEqual([], fdroidserver.install.devices())
|
||||
|
||||
@patch('fdroidserver.common.SdkToolsPopen')
|
||||
def test_devices_with_mock_no_permissions_with_serial(self, mock_SdkToolsPopen):
|
||||
p = Mock()
|
||||
mock_SdkToolsPopen.return_value = p
|
||||
p.output = textwrap.dedent(
|
||||
"""List of devices attached
|
||||
4DCESKVGUC85VOTO no permissions (missing udev rules? user is in the plugdev group); see [http://developer.android.com/tools/device.html]
|
||||
|
||||
"""
|
||||
)
|
||||
p.returncode = 0
|
||||
common.config = dict()
|
||||
common.fill_config_defaults(common.config)
|
||||
self.assertEqual([], fdroidserver.install.devices())
|
||||
|
||||
@staticmethod
|
||||
def _download_raise(privacy_mode):
|
||||
raise Exception('fake failed download')
|
||||
|
||||
@patch('fdroidserver.install.download_apk')
|
||||
@patch('fdroidserver.install.download_fdroid_apk')
|
||||
@patch('fdroidserver.install.download_fdroid_apk_from_github')
|
||||
@patch('fdroidserver.install.download_fdroid_apk_from_ipns')
|
||||
@patch('fdroidserver.install.download_fdroid_apk_from_maven')
|
||||
def test_install_fdroid_apk_privacy_mode_true(
|
||||
self, maven, ipns, github, download_fdroid_apk, download_apk
|
||||
):
|
||||
download_apk.side_effect = self._download_raise
|
||||
download_fdroid_apk.side_effect = self._download_raise
|
||||
github.side_effect = self._download_raise
|
||||
ipns.side_effect = self._download_raise
|
||||
maven.side_effect = self._download_raise
|
||||
fdroidserver.common.config = {'jarsigner': 'fakepath'}
|
||||
install.install_fdroid_apk(privacy_mode=True)
|
||||
download_apk.assert_not_called()
|
||||
download_fdroid_apk.assert_not_called()
|
||||
github.assert_called_once()
|
||||
ipns.assert_called_once()
|
||||
maven.assert_called_once()
|
||||
|
||||
@patch('fdroidserver.install.download_apk')
|
||||
@patch('fdroidserver.install.download_fdroid_apk')
|
||||
@patch('fdroidserver.install.download_fdroid_apk_from_github')
|
||||
@patch('fdroidserver.install.download_fdroid_apk_from_ipns')
|
||||
@patch('fdroidserver.install.download_fdroid_apk_from_maven')
|
||||
def test_install_fdroid_apk_privacy_mode_false(
|
||||
self, maven, ipns, github, download_fdroid_apk, download_apk
|
||||
):
|
||||
download_apk.side_effect = self._download_raise
|
||||
download_fdroid_apk.side_effect = self._download_raise
|
||||
github.side_effect = self._download_raise
|
||||
ipns.side_effect = self._download_raise
|
||||
maven.side_effect = self._download_raise
|
||||
fdroidserver.common.config = {'jarsigner': 'fakepath'}
|
||||
install.install_fdroid_apk(privacy_mode=False)
|
||||
download_apk.assert_called_once()
|
||||
download_fdroid_apk.assert_called_once()
|
||||
github.assert_called_once()
|
||||
ipns.assert_called_once()
|
||||
maven.assert_called_once()
|
||||
|
||||
@patch('fdroidserver.install.download_apk')
|
||||
@patch('fdroidserver.install.download_fdroid_apk')
|
||||
@patch('fdroidserver.install.download_fdroid_apk_from_github')
|
||||
@patch('fdroidserver.install.download_fdroid_apk_from_ipns')
|
||||
@patch('fdroidserver.install.download_fdroid_apk_from_maven')
|
||||
@patch('locale.getlocale', lambda: ('zh_CN', 'UTF-8'))
|
||||
def test_install_fdroid_apk_privacy_mode_locale_auto(
|
||||
self, maven, ipns, github, download_fdroid_apk, download_apk
|
||||
):
|
||||
download_apk.side_effect = self._download_raise
|
||||
download_fdroid_apk.side_effect = self._download_raise
|
||||
github.side_effect = self._download_raise
|
||||
ipns.side_effect = self._download_raise
|
||||
maven.side_effect = self._download_raise
|
||||
fdroidserver.common.config = {'jarsigner': 'fakepath'}
|
||||
install.install_fdroid_apk(privacy_mode=None)
|
||||
download_apk.assert_not_called()
|
||||
download_fdroid_apk.assert_not_called()
|
||||
github.assert_called_once()
|
||||
ipns.assert_called_once()
|
||||
maven.assert_called_once()
|
||||
|
||||
@patch('fdroidserver.net.download_using_mirrors', lambda m: 'testvalue')
|
||||
def test_download_fdroid_apk_smokecheck(self):
|
||||
self.assertEqual('testvalue', install.download_fdroid_apk())
|
||||
|
||||
@unittest.skipUnless(os.getenv('test_download_fdroid_apk'), 'requires net access')
|
||||
def test_download_fdroid_apk(self):
|
||||
f = install.download_fdroid_apk()
|
||||
self.assertTrue(Path(f).exists())
|
||||
|
||||
@unittest.skipUnless(os.getenv('test_download_fdroid_apk'), 'requires net access')
|
||||
def test_download_fdroid_apk_from_maven(self):
|
||||
f = install.download_fdroid_apk_from_maven()
|
||||
self.assertTrue(Path(f).exists())
|
||||
|
||||
@unittest.skipUnless(os.getenv('test_download_fdroid_apk'), 'requires net access')
|
||||
def test_download_fdroid_apk_from_ipns(self):
|
||||
f = install.download_fdroid_apk_from_ipns()
|
||||
self.assertTrue(Path(f).exists())
|
||||
|
||||
@unittest.skipUnless(os.getenv('test_download_fdroid_apk'), 'requires net access')
|
||||
def test_download_fdroid_apk_from_github(self):
|
||||
f = install.download_fdroid_apk_from_github()
|
||||
self.assertTrue(Path(f).exists())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
os.chdir(os.path.dirname(__file__))
|
||||
|
||||
@ -107,6 +107,7 @@ class NetTest(unittest.TestCase):
|
||||
self.assertTrue(os.path.exists(f))
|
||||
self.assertEqual('tmp/com.downloader.aegis-3175421.apk', f)
|
||||
|
||||
@patch.dict(os.environ, clear=True)
|
||||
def test_download_file_retries(self):
|
||||
server = RetryServer()
|
||||
f = net.download_file('http://localhost:%d/f.txt' % server.port)
|
||||
@ -114,6 +115,7 @@ class NetTest(unittest.TestCase):
|
||||
self.assertEqual(server.reply.split(b'\n\n')[1], Path(f).read_bytes())
|
||||
server.stop()
|
||||
|
||||
@patch.dict(os.environ, clear=True)
|
||||
def test_download_file_retries_not_forever(self):
|
||||
"""The retry logic should eventually exit with an error."""
|
||||
server = RetryServer(failures=5)
|
||||
@ -121,6 +123,28 @@ class NetTest(unittest.TestCase):
|
||||
net.download_file('http://localhost:%d/f.txt' % server.port)
|
||||
server.stop()
|
||||
|
||||
def test_download_using_mirrors_retries(self):
|
||||
server = RetryServer()
|
||||
f = net.download_using_mirrors(
|
||||
[
|
||||
'https://fake.com/f.txt', # 404 or 301 Redirect
|
||||
'https://httpbin.org/status/403',
|
||||
'https://httpbin.org/status/500',
|
||||
'http://localhost:1/f.txt', # ConnectionError
|
||||
'http://localhost:%d/' % server.port,
|
||||
],
|
||||
)
|
||||
# strip the HTTP headers and compare the reply
|
||||
self.assertEqual(server.reply.split(b'\n\n')[1], Path(f).read_bytes())
|
||||
server.stop()
|
||||
|
||||
def test_download_using_mirrors_retries_not_forever(self):
|
||||
"""The retry logic should eventually exit with an error."""
|
||||
server = RetryServer(failures=5)
|
||||
with self.assertRaises(requests.exceptions.ConnectionError):
|
||||
net.download_using_mirrors(['http://localhost:%d/' % server.port])
|
||||
server.stop()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
os.chdir(os.path.dirname(__file__))
|
||||
|
||||
@ -172,10 +172,6 @@ echo_header "run unit tests"
|
||||
|
||||
cd $WORKSPACE/tests
|
||||
for testcase in $WORKSPACE/tests/*.TestCase; do
|
||||
if [ $testcase == $WORKSPACE/tests/install.TestCase ]; then
|
||||
echo "skipping install.TestCase, its too troublesome in CI builds"
|
||||
continue
|
||||
fi
|
||||
if [ $(uname) != "Linux" ] && [ $testcase == $WORKSPACE/tests/nightly.TestCase ]; then
|
||||
echo "skipping nightly.TestCase, it currently only works GNU/Linux"
|
||||
continue
|
||||
@ -730,7 +726,7 @@ $fdroid scanner
|
||||
# run these to get their output, but the are not setup, so don't fail
|
||||
$fdroid build || true
|
||||
$fdroid import || true
|
||||
$fdroid install || true
|
||||
$fdroid install --no || true
|
||||
|
||||
|
||||
#------------------------------------------------------------------------------#
|
||||
|
||||
@ -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."""
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user