mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-06 05:02:50 +02:00
Fix the follow-up review's findings
- APK reader: read members with a bounded read, since ZipFile.read inflates a member fully before trimming it to a forged declared size; the reference walk counts every entry it examines, dead ends and cycles included. - Titles: a path that exists under the root wins over stripping the archive prefix; the prefix is taken before a linked folder is staged elsewhere (another drive on Windows); the install dialog loads a fresh title list first and says so if it couldn't check. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
0530a6d045
commit
dfacf43e55
5 files changed
+85
-15
No files matched your search
@@ -4,6 +4,7 @@ import os
|
||||
import struct
|
||||
import sys
|
||||
import tempfile
|
||||
import tracemalloc
|
||||
import unittest
|
||||
import zipfile
|
||||
|
||||
@@ -144,6 +145,26 @@ class ApkInfo(unittest.TestCase):
|
||||
finally:
|
||||
frame_apk.MAX_MANIFEST = limit
|
||||
|
||||
def test_forged_sizes_dont_inflate_everything(self):
|
||||
# The central directory claims 1 byte; the deflated data holds 16 MB of zeros.
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as z:
|
||||
z.writestr('AndroidManifest.xml', bytes(16 * 1024**2))
|
||||
data = bytearray(buf.getvalue())
|
||||
for sig, field in ((b'PK\x01\x02', 24), (b'PK\x03\x04', 22)):
|
||||
at = data.index(sig)
|
||||
data[at + field:at + field + 4] = struct.pack('<I', 1)
|
||||
limit, frame_apk.MAX_MANIFEST = frame_apk.MAX_MANIFEST, 1024**2
|
||||
tracemalloc.start()
|
||||
try:
|
||||
with self.assertRaises(frame_apk.ApkError):
|
||||
self.read(bytes(data))
|
||||
peak = tracemalloc.get_traced_memory()[1]
|
||||
finally:
|
||||
tracemalloc.stop()
|
||||
frame_apk.MAX_MANIFEST = limit
|
||||
self.assertLess(peak, 8 * 1024**2)
|
||||
|
||||
def test_reference_cycles_and_fan_out_are_bounded(self):
|
||||
res = frame_apk.Resources(b'')
|
||||
ref = frame_apk.T_REF
|
||||
@@ -153,6 +174,16 @@ class ApkInfo(unittest.TestCase):
|
||||
res.entries = {i: [('', 0, ref, i + 1)] * 5 for i in range(1, 6)}
|
||||
res.entries[6] = [('', 0, frame_apk.T_STRING, 0)]
|
||||
self.assertEqual(len(res.values(1)), frame_apk.MAX_VALUES)
|
||||
# Forty references at each hop round a four-id cycle: millions of dead ends.
|
||||
looked = []
|
||||
|
||||
class Counting(dict):
|
||||
def get(self, key, default=None):
|
||||
looked.append(key)
|
||||
return dict.get(self, key, default)
|
||||
res.entries = Counting({i: [('', 0, ref, i % 4 + 1)] * 40 for i in range(1, 5)})
|
||||
self.assertEqual(res.values(1), [])
|
||||
self.assertLess(len(looked), frame_apk.MAX_STEPS + 10)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
@@ -153,6 +153,26 @@ class Targets(unittest.TestCase):
|
||||
with self.assertRaises(FrameError):
|
||||
frame_titles._choose(p, 'Game/../../outside.exe')
|
||||
|
||||
def test_root_relative_path_wins_over_archive_prefix(self):
|
||||
# Game/Game/A.exe and Game/A.exe: 'Game/A.exe' is a real path under the root Game/.
|
||||
p = self.plan({'Game/Game/A.exe': pe(0x8664), 'Game/A.exe': pe(0x8664)}, 'Game')
|
||||
self.assertEqual(p['unwrapped'], 'Game')
|
||||
frame_titles._choose(p, 'Game/A.exe')
|
||||
self.assertEqual(p['target'], 'Game/A.exe')
|
||||
|
||||
@unittest.skipIf(os.name == 'nt', 'needs symlinks')
|
||||
def test_prefix_is_taken_before_staging(self):
|
||||
# A folder with a link is staged into a temporary copy; the prefix still names
|
||||
# the folders stepped into in the original.
|
||||
d = self.tree({'Game/A.exe': pe(0x8664)})
|
||||
os.symlink('A.exe', os.path.join(d, 'Game', 'link.exe'))
|
||||
p = frame_titles.inspect(d, 'Game')
|
||||
try:
|
||||
self.assertEqual(p['unwrapped'], 'Game')
|
||||
self.assertTrue(p['work'])
|
||||
finally:
|
||||
frame_titles.discard(p)
|
||||
|
||||
|
||||
class Zips(unittest.TestCase):
|
||||
def setUp(self):
|
||||
|
||||
+15
-6
@@ -18,6 +18,7 @@ MAX_MANIFEST = 16 * 1024**2
|
||||
MAX_ARSC = 128 * 1024**2 # real ones are a few MB; the largest apps' tens of MB
|
||||
MAX_ICON = 8 * 1024**2
|
||||
MAX_VALUES = 256 # resolved values per reference, across all its hops
|
||||
MAX_STEPS = 4096 # entries examined per reference, dead ends and cycles included
|
||||
|
||||
|
||||
class ApkError(Exception):
|
||||
@@ -135,18 +136,20 @@ class Resources:
|
||||
resid = (pid << 24) | (tid << 16) | index
|
||||
self.entries.setdefault(resid, []).append((language, density, dtype, value))
|
||||
|
||||
def values(self, resid, depth=0, seen=frozenset()):
|
||||
"""[(language, density, type, data)] with references followed, at most
|
||||
MAX_VALUES of them and never round a cycle."""
|
||||
def values(self, resid, depth=0, seen=frozenset(), steps=None):
|
||||
"""[(language, density, type, data)] with references followed: never round a
|
||||
cycle, at most MAX_VALUES results and MAX_STEPS entries examined in all."""
|
||||
steps = steps if steps is not None else [MAX_STEPS]
|
||||
out = []
|
||||
seen = seen | {resid}
|
||||
for lang, dens, dtype, value in self.entries.get(resid, []):
|
||||
if len(out) >= MAX_VALUES:
|
||||
steps[0] -= 1
|
||||
if steps[0] < 0 or len(out) >= MAX_VALUES:
|
||||
break
|
||||
if dtype == T_REF and depth < 5:
|
||||
if value not in seen:
|
||||
out += [(lang or l2, dens or d2, t2, v2)
|
||||
for l2, d2, t2, v2 in self.values(value, depth + 1, seen)]
|
||||
for l2, d2, t2, v2 in self.values(value, depth + 1, seen, steps)]
|
||||
else:
|
||||
out.append((lang, dens, dtype, value))
|
||||
return out[:MAX_VALUES]
|
||||
@@ -184,10 +187,16 @@ def _icons(attr, res):
|
||||
|
||||
|
||||
def _read(z, name, limit):
|
||||
"""A member's bytes, inflating at most limit + 1 of them whatever its header claims
|
||||
(ZipFile.read inflates everything first, then trims to the declared size)."""
|
||||
size = z.getinfo(name).file_size
|
||||
if size > limit:
|
||||
raise ApkError(f'{name} in the APK is {size / 1024**2:.0f} MB, more than a real one ({limit // 1024**2} MB)')
|
||||
return z.read(name)
|
||||
with z.open(name) as f:
|
||||
data = f.read(limit + 1)
|
||||
if len(data) > limit:
|
||||
raise ApkError(f'{name} in the APK is larger than a real one ({limit // 1024**2} MB)')
|
||||
return data
|
||||
|
||||
|
||||
def apk_info(path):
|
||||
|
||||
+15
-7
@@ -321,6 +321,13 @@ def _stage_folder(src, dest):
|
||||
return dest
|
||||
|
||||
|
||||
def _below(top, root):
|
||||
"""The folders _unwrap stepped through from top to root, as 'a/b', or ''. Taken
|
||||
before staging moves root elsewhere (possibly another drive on Windows)."""
|
||||
rel = os.path.relpath(root, os.path.realpath(top)).replace(os.sep, '/')
|
||||
return '' if rel == '.' else rel
|
||||
|
||||
|
||||
def _unwrap(root):
|
||||
"""Step into a single top-level folder, the usual shape of a zipped build.
|
||||
|
||||
@@ -424,8 +431,8 @@ def inspect(path, name=None):
|
||||
work = None
|
||||
try:
|
||||
if os.path.isdir(path):
|
||||
base = path
|
||||
root = _unwrap(path)
|
||||
unwrapped = _below(path, root)
|
||||
if _has_links(root):
|
||||
# scp -r follows links, so a link out of the folder could upload
|
||||
# anything; copy the folder with its links made safe first.
|
||||
@@ -434,24 +441,22 @@ def inspect(path, name=None):
|
||||
elif path.lower().endswith('.zip'):
|
||||
work = tempfile.mkdtemp(prefix=f'{TMP_PREFIX}{os.getpid()}-')
|
||||
extract_zip(path, work)
|
||||
base = work
|
||||
root = _unwrap(work)
|
||||
unwrapped = _below(work, root)
|
||||
elif classify(path):
|
||||
# A single executable is uploaded on its own; don't copy a whole Downloads folder.
|
||||
work = tempfile.mkdtemp(prefix=f'{TMP_PREFIX}{os.getpid()}-')
|
||||
shutil.copy2(path, os.path.join(work, os.path.basename(path)))
|
||||
base = root = work
|
||||
root, unwrapped = work, ''
|
||||
else:
|
||||
raise FrameError(f'{os.path.basename(path)} is not a .zip, a folder or a program')
|
||||
title = name or display_name(os.path.basename(path.rstrip('/\\')))
|
||||
found = candidates(root, title)
|
||||
if not found:
|
||||
raise FrameError(f'no Linux or Windows program found in {os.path.basename(path)}')
|
||||
# An exe path may be given as it is in the archive, above the folder _unwrap stepped into.
|
||||
unwrapped = os.path.relpath(root, os.path.realpath(base)).replace(os.sep, '/')
|
||||
plan = {'source': os.path.basename(path.rstrip('/\\')), 'name': title, 'id': title_id(title),
|
||||
'root': root, 'work': work, 'candidates': found,
|
||||
'unwrapped': '' if unwrapped == '.' else unwrapped,
|
||||
'unwrapped': unwrapped,
|
||||
'size': _tree_size(root), 'warnings': []}
|
||||
_choose(plan, found[0]['path'])
|
||||
return plan
|
||||
@@ -474,8 +479,11 @@ def _tree_size(root):
|
||||
def _choose(plan, rel, runtime=None):
|
||||
"""Set plan's launch target (a path relative to root) and its runtime."""
|
||||
rel = rel.replace('\\', '/')
|
||||
# A manifest may name the program as it is in the archive, above the folder
|
||||
# _unwrap stepped into. A path that works as it is always wins.
|
||||
prefix = plan.get('unwrapped') and plan['unwrapped'] + '/'
|
||||
if prefix and rel.startswith(prefix):
|
||||
if (prefix and rel.startswith(prefix) and not any(c['path'] == rel for c in plan['candidates'])
|
||||
and not os.path.isfile(os.path.join(plan['root'], *rel.split('/')))):
|
||||
rel = rel[len(prefix):]
|
||||
target = next((c for c in plan['candidates'] if c['path'] == rel), None)
|
||||
if target is None:
|
||||
|
||||
+4
-2
@@ -1200,6 +1200,7 @@ async function sideloadOne(file, path, isDir = false) {
|
||||
await act(`Copy ${file.name} to ~/Downloads`, () => upload(file, "push"));
|
||||
return;
|
||||
}
|
||||
await loadTitles(); // fresh, so the dialog can say whether this replaces an installed title
|
||||
const choice = await confirmTitle(r.plan, canPush);
|
||||
if (choice === "push") {
|
||||
api("/api/titles", { action: "discard", token: r.token }).catch(() => {});
|
||||
@@ -1217,6 +1218,7 @@ function confirmTitle(plan, canPush) {
|
||||
$("titleName").value = plan.name;
|
||||
const idNote = () => { const id = titleId($("titleName").value);
|
||||
$("titleIdNote").textContent = !id ? "Needs some letters or digits"
|
||||
: !installedTitles ? `Shows in Steam as ${id}. Couldn't check whether it's installed already; if it is, this replaces it`
|
||||
: installedTitles.has(id.toLowerCase()) ? `Replaces the installed ${id}, and everything in its folder on the Frame`
|
||||
: `Shows in Steam as ${id}`; };
|
||||
$("titleName").oninput = idNote; idNote();
|
||||
@@ -1274,11 +1276,11 @@ async function installTitle(token, choice) {
|
||||
}
|
||||
} finally { $("prog").style.display = "none"; loadTitles(); }
|
||||
}
|
||||
let installedTitles = new Set(); // lower-case ids, so the install dialog can warn before replacing one
|
||||
let installedTitles = null; // lower-case ids, so the install dialog can warn before replacing one; null: unknown
|
||||
async function loadTitles() {
|
||||
let list;
|
||||
try { list = (await api("/api/titles")).titles; }
|
||||
catch (e) { $("titleList").innerHTML = `<div class="sub">${esc(e.message)}</div>`; return; }
|
||||
catch (e) { installedTitles = null; $("titleList").innerHTML = `<div class="sub">${esc(e.message)}</div>`; return; }
|
||||
installedTitles = new Set(list.map(t => String(t.id).toLowerCase()));
|
||||
$("titleCount").textContent = list.length ? `${list.length}` : "";
|
||||
$("titleList").innerHTML = list.length ? list.map(t => `
|
||||
|
||||
Reference in new issue
Block a user