mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-06 08:00:32 +02:00
saphidandClaude Opus 5.5
69f790b83a
- F-Droid: percent-encode repo file names (a '#' in one screenshot name broke the whole main repo); a bad image name drops that image, not the app. - Search: sources still fetching report 'loading' (UI says so and refreshes quietly); indexes warm up at server start; page-only SideQuest is not searched and appears as a 'Browse SideQuest' link instead of an error. - Browse (empty query) ranks VR, artwork and recent updates first; the F-Droid archive is off by default (old versions only). - Throttled sources fall back to their last cached copy; per-host message. - Curated GitHub list gains Open Saber Plus (MIT) with icon and screenshots. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
549 lines
23 KiB
Python
549 lines
23 KiB
Python
"""Signed F-Droid repositories. CLI: add|remove|list|search|download."""
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
from html.parser import HTMLParser
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import zipfile
|
|
|
|
if __package__ in (None, ''):
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from apk_sources import SourceError
|
|
import frame_host
|
|
from frame_apk_sign import _der_parts, _cert_key, der
|
|
from frame_catalog import _IndexReader, _reduce_index, _sha256
|
|
|
|
KIND = 'fdroid'
|
|
CACHE_VERSION = 2
|
|
_LOCK = threading.RLock()
|
|
# Published by the repository operators; a user repository without a pin uses TOFU.
|
|
FDROID_PIN = '43238d512c1e5eb2d6569f4a3afbf5523418b82e0a3ed1552770abb9a9c9ccab'
|
|
IZZY_PIN = '3bf0d6abfeae2f401707b6d966be743bf0eee49c2561b9ba39073711f628937a'
|
|
_DIGESTS = {
|
|
'608648016503040201': ('sha256', '3031300d060960864801650304020105000420'),
|
|
'608648016503040202': ('sha384', '3041300d060960864801650304020205000430'),
|
|
'608648016503040203': ('sha512', '3051300d060960864801650304020305000440'),
|
|
'2b0e03021a': ('sha1', '3021300906052b0e03021a05000414'),
|
|
}
|
|
|
|
|
|
def _fingerprint(value):
|
|
value = re.sub(r'[:\s]', '', value or '').lower()
|
|
if not re.fullmatch('[0-9a-f]{64}', value):
|
|
raise SourceError('fingerprint must be a SHA-256 certificate fingerprint (64 hex digits)')
|
|
return value
|
|
|
|
|
|
def _url(url, fingerprint=None):
|
|
if url.startswith('fdroidrepos://'):
|
|
url = 'https://' + url[len('fdroidrepos://'):]
|
|
p = urllib.parse.urlsplit(url)
|
|
if p.scheme != 'https' or not p.hostname or p.username or p.password or p.fragment:
|
|
raise SourceError('repository URL must use HTTPS without credentials or a fragment')
|
|
params = urllib.parse.parse_qs(p.query)
|
|
pins = params.pop('fingerprint', [])
|
|
if params or len(pins) > 1:
|
|
raise SourceError('only one fingerprint query parameter is supported')
|
|
pin = _fingerprint(fingerprint) if fingerprint else None
|
|
if pins:
|
|
linked = _fingerprint(pins[0])
|
|
if pin and pin != linked:
|
|
raise SourceError('conflicting fingerprints')
|
|
pin = linked
|
|
return urllib.parse.urlunsplit(('https', p.netloc.lower(), p.path.rstrip('/') + '/', '', '')), pin
|
|
|
|
|
|
def _child(base, name):
|
|
name = str(name).lstrip('/')
|
|
decoded = urllib.parse.unquote(name)
|
|
if not name or '\\' in decoded or any(x in ('.', '..') for x in decoded.split('/')):
|
|
raise SourceError('unsafe repository file name')
|
|
if '?' in decoded or urllib.parse.urlsplit(decoded).scheme:
|
|
raise SourceError('repository file is outside its repository')
|
|
url = urllib.parse.urljoin(base, urllib.parse.quote(decoded, safe='/')) # names may contain '#' or spaces
|
|
if not url.startswith(base) or urllib.parse.urlsplit(url).query or urllib.parse.urlsplit(url).fragment:
|
|
raise SourceError('repository file is outside its repository')
|
|
return url
|
|
|
|
|
|
class _HTTPSRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
if urllib.parse.urlsplit(newurl).scheme != 'https':
|
|
raise SourceError('refusing non-HTTPS redirect')
|
|
return super().redirect_request(req, fp, code, msg, headers, newurl)
|
|
|
|
|
|
def _fetch(url, path, maximum):
|
|
request = urllib.request.Request(url, headers={'User-Agent': 'FrameControl/1.0'})
|
|
with urllib.request.build_opener(_HTTPSRedirect()).open(request, timeout=60) as r, open(path, 'wb') as f:
|
|
total = 0
|
|
while True:
|
|
chunk = r.read(1 << 20)
|
|
if not chunk:
|
|
break
|
|
total += len(chunk)
|
|
if total > maximum:
|
|
raise SourceError('repository file exceeds size limit')
|
|
f.write(chunk)
|
|
|
|
|
|
def _children(item):
|
|
return _der_parts(item[1])
|
|
|
|
|
|
def _cms(data, content):
|
|
outer = _der_parts(data)
|
|
if len(outer) != 1:
|
|
raise ValueError('invalid CMS wrapper')
|
|
wrapper = _children(outer[0])
|
|
if wrapper[0][1].hex() != '2a864886f70d010702':
|
|
raise ValueError('not CMS SignedData')
|
|
fields = _children(_children(wrapper[1])[0])
|
|
certs = next(_children(f) for f in fields[3:] if f[0] == 0xa0)
|
|
signers = _children(fields[-1])
|
|
if len(signers) != 1:
|
|
raise ValueError('exactly one repository signer required')
|
|
signer = _children(signers[0])
|
|
sid = _children(signer[1])
|
|
matching = []
|
|
for cert in certs:
|
|
tbs = _children(_children(cert)[0])
|
|
offset = 1 if tbs[0][0] == 0xa0 else 0
|
|
if tbs[offset][1] == sid[1][1] and tbs[offset + 2][2] == sid[0][2]:
|
|
matching.append(cert[2])
|
|
if len(matching) != 1:
|
|
raise ValueError('missing or ambiguous signer certificate')
|
|
cert = matching[0]
|
|
digest, prefix = _DIGESTS[_children(signer[2])[0][1].hex()]
|
|
at, signed = 3, content
|
|
if signer[at][0] == 0xa0:
|
|
attrs = {}
|
|
for attr in _children(signer[at]):
|
|
pair = _children(attr)
|
|
oid = pair[0][1].hex()
|
|
if oid in attrs:
|
|
raise ValueError('duplicate CMS attribute')
|
|
attrs[oid] = _children(pair[1])
|
|
if attrs['2a864886f70d010904'][0][1] != hashlib.new(digest, content).digest():
|
|
raise ValueError('CMS content digest mismatch')
|
|
if attrs['2a864886f70d010903'][0][1].hex() != '2a864886f70d010701':
|
|
raise ValueError('unexpected CMS content type')
|
|
signed = der(0x31, signer[at][1])
|
|
at += 1
|
|
algorithm = _children(signer[at])[0][1].hex()
|
|
allowed = {'sha1': '2a864886f70d010105', 'sha256': '2a864886f70d01010b',
|
|
'sha384': '2a864886f70d01010c', 'sha512': '2a864886f70d01010d'}
|
|
if algorithm not in ('2a864886f70d010101', allowed[digest]):
|
|
raise ValueError('unsupported repository signature algorithm (RSA PKCS#1 required)')
|
|
n, e, _ = _cert_key(cert)
|
|
sig = signer[at + 1][1]
|
|
size = (n.bit_length() + 7) // 8
|
|
if not 256 <= size <= 1024 or n % 2 != 1 or not 3 <= e <= 0xffffffff or e % 2 != 1 or len(sig) != size or int.from_bytes(sig, 'big') >= n:
|
|
raise ValueError('invalid RSA signature/key size')
|
|
value = bytes.fromhex(prefix) + hashlib.new(digest, signed).digest()
|
|
expected = b'\0\1' + b'\xff' * (size - len(value) - 3) + b'\0' + value
|
|
if pow(int.from_bytes(sig, 'big'), e, n).to_bytes(size, 'big') != expected:
|
|
raise ValueError('repository RSA signature mismatch')
|
|
return hashlib.sha256(cert).hexdigest()
|
|
|
|
|
|
def _sections(data):
|
|
sections = []
|
|
for block in re.split(b'\r?\n\r?\n', data):
|
|
if not block:
|
|
continue
|
|
attrs = {}
|
|
for line in re.sub(b'\r?\n ', b'', block).splitlines():
|
|
key, value = line.decode('utf-8').split(': ', 1)
|
|
key = key.lower()
|
|
if key in attrs:
|
|
raise ValueError('duplicate manifest attribute')
|
|
attrs[key] = value
|
|
sections.append(attrs)
|
|
return sections
|
|
|
|
|
|
def _digest_check(attrs, suffix, content):
|
|
for label, digest in (('sha-512', 'sha512'), ('sha-384', 'sha384'), ('sha-256', 'sha256'), ('sha1', 'sha1'), ('sha-1', 'sha1')):
|
|
if label + suffix in attrs:
|
|
if base64.b64decode(attrs[label + suffix], validate=True) != hashlib.new(digest, content).digest():
|
|
raise ValueError('JAR digest mismatch')
|
|
return
|
|
raise ValueError('missing supported JAR digest')
|
|
|
|
|
|
def _jar(path, member, pin):
|
|
try:
|
|
with zipfile.ZipFile(path) as z:
|
|
names = z.namelist()
|
|
if len(names) > 64 or len(names) != len(set(names)) or any(
|
|
i.file_size > (1024 * 1024 if i.filename.upper().startswith('META-INF/') else 256 * 1024 * 1024)
|
|
for i in z.infolist()):
|
|
raise ValueError('duplicate or oversized JAR member')
|
|
blocks = [n for n in names if n.upper().startswith('META-INF/') and n.upper().endswith('.RSA')]
|
|
if len(blocks) != 1:
|
|
raise ValueError('exactly one RSA JAR signer required')
|
|
sf = z.read(blocks[0][:-4] + '.SF')
|
|
fingerprint = _cms(z.read(blocks[0]), sf)
|
|
if pin and fingerprint != pin:
|
|
raise ValueError('repository fingerprint mismatch')
|
|
manifest = z.read('META-INF/MANIFEST.MF')
|
|
_digest_check(_sections(sf)[0], '-digest-manifest', manifest)
|
|
entries = [s for s in _sections(manifest)[1:] if s.get('name') == member]
|
|
if len(entries) != 1:
|
|
raise ValueError('index is not uniquely signed')
|
|
content = z.read(member)
|
|
_digest_check(entries[0], '-digest', content)
|
|
return content, fingerprint
|
|
except (ValueError, KeyError, IndexError, StopIteration, RuntimeError, NotImplementedError, zipfile.BadZipFile) as e:
|
|
raise SourceError('invalid signed repository: ' + str(e)) from e
|
|
|
|
|
|
def _storage():
|
|
return frame_host.data_dir('apk-repos.json')
|
|
|
|
|
|
def _read():
|
|
try:
|
|
settings = json.loads(_storage().read_text())
|
|
if not isinstance(settings, dict) or not isinstance(settings.get('repos'), list) or not isinstance(settings.get('enabled'), dict):
|
|
raise ValueError('invalid settings structure')
|
|
return settings
|
|
except FileNotFoundError:
|
|
return {'repos': [], 'enabled': {}}
|
|
except (OSError, ValueError) as e:
|
|
raise SourceError('cannot read repository settings: ' + str(e)) from e
|
|
|
|
|
|
def _write(path, value):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, tmp = tempfile.mkstemp(dir=str(path.parent), suffix='.part')
|
|
try:
|
|
with os.fdopen(fd, 'w') as f:
|
|
json.dump(value, f, separators=(',', ':'))
|
|
os.replace(tmp, path)
|
|
finally:
|
|
if os.path.exists(tmp):
|
|
os.unlink(tmp)
|
|
|
|
|
|
def user_repos():
|
|
with _LOCK:
|
|
return _read()['repos']
|
|
|
|
|
|
def sources():
|
|
builtins = [('fdroid', 'F-Droid', 'https://f-droid.org/repo/', FDROID_PIN),
|
|
('fdroid-archive', 'F-Droid archive', 'https://f-droid.org/archive/', FDROID_PIN),
|
|
('izzyondroid', 'IzzyOnDroid', 'https://apt.izzysoft.de/fdroid/repo/', IZZY_PIN)]
|
|
settings = _read()
|
|
return [dict(id=i, kind=KIND, name=n, url=u, fingerprint=p, builtin=True,
|
|
# The archive only holds superseded versions; it clutters search unless asked for.
|
|
enabled=settings['enabled'].get(i, i != 'fdroid-archive'), trust='community')
|
|
for i, n, u, p in builtins] + settings['repos']
|
|
|
|
|
|
def _text(value):
|
|
if isinstance(value, dict):
|
|
return value.get('en-US') or next((v for v in value.values() if v), '')
|
|
return value or ''
|
|
|
|
|
|
class _PlainText(HTMLParser):
|
|
def __init__(self):
|
|
super().__init__(convert_charrefs=True)
|
|
self.parts, self.hidden = [], 0
|
|
|
|
def handle_starttag(self, tag, attrs):
|
|
if tag in ('script', 'style'):
|
|
self.hidden += 1
|
|
elif tag in ('br', 'p', 'div', 'li'):
|
|
self.parts.append(' ')
|
|
|
|
def handle_endtag(self, tag):
|
|
if tag in ('script', 'style'):
|
|
self.hidden = max(0, self.hidden - 1)
|
|
elif tag in ('p', 'div', 'li'):
|
|
self.parts.append(' ')
|
|
|
|
def handle_data(self, data):
|
|
if not self.hidden:
|
|
self.parts.append(data)
|
|
|
|
|
|
def _summary(value):
|
|
parser = _PlainText()
|
|
parser.feed(_text(value))
|
|
parser.close()
|
|
return ' '.join(''.join(parser.parts).split())
|
|
|
|
|
|
def _images(meta, base):
|
|
def url(file):
|
|
name = file.get('name') if isinstance(file, dict) else file
|
|
try:
|
|
return _child(base, name) if isinstance(name, str) and name else None
|
|
except SourceError:
|
|
return None # one odd image name mustn't hide the app
|
|
|
|
icon = url(_text(meta.get('icon')))
|
|
banner = url(_text(meta.get('featureGraphic')))
|
|
screenshots = []
|
|
groups = meta.get('screenshots') or {}
|
|
for device, legacy in (('phone', 'phoneScreenshots'), ('sevenInch', 'sevenInchScreenshots')):
|
|
files = _text(groups.get(device)) or _text(meta.get(legacy)) or []
|
|
for file in files if isinstance(files, list) else []:
|
|
image = url(file)
|
|
if image and image not in screenshots:
|
|
screenshots.append(image)
|
|
if len(screenshots) == 6:
|
|
break
|
|
if len(screenshots) == 6:
|
|
break
|
|
return {'icon': icon, 'banner': banner, 'screenshots': screenshots}
|
|
|
|
|
|
def _reduce(path, source):
|
|
compatible = _reduce_index(path)
|
|
result = {}
|
|
with open(path, encoding='utf-8') as f:
|
|
reader = _IndexReader(f)
|
|
for key in reader.members():
|
|
if key != 'packages':
|
|
reader.value()
|
|
continue
|
|
for pkg in reader.members():
|
|
item = reader.value()
|
|
if pkg not in compatible:
|
|
continue
|
|
meta = item.get('metadata', {})
|
|
files = {v['file'].get('name'): v for v in item.get('versions', {}).values()
|
|
if isinstance(v, dict) and isinstance(v.get('file'), dict)}
|
|
versions = []
|
|
for v in compatible[pkg]:
|
|
original = files[v['name']]
|
|
versions.append(dict(v, size=original['file'].get('size'), updated=_date(original.get('added'))))
|
|
versions.sort(key=lambda v: (v['version_code'], v['abis'] == ['arm64-v8a']), reverse=True)
|
|
latest = versions[0]
|
|
images = _images(meta, source['url'])
|
|
result[pkg] = dict(source=source['id'], id=pkg, package=pkg,
|
|
name=_text(meta.get('name')) or pkg, summary=_summary(meta.get('summary')),
|
|
icon=images['icon'], images=images, developer=_text(meta.get('authorName')) or None,
|
|
page=meta.get('webSite') or source['url'], vr=None, free=True,
|
|
license=meta.get('license'), downloadable=bool(latest.get('sha256')),
|
|
versions=versions, **{k: latest[k] for k in ('version', 'version_code', 'min_sdk', 'abis', 'size', 'updated')})
|
|
return result
|
|
|
|
|
|
def _date(value):
|
|
return time.strftime('%Y-%m-%d', time.gmtime(value / 1000)) if isinstance(value, (int, float)) else None
|
|
|
|
|
|
def _v1(content, path):
|
|
index = json.loads(content)
|
|
apps = {a['packageName']: a for a in index['apps']}
|
|
packages = {}
|
|
for pkg, builds in index['packages'].items():
|
|
app = apps.get(pkg, {})
|
|
localized = app.get('localized', {})
|
|
meta = {k: _text({locale: fields[k] for locale, fields in localized.items() if fields.get(k)}) or app.get(k)
|
|
for k in ('name', 'summary', 'license', 'webSite', 'authorName')}
|
|
for field in ('icon', 'featureGraphic', 'phoneScreenshots', 'sevenInchScreenshots'):
|
|
images = {}
|
|
for locale, fields in localized.items():
|
|
value = fields.get(field)
|
|
if not value:
|
|
continue
|
|
prefix = pkg + '/' + locale + '/'
|
|
if field.endswith('Screenshots'):
|
|
images[locale] = [{'name': prefix + field + '/' + name} for name in value[:6]]
|
|
else:
|
|
images[locale] = {'name': prefix + value}
|
|
meta[field] = images
|
|
if not meta['icon'] and app.get('icon'):
|
|
meta['icon'] = {'en-US': {'name': 'icons/' + app['icon']}}
|
|
versions = {}
|
|
for i, v in enumerate(builds):
|
|
versions[str(i)] = {'manifest': {'versionName': v.get('versionName'), 'versionCode': v['versionCode'],
|
|
'usesSdk': {'minSdkVersion': v.get('minSdkVersion', 1)}, 'nativecode': v.get('nativecode', [])},
|
|
'file': {'name': v['apkName'], 'sha256': v.get('hash') if v.get('hashType') == 'sha256' else None,
|
|
'size': v.get('size')}, 'added': v.get('added')}
|
|
packages[pkg] = {'metadata': meta, 'versions': versions}
|
|
path.write_text(json.dumps({'packages': packages}))
|
|
|
|
|
|
def _load(source, force=False):
|
|
if not re.fullmatch(r'[a-z0-9-]+', source['id']):
|
|
raise SourceError('invalid source id')
|
|
_url(source['url'], source.get('fingerprint'))
|
|
cache = frame_host.cache_dir('apk-sources', source['id'] + '.json')
|
|
with _LOCK:
|
|
if not force and cache.exists() and time.time() - cache.stat().st_mtime < 86400:
|
|
try:
|
|
saved = json.loads(cache.read_text())
|
|
if (saved.get('version') == CACHE_VERSION and saved.get('fingerprint') == source.get('fingerprint')
|
|
and saved.get('url') == source['url']):
|
|
return saved['apps'], saved['fingerprint']
|
|
except (OSError, ValueError, KeyError, AttributeError):
|
|
pass
|
|
cache.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
with tempfile.TemporaryDirectory(dir=str(cache.parent)) as tmp:
|
|
jar, raw = Path(tmp) / 'index.jar', Path(tmp) / 'index.json'
|
|
try:
|
|
_fetch(source['url'] + 'entry.jar', jar, 8 * 1024 * 1024)
|
|
except urllib.error.HTTPError as e:
|
|
if e.code not in (404, 410):
|
|
raise
|
|
_fetch(source['url'] + 'index-v1.jar', jar, 256 * 1024 * 1024)
|
|
content, pin = _jar(jar, 'index-v1.json', source.get('fingerprint'))
|
|
_v1(content, raw)
|
|
else:
|
|
content, pin = _jar(jar, 'entry.json', source.get('fingerprint'))
|
|
entry = json.loads(content)['index']
|
|
_fetch(_child(source['url'], entry['name']), raw, 512 * 1024 * 1024)
|
|
if _sha256(raw) != entry['sha256'] or (entry.get('size') is not None and raw.stat().st_size != entry['size']):
|
|
raise SourceError('index SHA-256 or size mismatch')
|
|
apps = _reduce(raw, source)
|
|
_write(cache, {'version': CACHE_VERSION, 'url': source['url'], 'fingerprint': pin, 'apps': apps})
|
|
return apps, pin
|
|
except SourceError:
|
|
raise
|
|
except (OSError, ValueError, KeyError, TypeError, IndexError) as e:
|
|
raise SourceError('cannot load repository: ' + str(e)) from e
|
|
|
|
|
|
def add_repo(url, fingerprint=None, name=None):
|
|
url, pin = _url(url, fingerprint)
|
|
with _LOCK:
|
|
settings = _read()
|
|
existing = next((s for s in settings['repos'] if s['url'] == url), None)
|
|
if existing:
|
|
if pin and pin != existing['fingerprint']:
|
|
raise SourceError('repository already has a different pinned fingerprint; remove it first')
|
|
pin = existing['fingerprint']
|
|
source = dict(id='fdroid-user-' + hashlib.sha256(url.encode()).hexdigest()[:20], kind=KIND,
|
|
name=name or (existing or {}).get('name') or urllib.parse.urlsplit(url).hostname,
|
|
url=url, builtin=False, enabled=True, trust='user', fingerprint=pin)
|
|
_, source['fingerprint'] = _load(source, force=True)
|
|
source['trust_on_first_use'] = existing.get('trust_on_first_use', False) if existing else pin is None
|
|
settings['repos'] = [s for s in settings['repos'] if s['id'] != source['id']] + [source]
|
|
_write(_storage(), settings)
|
|
return source
|
|
|
|
|
|
def remove_repo(source_id):
|
|
with _LOCK:
|
|
settings = _read()
|
|
if not any(s['id'] == source_id for s in settings['repos']):
|
|
raise SourceError('unknown user repository')
|
|
settings['repos'] = [s for s in settings['repos'] if s['id'] != source_id]
|
|
_write(_storage(), settings)
|
|
|
|
|
|
def set_enabled(source_id, enabled):
|
|
if not isinstance(enabled, bool):
|
|
raise SourceError('enabled must be a boolean')
|
|
with _LOCK:
|
|
settings = _read()
|
|
source = next((s for s in sources() if s['id'] == source_id), None)
|
|
if not source:
|
|
raise SourceError('unknown repository')
|
|
if source['builtin']:
|
|
settings['enabled'][source_id] = enabled
|
|
else:
|
|
for s in settings['repos']:
|
|
if s['id'] == source_id:
|
|
s['enabled'] = enabled
|
|
_write(_storage(), settings)
|
|
|
|
|
|
def search(source, query, limit=50):
|
|
if not source.get('enabled', True):
|
|
return []
|
|
apps, _ = _load(source)
|
|
words = query.casefold().split()
|
|
found = [a for a in apps.values() if all(w in (a['id'] + ' ' + a['name'] + ' ' + a['summary']).casefold() for w in words)]
|
|
found.sort(key=lambda a: (a['id'].casefold() != query.casefold(), a['name'].casefold()))
|
|
return [{k: v for k, v in a.items() if k != 'versions'} for a in found[:max(0, limit)]]
|
|
|
|
|
|
def details(source, entry_id):
|
|
if not source.get('enabled', True):
|
|
raise SourceError('repository is disabled')
|
|
apps, _ = _load(source)
|
|
if entry_id not in apps:
|
|
raise SourceError('app has no Lepton-compatible version in this repository')
|
|
return apps[entry_id]
|
|
|
|
|
|
def download(source, entry_id, version_code=None):
|
|
entry = details(source, entry_id)
|
|
version = next((v for v in entry['versions'] if version_code is None or str(v['version_code']) == str(version_code)), None)
|
|
if not version or not re.fullmatch('[0-9a-f]{64}', version.get('sha256') or ''):
|
|
raise SourceError('version is missing or has no SHA-256 digest')
|
|
sha = version['sha256']
|
|
path = frame_host.cache_dir('apk-sources', sha + '.apk')
|
|
try:
|
|
if not path.exists() or _sha256(path) != sha:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, tmp = tempfile.mkstemp(dir=str(path.parent), suffix='.part')
|
|
os.close(fd)
|
|
try:
|
|
_fetch(_child(source['url'], version['name']), tmp, 4 * 1024 ** 3)
|
|
if _sha256(tmp) != sha:
|
|
raise SourceError('APK SHA-256 mismatch; download discarded')
|
|
os.replace(tmp, path)
|
|
finally:
|
|
if os.path.exists(tmp):
|
|
os.unlink(tmp)
|
|
return {'apk': str(path), 'obb': [], 'sha256': sha, 'verified': True}
|
|
except OSError as e:
|
|
raise SourceError('cannot download APK: ' + str(e)) from e
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
sub = parser.add_subparsers(dest='command', required=True)
|
|
add = sub.add_parser('add')
|
|
add.add_argument('url')
|
|
add.add_argument('--fingerprint')
|
|
add.add_argument('--name')
|
|
sub.add_parser('list')
|
|
remove = sub.add_parser('remove')
|
|
remove.add_argument('source')
|
|
for command in ('search', 'download'):
|
|
p = sub.add_parser(command)
|
|
p.add_argument('source')
|
|
p.add_argument('query' if command == 'search' else 'package')
|
|
args = parser.parse_args()
|
|
try:
|
|
if args.command == 'add':
|
|
result = add_repo(args.url, args.fingerprint, args.name)
|
|
elif args.command == 'list':
|
|
result = sources()
|
|
elif args.command == 'remove':
|
|
result = remove_repo(args.source)
|
|
else:
|
|
source = next((s for s in sources() if s['id'] == args.source), None)
|
|
if not source:
|
|
raise SourceError('unknown repository id; use list')
|
|
result = search(source, args.query) if args.command == 'search' else download(source, args.package)
|
|
print(json.dumps(result, indent=2))
|
|
except SourceError as e:
|
|
parser.exit(1, 'error: ' + str(e) + '\n')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|