mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-27 09:01:57 +01:00
600 lines
34 KiB
Python
600 lines
34 KiB
Python
"""Official Decky store transport and owned plugin transaction."""
|
|
import sandbox # noqa: F401
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import secrets
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import warnings
|
|
import zipfile
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'ui'))
|
|
import frame_decky_store
|
|
import frame_decky
|
|
|
|
|
|
def archive(flags=(), extra=None):
|
|
stream = io.BytesIO()
|
|
with zipfile.ZipFile(stream, 'w') as bundle:
|
|
bundle.writestr('Example/plugin.json', json.dumps({'name': 'Example', 'flags': list(flags)}))
|
|
bundle.writestr('Example/dist/index.js', 'export default {};' + ' ' * 1200)
|
|
if extra:
|
|
bundle.writestr(*extra)
|
|
data = stream.getvalue()
|
|
return data, hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def duplicate_archive(second=b'#!/bin/sh\nexit 0\n', second_mode=0o100755):
|
|
stream = io.BytesIO()
|
|
with warnings.catch_warnings(), zipfile.ZipFile(stream, 'w') as bundle:
|
|
warnings.simplefilter('ignore', UserWarning)
|
|
bundle.writestr('Example/plugin.json', json.dumps({'name': 'Example', 'flags': []}))
|
|
bundle.writestr('Example/dist/index.js', 'export default {};' + ' ' * 1200)
|
|
for mode, payload in ((0o100644, b'#!/bin/sh\nexit 0\n'), (second_mode, second)):
|
|
info = zipfile.ZipInfo('Example/bin/helper')
|
|
info.create_system = 3
|
|
info.external_attr = mode << 16
|
|
bundle.writestr(info, payload)
|
|
data = stream.getvalue()
|
|
return data, hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
class ArchiveTests(unittest.TestCase):
|
|
def test_catalog_selects_valid_visible_latest_versions(self):
|
|
records = [{'id': 42, 'name': 'Example', 'author': 'Panel Author', 'visible': True, 'versions':
|
|
[{'name': '1.0', 'hash': 'a' * 64}], 'tags': []},
|
|
{'id': 43, 'name': 'Hidden', 'visible': False, 'versions':
|
|
[{'name': '1.0', 'hash': 'b' * 64}]},
|
|
{'id': 44, 'name': 'Unsafe', 'visible': True, 'versions':
|
|
[{'name': '1.0; touch /tmp/bad', 'hash': 'c' * 64}]}]
|
|
with patch.object(frame_decky_store.urllib.request, 'urlopen', return_value=io.BytesIO(json.dumps(records).encode())):
|
|
self.assertEqual([(p['id'], p['author']) for p in frame_decky_store.catalog()], [(42, 'Panel Author')])
|
|
|
|
def test_catalog_marks_underscore_root_and_omits_ambiguous_store_ids(self):
|
|
records = [{'id': 42, 'name': 'Needs root', 'tags': ['_root'],
|
|
'versions': [{'name': '1.0', 'hash': 'a' * 64}]},
|
|
{'id': 43, 'name': 'First', 'versions': [{'name': '1.0', 'hash': 'b' * 64}]},
|
|
{'id': 43, 'name': 'Second', 'versions': [{'name': '2.0', 'hash': 'c' * 64}]}]
|
|
with patch.object(frame_decky_store.urllib.request, 'urlopen',
|
|
return_value=io.BytesIO(json.dumps(records).encode())):
|
|
self.assertEqual([(p['id'], p['root']) for p in frame_decky_store.catalog()], [(42, True)])
|
|
|
|
def test_accepts_hashed_non_root_archive(self):
|
|
data, digest = archive()
|
|
self.assertEqual(frame_decky_store.archive_info(data, digest, 'Example'), 'Example')
|
|
with self.assertRaisesRegex(ValueError, 'SHA-256'):
|
|
frame_decky_store.archive_info(data, '0' * 64, 'Example')
|
|
|
|
def test_rejects_root_x86_and_escape(self):
|
|
for data in (archive(flags=['root'])[0],
|
|
archive(extra=('Example/bin/helper', b'\x7fELF' + b'\x00' * 14 + (62).to_bytes(2, 'little')))[0],
|
|
archive(extra=('../outside', b'oops'))[0]):
|
|
with self.subTest(data=hashlib.sha256(data).hexdigest()[:8]):
|
|
with self.assertRaises(ValueError):
|
|
frame_decky_store.archive_info(data, hashlib.sha256(data).hexdigest(), 'Example')
|
|
|
|
def test_rejects_zip_path_collisions_and_non_object_manifest(self):
|
|
for name in ('Example/dist/index.js', 'Example/./dist/index.js',
|
|
'Example//dist/index.js'):
|
|
with self.subTest(name=name), warnings.catch_warnings():
|
|
warnings.simplefilter('ignore', UserWarning)
|
|
data, digest = archive(extra=(name, 'replacement'))
|
|
with self.assertRaisesRegex(ValueError, 'Unsafe|duplicate'):
|
|
frame_decky_store.archive_info(data, digest, 'Example')
|
|
stream = io.BytesIO()
|
|
with zipfile.ZipFile(stream, 'w') as bundle:
|
|
bundle.writestr('Example/plugin.json', '[]')
|
|
bundle.writestr('Example/dist/index.js', ' ' * 1200)
|
|
data = stream.getvalue()
|
|
with self.assertRaisesRegex(ValueError, 'manifest'):
|
|
frame_decky_store.archive_info(data, hashlib.sha256(data).hexdigest(), 'Example')
|
|
|
|
def test_only_identical_duplicate_payloads_with_execute_bit_difference_are_accepted(self):
|
|
data, digest = duplicate_archive()
|
|
self.assertEqual(frame_decky_store.archive_info(data, digest, 'Example'), 'Example')
|
|
for data, digest in (duplicate_archive(second=b'#!/bin/sh\nexit 1\n'),
|
|
duplicate_archive(second_mode=0o100666)):
|
|
with self.subTest(digest=digest), self.assertRaisesRegex(ValueError, 'Unsafe'):
|
|
frame_decky_store.archive_info(data, digest, 'Example')
|
|
|
|
|
|
@unittest.skipIf(os.name == 'nt', 'remote plugin transaction runs on Linux')
|
|
class StoreTransaction(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
home = Path(self.tmp.name)
|
|
unit = home / '.config/systemd/user/frame-control-decky.service'
|
|
unit.parent.mkdir(parents=True)
|
|
unit.write_text('[Service]\n')
|
|
fake = home / 'systemctl'
|
|
fake.write_text('#!/bin/sh\nif [ "$2" = is-active ]; then exit 0; fi\nexit 0\n')
|
|
fake.chmod(0o700)
|
|
self.env = {**os.environ, 'HOME': self.tmp.name,
|
|
'PATH': self.tmp.name + os.pathsep + os.environ['PATH']}
|
|
self.folder = home / '.local/share/frame-control/decky/plugins/Example'
|
|
self.folder.parent.mkdir(parents=True)
|
|
|
|
def apply(self, action, data=None, expected='-', plugin_id=42, folder='Example', version='1.0'):
|
|
token = secrets.token_hex(12)
|
|
source = Path('/tmp/frame-control-decky-plugin-' + token)
|
|
if data is not None:
|
|
source.write_bytes(data)
|
|
digest = hashlib.sha256(data or b'').hexdigest()
|
|
try:
|
|
return subprocess.run([sys.executable, '-', action, token, str(plugin_id), digest, folder, version, expected],
|
|
input=frame_decky_store.apply_script(plugin_id, digest), text=True, capture_output=True,
|
|
env=self.env, timeout=15)
|
|
finally:
|
|
source.unlink(missing_ok=True)
|
|
|
|
def test_install_update_remove_only_owned_folder(self):
|
|
first, _ = archive()
|
|
result = self.apply('install', first)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(json.loads((self.folder / '.frame-control-store.json').read_text())['storeId'], 42)
|
|
second, _ = archive(extra=('Example/data.txt', 'new'))
|
|
result = self.apply('update', second)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual((self.folder / 'data.txt').read_text(), 'new')
|
|
result = self.apply('remove')
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertFalse(self.folder.exists())
|
|
self.folder.mkdir()
|
|
(self.folder / 'plugin.json').write_text('{}')
|
|
result = self.apply('remove')
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertTrue(self.folder.exists())
|
|
external = Path(self.tmp.name) / 'other-plugin'
|
|
external.mkdir()
|
|
(external / '.frame-control-store.json').write_text('{"storeId":42}')
|
|
for item in self.folder.iterdir(): item.unlink()
|
|
self.folder.rmdir()
|
|
self.folder.symlink_to(external, target_is_directory=True)
|
|
result = self.apply('remove')
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertTrue(external.exists())
|
|
|
|
@unittest.skipUnless(__import__('platform').machine() in ('x86_64', 'aarch64'), 'native Linux wheel architecture')
|
|
def test_exact_deckyspy_gets_native_dependency_and_license_only_in_its_folder(self):
|
|
from framey_official import PLUGINS, archive as official_archive
|
|
import platform
|
|
plugin = next(p for p in PLUGINS if p['id'] == 89)
|
|
result = self.apply('install', official_archive(plugin), plugin_id=89,
|
|
folder=plugin['folder'], version=plugin['version'])
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
folder = self.folder.parent / plugin['folder']
|
|
receipt = json.loads((folder / '.frame-control-store.json').read_text())
|
|
self.assertEqual(receipt['compatShims'], ['deckyspy-psutil-and-battery'])
|
|
modules = folder / 'py_modules'
|
|
native = (modules / 'psutil/_psutil_linux.abi3.so').read_bytes()
|
|
self.assertEqual(native[:4], b'\x7fELF')
|
|
self.assertEqual(int.from_bytes(native[18:20], 'little'),
|
|
{'x86_64': 62, 'aarch64': 183}[platform.machine()])
|
|
self.assertIn('BSD 3-Clause', (modules / 'psutil-7.2.2.dist-info/LICENSE').read_text())
|
|
result = self.apply('remove', plugin_id=89, folder=plugin['folder'])
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertFalse(folder.exists())
|
|
|
|
@unittest.skipUnless(__import__('platform').machine() in ('x86_64', 'aarch64'), 'native Linux wheel architecture')
|
|
def test_deckyspy_missing_or_corrupt_payload_leaves_no_installed_copy(self):
|
|
from framey_official import PLUGINS, archive as official_archive
|
|
plugin = next(p for p in PLUGINS if p['id'] == 89)
|
|
for prefix, error in [('', 'Missing native psutil'),
|
|
("FRAMEY_BACKEND_WHEELS = {'x86_64':'AAAA','aarch64':'AAAA'}\n",
|
|
'SHA-256 mismatch')]:
|
|
with patch.object(frame_decky_store, 'apply_script', return_value=prefix+frame_decky_store.APPLY):
|
|
result = self.apply('install', official_archive(plugin), plugin_id=89,
|
|
folder=plugin['folder'], version=plugin['version'])
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn(error, result.stderr)
|
|
self.assertFalse((self.folder.parent / plugin['folder']).exists())
|
|
def test_exact_simple_timer_gets_lifecycle_repair_and_keeps_api1_and_license(self):
|
|
from framey_official import PLUGINS, archive as official_archive
|
|
plugin = next(p for p in PLUGINS if p['id'] == 96)
|
|
result = self.apply('install', official_archive(plugin), plugin_id=96,
|
|
folder=plugin['folder'], version=plugin['version'])
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
folder = self.folder.parent / plugin['folder']
|
|
self.assertEqual(json.loads((folder / '.frame-control-store.json').read_text())['compatShims'],
|
|
['simple-timer-lifecycle'])
|
|
self.assertEqual(json.loads((folder / 'plugin.json').read_text())['api_version'], 1)
|
|
with zipfile.ZipFile(io.BytesIO(official_archive(plugin))) as original:
|
|
self.assertEqual((folder / 'LICENSE').read_bytes(), original.read('simple-timer/LICENSE'))
|
|
# Backend behavior is verified separately by the real Decky lifecycle gate.
|
|
result = self.apply('remove', plugin_id=96, folder=plugin['folder'])
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertFalse(folder.exists())
|
|
|
|
def test_changed_simple_timer_digest_does_not_get_pinned_repair(self):
|
|
from framey_official import PLUGINS, archive as official_archive
|
|
plugin = next(p for p in PLUGINS if p['id'] == 96)
|
|
data = io.BytesIO(official_archive(plugin))
|
|
with zipfile.ZipFile(data, 'a') as changed:
|
|
original_main = changed.read('simple-timer/main.py')
|
|
changed.writestr('simple-timer/new-version.txt', 'unrecognized new payload')
|
|
result = self.apply('install', data.getvalue(), plugin_id=96,
|
|
folder=plugin['folder'], version=plugin['version'])
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
folder = self.folder.parent / plugin['folder']
|
|
self.assertEqual((folder / 'main.py').read_bytes(), original_main)
|
|
self.assertEqual(json.loads((folder / '.frame-control-store.json').read_text()).get('compatShims', []), [])
|
|
|
|
def test_exact_shotty_archive_gets_isolated_click_import(self):
|
|
from framey_official import PLUGINS, archive as official_archive
|
|
plugin = next(p for p in PLUGINS if p['id'] == 53)
|
|
data = official_archive(plugin)
|
|
self.assertEqual(hashlib.sha256(data).hexdigest(), plugin['sha256'])
|
|
shotty = self.folder.parent / plugin['folder']
|
|
args = {'plugin_id': plugin['id'], 'folder': plugin['folder'], 'version': plugin['version']}
|
|
result = self.apply('install', data, **args)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
receipt = json.loads((shotty / '.frame-control-store.json').read_text())
|
|
self.assertEqual(receipt['sha256'], plugin['sha256'])
|
|
self.assertEqual(receipt['compatShims'], ['shotty-click-app-dir'])
|
|
compat = shotty / 'py_modules/click.py'
|
|
self.assertTrue(compat.is_file())
|
|
source = '''import asyncio, importlib.util, logging, sys, types
|
|
from pathlib import Path
|
|
folder = Path(sys.argv[1])
|
|
sys.path.insert(0, str(folder/'py_modules'))
|
|
decky = types.ModuleType('decky_plugin')
|
|
decky.logger = logging.getLogger('shotty-test')
|
|
decky.DECKY_PLUGIN_DIR = str(folder)
|
|
sys.modules['decky_plugin'] = decky
|
|
spec = importlib.util.spec_from_file_location('shotty', folder/'main.py')
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
assert module.Plugin.__name__ == 'Plugin'
|
|
from click import get_app_dir
|
|
assert get_app_dir('Shotty') == str(Path.home()/'.config/Shotty')
|
|
photo = Path.home()/'.local/share/Steam/userdata/100/760/remote/42/screenshots/test.jpg'
|
|
photo.parent.mkdir(parents=True)
|
|
photo.write_bytes(b'local screenshot fixture')
|
|
plugin = module.Plugin()
|
|
assert asyncio.run(plugin.aggregate_all([[42, 'Local Test Game']])) == 1
|
|
copied = Path.home()/'Pictures/Screenshots/Local Test Game/test.jpg'
|
|
assert copied.read_bytes() == b'local screenshot fixture'
|
|
'''
|
|
imported = subprocess.run([sys.executable, '-c', source, str(shotty)],
|
|
capture_output=True, text=True,
|
|
env={**self.env, 'XDG_CONFIG_HOME': str(Path(self.tmp.name)/'.config')}, timeout=10)
|
|
self.assertEqual(imported.returncode, 0, imported.stderr)
|
|
result = self.apply('remove', expected=plugin['sha256'], **args)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertFalse(shotty.exists())
|
|
stream = io.BytesIO()
|
|
with zipfile.ZipFile(stream, 'w') as bundle:
|
|
bundle.writestr('Shotty/plugin.json', json.dumps({'name': 'Shotty', 'flags': []}))
|
|
bundle.writestr('Shotty/dist/index.js', 'export default {};' + ' ' * 1200)
|
|
other = stream.getvalue()
|
|
result = self.apply('install', other, **args)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertFalse((shotty / 'py_modules/click.py').exists())
|
|
self.assertNotIn('compatShims', json.loads((shotty / '.frame-control-store.json').read_text()))
|
|
|
|
def test_legacy_music_control_keeps_decky_class_calling_convention(self):
|
|
from framey_official import PLUGINS, archive as official_archive
|
|
plugin = next(p for p in PLUGINS if p['id'] == 111)
|
|
data = official_archive(plugin)
|
|
self.assertEqual(hashlib.sha256(data).hexdigest(), plugin['sha256'])
|
|
folder = self.folder.parent / plugin['folder']
|
|
args = {'plugin_id': plugin['id'], 'folder': plugin['folder'], 'version': plugin['version']}
|
|
result = self.apply('install', data, **args)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
receipt = json.loads((folder / '.frame-control-store.json').read_text())
|
|
self.assertEqual(receipt['sha256'], plugin['sha256'])
|
|
self.assertEqual(receipt['compatShims'], ['music-control-system-dbus-env'])
|
|
self.assertNotIn('api_version', json.loads((folder / 'plugin.json').read_text()))
|
|
self.assertIn('self._get_dbus_env(self)', (folder / 'main.py').read_text())
|
|
|
|
dbus = Path(self.tmp.name) / 'dbus-send'
|
|
dbus.write_text('''#!/usr/bin/env python3
|
|
import os, sys
|
|
from pathlib import Path
|
|
args = ' '.join(sys.argv[1:])
|
|
assert os.environ['LD_LIBRARY_PATH'] == os.environ.get('LD_LIBRARY_PATH_ORIG', '')
|
|
with Path(os.environ['MPRIS_CALLS']).open('a') as log: log.write(args + '\\n')
|
|
if 'ListNames' in args:
|
|
print('method return\\n array [\\n string "org.freedesktop.DBus"\\n string "org.mpris.MediaPlayer2.fixture"\\n ]')
|
|
elif 'string:PlaybackStatus' in args:
|
|
print('method return\\n variant string "Paused"')
|
|
else:
|
|
print('method return')
|
|
''')
|
|
dbus.chmod(0o700)
|
|
calls = Path(self.tmp.name) / 'mpris-calls'
|
|
source = '''import asyncio, importlib.util, os, sys, types
|
|
from pathlib import Path
|
|
decky = types.ModuleType('decky_plugin')
|
|
sys.modules['decky_plugin'] = decky
|
|
spec = importlib.util.spec_from_file_location('music_control', Path(sys.argv[1])/'main.py')
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
# Decky 3.3 uses the class for manifests with no api_version.
|
|
plugin = module.Plugin
|
|
async def check():
|
|
assert await plugin.sp_list_media_players(plugin) == 'org.mpris.MediaPlayer2.fixture'
|
|
await plugin.sp_set_media_player(plugin, 'org.mpris.MediaPlayer2.fixture')
|
|
assert await plugin.sp_get_media_player(plugin) == 'org.mpris.MediaPlayer2.fixture'
|
|
await plugin.sp_play(plugin)
|
|
await plugin.sp_pause(plugin)
|
|
await plugin.sp_next(plugin)
|
|
await plugin.sp_previous(plugin)
|
|
await plugin.sp_set_volume(plugin, 0.25)
|
|
assert await plugin.sp_track_status(plugin) == 'Paused'
|
|
asyncio.run(check())
|
|
assert os.environ['LD_LIBRARY_PATH'] == 'FrameyBundledLibraries'
|
|
'''
|
|
called = subprocess.run([sys.executable, '-c', source, str(folder)],
|
|
capture_output=True, text=True,
|
|
env={**self.env, 'MPRIS_CALLS': str(calls), 'LD_LIBRARY_PATH': 'FrameyBundledLibraries',
|
|
'LD_LIBRARY_PATH_ORIG': 'FrameySystemLibraries'}, timeout=10)
|
|
self.assertEqual(called.returncode, 0, called.stderr)
|
|
logged = calls.read_text()
|
|
self.assertIn('org.freedesktop.DBus.ListNames', logged)
|
|
self.assertIn('org.mpris.MediaPlayer2.Player.PlayPause', logged)
|
|
self.assertIn('org.mpris.MediaPlayer2.Player.Pause', logged)
|
|
self.assertIn('string:Volume variant:double:0.25', logged)
|
|
self.assertIn('org.mpris.MediaPlayer2.fixture', logged)
|
|
# An explicit same-version update replaces the previously shipped bad shim.
|
|
main = folder / 'main.py'
|
|
main.write_text(main.read_text().replace('self._get_dbus_env(self)', 'self._get_dbus_env()'))
|
|
receipt['compatShims'] = ['music-control-bound-methods']
|
|
(folder / '.frame-control-store.json').write_text(json.dumps(receipt))
|
|
result = self.apply('update', data, expected=plugin['sha256'], **args)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
called = subprocess.run([sys.executable, '-c', source, str(folder)],
|
|
capture_output=True, text=True,
|
|
env={**self.env, 'MPRIS_CALLS': str(calls),
|
|
'LD_LIBRARY_PATH': 'FrameyBundledLibraries'}, timeout=10)
|
|
self.assertEqual(called.returncode, 0, called.stderr)
|
|
result = self.apply('remove', expected=plugin['sha256'], **args)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertFalse(folder.exists())
|
|
|
|
older = next(p for p in PLUGINS if p['id'] == 12)
|
|
old_data = official_archive(older)
|
|
old_args = {'plugin_id': older['id'], 'folder': older['folder'], 'version': older['version']}
|
|
result = self.apply('install', old_data, **old_args)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(json.loads((folder / '.frame-control-store.json').read_text())['compatShims'],
|
|
['music-control-system-dbus-env'])
|
|
self.assertNotIn('api_version', json.loads((folder / 'plugin.json').read_text()))
|
|
self.assertIn('self._get_dbus_env(self)', (folder / 'main.py').read_text())
|
|
old_env = {**self.env, 'MPRIS_CALLS': str(calls), 'LD_LIBRARY_PATH': 'FrameyBundledLibraries'}
|
|
old_env.pop('LD_LIBRARY_PATH_ORIG', None)
|
|
called = subprocess.run([sys.executable, '-c', source, str(folder)],
|
|
capture_output=True, text=True, env=old_env, timeout=10)
|
|
self.assertEqual(called.returncode, 0, called.stderr)
|
|
result = self.apply('remove', expected=older['sha256'], **old_args)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
# A future archive at the same store ID must retain its own source.
|
|
other = io.BytesIO()
|
|
untouched = 'class Plugin: pass\n'
|
|
with zipfile.ZipFile(other, 'w') as bundle:
|
|
bundle.writestr('MusicControl/plugin.json', json.dumps({'name': 'Music Control', 'flags': []}))
|
|
bundle.writestr('MusicControl/dist/index.js', 'export default {};' + ' ' * 1200)
|
|
bundle.writestr('MusicControl/main.py', untouched)
|
|
result = self.apply('install', other.getvalue(), **args)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual((folder / 'main.py').read_text(), untouched)
|
|
self.assertNotIn('compatShims', json.loads((folder / '.frame-control-store.json').read_text()))
|
|
|
|
def test_installed_script_entrypoints_execute_without_zip_exec_bits(self):
|
|
data, _ = archive(extra=('Example/scripts/start.py',
|
|
b'#!/bin/sh\nprintf "Framey script: %s\\n" "$1"\n'))
|
|
result = self.apply('install', data)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
script = self.folder / 'scripts/start.py'
|
|
self.assertTrue(script.stat().st_mode & 0o100)
|
|
self.assertFalse(script.stat().st_mode & 0o6111 & ~0o100)
|
|
ran = subprocess.run([str(script), 'ready'], text=True, capture_output=True, timeout=5)
|
|
self.assertEqual(ran.returncode, 0, ran.stderr)
|
|
self.assertEqual(ran.stdout, 'Framey script: ready\n')
|
|
self.assertFalse((self.folder / 'plugin.json').stat().st_mode & 0o111)
|
|
self.assertFalse((self.folder / 'dist/index.js').stat().st_mode & 0o111)
|
|
|
|
def test_identical_duplicate_archive_installs_without_widening_permissions(self):
|
|
data, _ = duplicate_archive()
|
|
result = self.apply('install', data)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
helper = self.folder / 'bin/helper'
|
|
self.assertEqual(helper.read_bytes(), b'#!/bin/sh\nexit 0\n')
|
|
self.assertTrue(helper.stat().st_mode & 0o100)
|
|
self.assertFalse(helper.stat().st_mode & 0o6000)
|
|
self.assertEqual(self.apply('remove').returncode, 0)
|
|
|
|
def test_declared_executable_is_owner_only_without_privilege_bits(self):
|
|
info = zipfile.ZipInfo('Example/bin/entrypoint')
|
|
info.create_system = 3
|
|
info.external_attr = 0o104777 << 16
|
|
data, _ = archive(extra=(info, b'entrypoint payload'))
|
|
result = self.apply('install', data)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
executable = self.folder / 'bin/entrypoint'
|
|
self.assertEqual(executable.stat().st_mode & 0o6111, 0o100)
|
|
self.assertEqual(executable.read_bytes(), b'entrypoint payload')
|
|
|
|
def test_remove_while_service_is_stopped_does_not_restart_it(self):
|
|
data, _ = archive()
|
|
self.assertEqual(self.apply('install', data).returncode, 0)
|
|
fake = Path(self.tmp.name) / 'systemctl'
|
|
fake.write_text('#!/bin/sh\nif [ "$2" = is-active ]; then exit 1; fi\n'
|
|
'echo "$*" >> "$HOME/service-calls"\nexit 0\n')
|
|
result = self.apply('remove')
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertFalse(self.folder.exists())
|
|
self.assertEqual((Path(self.tmp.name) / 'service-calls').read_text().splitlines(),
|
|
['--user stop frame-control-decky.service'])
|
|
|
|
def test_shared_lock_rejects_plugin_and_loader_changes(self):
|
|
import fcntl
|
|
|
|
lock_path = Path(self.tmp.name) / '.local/share/frame-control/decky-transaction.lock'
|
|
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
|
data, _ = archive()
|
|
with lock_path.open('w+b') as held:
|
|
fcntl.flock(held, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
plugin = self.apply('install', data)
|
|
loader = subprocess.run([sys.executable, '-', 'remove', 'none', 'none', 'none'],
|
|
input=frame_decky.APPLY, text=True, capture_output=True,
|
|
env=self.env, timeout=15)
|
|
self.assertNotEqual(plugin.returncode, 0)
|
|
self.assertNotEqual(loader.returncode, 0)
|
|
self.assertIn('Another Decky change is in progress', plugin.stderr)
|
|
self.assertIn('Another Decky change is in progress', loader.stderr)
|
|
self.assertFalse(self.folder.exists())
|
|
|
|
installed = self.apply('install', data)
|
|
self.assertEqual(installed.returncode, 0, installed.stderr)
|
|
self.assertEqual(self.apply('remove').returncode, 0)
|
|
|
|
def test_remote_apply_rejects_colliding_zip_paths(self):
|
|
for name in ('Example/dist/index.js', 'Example/./dist/index.js'):
|
|
with self.subTest(name=name), warnings.catch_warnings():
|
|
warnings.simplefilter('ignore', UserWarning)
|
|
data, _ = archive(extra=(name, 'replacement'))
|
|
result = self.apply('install', data)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(self.folder.exists())
|
|
|
|
def test_remote_apply_rechecks_plugin_name_under_transaction_lock(self):
|
|
data, _ = archive()
|
|
other = self.folder.parent / 'Other'
|
|
other.mkdir()
|
|
(other / 'plugin.json').write_text('{"name":"Example"}')
|
|
result = self.apply('install', data)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn('different plugin with this name', result.stderr)
|
|
self.assertFalse(self.folder.exists())
|
|
self.assertEqual((other / 'plugin.json').read_text(), '{"name":"Example"}')
|
|
self.assertEqual(list(self.folder.parent.glob('.frame-control-stage-*')), [])
|
|
|
|
(other / 'plugin.json').write_text('{"name":"Distinct"}')
|
|
self.assertEqual(self.apply('install', data).returncode, 0)
|
|
(other / 'plugin.json').write_text('{"name":"Example"}')
|
|
result = self.apply('update', data)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn('different plugin with this name', result.stderr)
|
|
self.assertTrue(self.folder.exists())
|
|
self.assertEqual(list(self.folder.parent.glob('.frame-control-stage-*')), [])
|
|
|
|
def test_update_and_remove_reject_stale_installed_receipt(self):
|
|
first, first_hash = archive()
|
|
second, _ = archive(extra=('Example/data.txt', 'new'))
|
|
self.assertEqual(self.apply('install', first).returncode, 0)
|
|
marker = self.folder / '.frame-control-store.json'
|
|
self.assertEqual(json.loads(marker.read_text())['sha256'], first_hash)
|
|
|
|
# A newer store operation wins before the caller reaches the lock.
|
|
current = json.loads(marker.read_text())
|
|
current['sha256'] = 'b' * 64
|
|
marker.write_text(json.dumps(current))
|
|
for action, data in [('update', second), ('remove', None)]:
|
|
with self.subTest(action=action):
|
|
result = self.apply(action, data, expected=first_hash)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn('Installed plugin changed', result.stderr)
|
|
self.assertTrue(self.folder.exists())
|
|
self.assertEqual(json.loads(marker.read_text())['sha256'], 'b' * 64)
|
|
self.assertEqual(list(self.folder.parent.glob('.frame-control-stage-*')), [])
|
|
self.assertEqual(list(self.folder.parent.glob('.frame-control-old-*')), [])
|
|
self.assertEqual(self.apply('remove', expected='b' * 64).returncode, 0)
|
|
self.assertFalse(self.folder.exists())
|
|
|
|
|
|
class StoreHost(unittest.TestCase):
|
|
def test_reviewed_archive_changed_before_install_does_not_download(self):
|
|
state = {'managed': True, 'running': True, 'plugins': []}
|
|
catalog = [{'id': 42, 'name': 'Example', 'root': False,
|
|
'sha256': 'b' * 64, 'version': '2.0'}]
|
|
with patch.object(frame_decky_store.frame_decky, 'status', return_value=state), \
|
|
patch.object(frame_decky_store, 'catalog', return_value=catalog), \
|
|
patch.object(frame_decky_store.urllib.request, 'urlopen',
|
|
side_effect=AssertionError('No download expected')):
|
|
with self.assertRaisesRegex(ValueError, 'version changed'):
|
|
frame_decky_store.change(lambda *args, **kwargs: self.fail('No headset write expected'),
|
|
'install', 42, reviewed_hash='a' * 64)
|
|
|
|
def test_stopped_managed_service_allows_removing_owned_plugin(self):
|
|
state = {'managed': True, 'running': False, 'plugins':
|
|
[{'storeId': 42, 'folder': 'Example', 'name': 'Example', 'version': '1.0',
|
|
'sha256': 'a' * 64}]}
|
|
calls = []
|
|
with patch.object(frame_decky_store.frame_decky, 'status', return_value=state), \
|
|
patch.object(frame_decky_store, 'catalog') as network:
|
|
result = frame_decky_store.change(lambda command, **kwargs: calls.append(command) or '{}',
|
|
'remove', 42)
|
|
self.assertEqual(result, {})
|
|
self.assertEqual(len(calls), 1)
|
|
self.assertTrue(calls[0].endswith(' ' + 'a' * 64))
|
|
network.assert_not_called()
|
|
|
|
def test_renamed_store_plugin_cannot_be_installed_twice(self):
|
|
state = {'managed': True, 'running': True, 'plugins':
|
|
[{'storeId': 42, 'folder': 'OldFolder', 'name': 'Old Name', 'version': '1.0'}]}
|
|
catalog = [{'id': 42, 'name': 'New Name', 'root': False, 'sha256': 'a' * 64, 'version': '2.0'}]
|
|
with patch.object(frame_decky_store.frame_decky, 'status', return_value=state), \
|
|
patch.object(frame_decky_store, 'catalog', return_value=catalog) as network, \
|
|
patch.object(frame_decky_store.urllib.request, 'urlopen', side_effect=AssertionError('No download expected')):
|
|
with self.assertRaisesRegex(ValueError, 'already installed'):
|
|
frame_decky_store.change(lambda *args, **kwargs: self.fail('No headset write expected'),
|
|
'install', 42)
|
|
network.assert_not_called()
|
|
|
|
def test_update_rejects_name_taken_by_another_plugin_before_download(self):
|
|
state = {'managed': True, 'running': True, 'plugins': [
|
|
{'storeId': 42, 'folder': 'Example', 'name': 'Old Name', 'version': '1.0'},
|
|
{'storeId': 43, 'folder': 'Other', 'name': 'Example', 'version': '1.0'}]}
|
|
catalog = [{'id': 42, 'name': 'Example', 'root': False,
|
|
'sha256': 'a' * 64, 'version': '2.0'}]
|
|
with patch.object(frame_decky_store.frame_decky, 'status', return_value=state), \
|
|
patch.object(frame_decky_store, 'catalog', return_value=catalog), \
|
|
patch.object(frame_decky_store.urllib.request, 'urlopen',
|
|
side_effect=AssertionError('No download expected')):
|
|
with self.assertRaisesRegex(ValueError, 'different plugin with this name'):
|
|
frame_decky_store.change(lambda *args, **kwargs: self.fail('No headset write expected'),
|
|
'update', 42)
|
|
|
|
def test_update_allows_its_own_existing_name(self):
|
|
data, digest = archive()
|
|
state = {'managed': True, 'running': True, 'plugins': [
|
|
{'storeId': 42, 'folder': 'Example', 'name': 'Example', 'version': '1.0',
|
|
'sha256': 'a' * 64}]}
|
|
catalog = [{'id': 42, 'name': 'Example', 'root': False,
|
|
'sha256': digest, 'version': '2.0'}]
|
|
calls = []
|
|
with patch.object(frame_decky_store.frame_decky, 'status', return_value=state), \
|
|
patch.object(frame_decky_store, 'catalog', return_value=catalog), \
|
|
patch.object(frame_decky_store.urllib.request, 'urlopen', return_value=io.BytesIO(data)):
|
|
result = frame_decky_store.change(lambda command, **kwargs: calls.append(command) or '{}',
|
|
'update', 42)
|
|
self.assertEqual(result, {})
|
|
self.assertEqual(len(calls), 3) # Upload, apply, and temporary archive cleanup.
|
|
self.assertIn('update', calls[1])
|
|
self.assertTrue(calls[1].endswith(' ' + 'a' * 64))
|
|
|
|
def test_owned_plugin_can_be_removed_without_store_network(self):
|
|
state = {'managed': True, 'running': True, 'plugins':
|
|
[{'storeId': 42, 'folder': 'Example', 'name': 'Example', 'version': '1.0'}]}
|
|
def ssh(command, **kwargs):
|
|
return '{}'
|
|
progress = []
|
|
with patch.object(frame_decky_store.frame_decky, 'status', return_value=state), \
|
|
patch.object(frame_decky_store, 'catalog') as network:
|
|
frame_decky_store.change(ssh, 'remove', 42, lambda stage, percent: progress.append((stage, percent)))
|
|
network.assert_not_called()
|
|
self.assertEqual(progress, [('Checking installed plugins', 5), ('Removing plugin on Frame', 80)])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|