Desktop: run programs, scripts and AppImages; Switch to Desktop without Deck mode (#549)

* Run programs, scripts and AppImages from the desktop

Windows programs open with the Proton Steam has chosen, and a Steam game's own
launch program is handed to Steam. x86-64 Linux programs and game scripts run
under FEX, and an exec the kernel refuses goes to droiddeck-open through
libblsession's binfmt.c. Programs on noexec shared storage pass KIO's checks; a
folder of Linux programs there, and sharun AppImages, are added by the app once
and run from the copy. Executable scripts run without KIO's prompt, mpv plays
on the software desktop, FEXServer's socket moves under XDG_RUNTIME_DIR, and
Proton runtimes no longer show up as games.

* Switch to Desktop from Big Picture without Deck mode

Big Picture's Switch to Desktop asks the client's SteamOS manager, which the
client only has when started as SteamOS (Deck mode). Without it the request
failed and the screen stayed on "Switching to Desktop..." for good. The live
compat helper now sees the request on its way to the client and hands it to
steamos-session-select, which opens the KDE desktop as Deck mode already did.

* Run x86 programs from shared storage where they lie

FEX asks for the guest's PROT_EXEC when it maps a program or library, which
the kernel refuses on noexec shared storage; FEX never runs those pages
itself, so libblsession maps them without it in FEX's processes. An x86
game, its libraries and its start script now run in place instead of being
copied in; only native code on noexec storage is still copied.

A Steam library's games are never copied, and a Linux game's launch program
opened from the desktop goes to Steam like a Windows one.
This commit is contained in:
MaxsTechReview authored and GitHub committed 2026-10-10 19:12:20 -04:00
1 parent eadebc46d5
commit d9cd0d2cc8
25 files changed
+2135 -28

No files matched your search

@@ -256,6 +256,20 @@ object AppImageManager {
"Terminal=false\nCategories=$categories\nX-DroidDeck-AppImage=$id\n")
}
/** The desktop's import of [key]; older ones of the image at [path], since replaced, are removed. */
fun desktopImport(context: Context, key: String, path: String): String? {
var found: String? = null
root(context).listFiles()?.forEach { dir ->
val asked = FileUtils.readString(File(dir, UserApps.DESKTOP_IMPORT))?.lines() ?: return@forEach
if (asked.firstOrNull() == key) {
if (File(dir, "app").isDirectory) found = dir.name
} else if (asked.getOrNull(1) == path) {
remove(context, dir.name)
}
}
return found
}
fun remove(context: Context, id: String) {
if (id.isEmpty() || '/' in id || id.startsWith(".")) return
FileUtils.delete(File(root(context), id))
@@ -32,6 +32,8 @@ object UserApps {
private const val TAG = "UserApps"
const val GUEST_DIR = "/opt/droiddeck-apps"
const val SCRIPT_LAUNCHER = "/usr/local/bin/droiddeck-script-run"
/** An app the desktop asked for (droiddeck-open): its key, then the path it was started from there. */
const val DESKTOP_IMPORT = "desktop-import"
private const val FLATPAK_OVERRIDES = ".flatpak"
private const val ICON_SIZE = 256
private const val ELF_SCAN_LIMIT = 4000
@@ -72,8 +74,8 @@ object UserApps {
class FlatpakBundle(val file: File) : Source()
}
/** [icon]: a picked file's path or a suggested icon's URL. */
class Request(val source: Source, val name: String?, val icon: String?)
/** [icon]: a picked file's path or a suggested icon's URL. [meta]: files written into the added app's folder. */
class Request(val source: Source, val name: String?, val icon: String?, val meta: Map<String, String> = emptyMap())
/** How a script's folder reaches the session: linked in place, or copied in ([bytes] to copy). */
class ScriptPlan(val folder: File, val copy: Boolean, val bytes: Long)
@@ -121,8 +123,8 @@ object UserApps {
val name = request.name?.trim()?.takeIf { it.isNotEmpty() }
return withIcon(context, request.icon) { icon ->
when (val s = request.source) {
is Source.Script -> addScript(context, s.file, name, icon, onProgress)
is Source.AppImage -> AppImageManager.import(context, s.file, name, icon) { onProgress(it, -1) }
is Source.Script -> addScript(context, s.file, name, icon, request.meta, onProgress)
is Source.AppImage -> AppImageManager.import(context, s.file, name, icon, request.meta) { onProgress(it, -1) }
is Source.GitHub -> addFromGitHub(context, s.repo, name, icon, onProgress)
is Source.Flatpak -> addFlatpak(context, s.id, name, icon, onProgress)
is Source.FlatpakBundle -> addFlatpakBundle(context, s.file, name, icon, onProgress)
@@ -197,6 +199,14 @@ object UserApps {
}.distinct()
}
/** The desktop's import of [key] as the session sees it: an added script's folder, or an AppImage's. */
fun desktopImport(context: Context, key: String, path: String): String? {
root(context).listFiles()?.firstOrNull { dir ->
File(dir, "entry").isFile && FileUtils.readString(File(dir, DESKTOP_IMPORT))?.lineSequence()?.firstOrNull() == key
}?.let { return "$GUEST_DIR/${it.name}" }
return AppImageManager.desktopImport(context, key, path)?.let { "${AppImageManager.GUEST_DIR}/$it" }
}
fun planScript(script: File): ScriptPlan {
val folder = script.canonicalFile.parentFile ?: script.parentFile!!
var scanned = 0
@@ -209,7 +219,9 @@ object UserApps {
RandomAccessFile(f, "r").use { val b = ByteArray(4); it.read(b) == 4 && b[0] == 0x7f.toByte() && b[1] == 'E'.code.toByte() && b[2] == 'L'.code.toByte() && b[3] == 'F'.code.toByte() }
} catch (e: Exception) { false }
private fun addScript(context: Context, script: File, name: String?, icon: File?, onProgress: (String, Int) -> Unit): String? {
private fun addScript(
context: Context, script: File, name: String?, icon: File?, meta: Map<String, String>, onProgress: (String, Int) -> Unit,
): String? {
if (!script.isFile) return context.getString(R.string.user_apps_file_gone)
val plan = planScript(script)
val base = root(context).apply { mkdirs() }
@@ -232,6 +244,7 @@ object UserApps {
FileUtils.writeString(File(dir, "name"), title)
FileUtils.writeString(File(dir, "source"), script.canonicalPath)
if (icon != null && !saveIcon(icon, File(dir, "icon.png"))) Log.w(TAG, "icon ${icon.path} could not be read")
meta.forEach { (k, v) -> FileUtils.writeString(File(dir, k), v) }
writeScriptEntry(context, dir)
return null
} catch (e: Exception) {
@@ -94,6 +94,10 @@ object SessionFiles {
// The user's own AppImages, extracted at import (AppImageManager).
"usr/local/bin/droiddeck-appimage-run" to "usr/local/bin/droiddeck-appimage-run",
"usr/local/bin/droiddeck-script-run" to "usr/local/bin/droiddeck-script-run",
// The desktop's Run entries and its exec fallback: Windows programs through Steam's Proton,
// x86 Linux programs through FEX.
"usr/local/bin/droiddeck-open" to "usr/local/bin/droiddeck-open",
"usr/local/bin/droiddeck-proton-run" to "usr/local/bin/droiddeck-proton-run",
// The SteamOS helpers the client calls in Deck mode: the two Armada found it needs, plus
// the three under /usr/bin, all no-ops that answer "nothing to do" (see each file).
"usr/bin/steamos-update" to "usr/bin/steamos-update",
@@ -38,9 +38,11 @@ import com.droiddeck.launcher.core.HostProcess
import com.droiddeck.launcher.frontend.GameLaunchLink
import com.droiddeck.launcher.input.FakeInputWriter
import com.droiddeck.launcher.input.SteamControllerBle
import com.droiddeck.launcher.runtime.BwrapSpawner
import com.droiddeck.launcher.runtime.LinuxNetworkLinkComponent
import com.droiddeck.launcher.runtime.LinuxRuntime
import com.droiddeck.launcher.runtime.ProotFastPath
import com.droiddeck.launcher.runtime.UserApps
import com.droiddeck.launcher.wayland.WaylandCompositor
import java.io.File
import java.text.SimpleDateFormat
@@ -681,7 +683,9 @@ class SessionService : Service() {
// changes what games do and leaves the client alone. The microphone is its own opt-in on
// top, and the helper only opens an input stream when asked - so a user who wants game
// sound but no recording gets exactly that, and Android's recording indicator stays off.
val wantsDirectAudio = SessionState.mode == MODE_STEAM && SessionPrefs.directAudio(this)
// The desktop starts Windows programs through the same Proton wrappers (droiddeck-proton-run).
val wantsDirectAudio = (SessionState.mode == MODE_STEAM || SessionState.mode == MODE_DESKTOP) &&
SessionPrefs.directAudio(this)
val wantsMic = SessionPrefs.micEnabled(this) &&
checkSelfPermission(android.Manifest.permission.RECORD_AUDIO) == PackageManager.PERMISSION_GRANTED
// Both paths sit under the app's files directory, which the session binds at its own path,
@@ -1219,6 +1223,13 @@ class SessionService : Service() {
}
return
}
if (path != null && path.startsWith("desktop-import-") && path.endsWith(".request")) {
val request = File(dir, path)
val text = runCatching { request.readText() }.getOrNull() ?: return
request.delete()
importForDesktop(dir, text)
return
}
if (path != "steam-launch") return
val file = File(dir, path)
val text = try { file.readText() } catch (e: Exception) { return }
@@ -1248,6 +1259,48 @@ class SessionService : Service() {
launchWatcher = watcher
}
private val desktopImports = java.util.concurrent.ConcurrentHashMap<String, Any>()
/**
* What the desktop cannot run where it lies (droiddeck-open), added as Add an app adds it, menu entry
* included: an AppImage of sharun's, whose hard links Android denies apps and only the app's own proot
* (link2symlink) makes, or a script or program whose folder of Linux programs is on noexec shared
* storage, copied in. The answer goes beside the request, "ok <folder>" or "error <why>".
*/
private fun importForDesktop(dir: File, text: String) {
val fields = text.lineSequence().mapNotNull { line ->
line.split('=', limit = 2).takeIf { it.size == 2 }?.let { it[0] to it[1].trim() }
}.toMap()
val key = fields["id"]?.takeIf { it.matches(Regex("[0-9a-f]{16}")) } ?: return
val guestPath = fields["path"]?.takeIf { it.startsWith("/") && !it.contains("/../") } ?: return
val folder = fields["kind"] == "folder"
val binds = LinuxRuntime.lastBinds(this).map { BwrapSpawner.Bind.parse(it) }
val file = File(BwrapSpawner.hostPath(binds, LinuxRuntime.rootDir(this).path, guestPath))
val context = applicationContext
// A request for what is already being added waits for that, then finds it.
val lock = desktopImports.computeIfAbsent(key) { Any() }
Thread({
fun imported() = UserApps.desktopImport(context, key, guestPath)?.let { "ok $it" }
val request = UserApps.Request(
if (folder) UserApps.Source.Script(file) else UserApps.Source.AppImage(file),
if (folder) file.parentFile?.name else null, null, mapOf(UserApps.DESKTOP_IMPORT to "$key\n$guestPath"),
)
val answer = synchronized(lock) {
try {
imported() ?: UserApps.add(context, request) { _, _ -> }?.let { "error $it" }
?: imported() ?: "error " + context.getString(R.string.user_apps_failed)
} catch (e: Exception) {
Log.e(TAG, "Desktop import of $guestPath", e)
"error " + (e.message ?: context.getString(R.string.user_apps_failed))
}
}
Log.i(TAG, "Desktop import of $guestPath: $answer")
val result = File(dir, "desktop-import-$key.result")
val partial = File(dir, result.name + ".tmp")
runCatching { partial.writeText(answer.replace('\n', ' ') + "\n"); partial.renameTo(result) }
}, "desktop-import").start()
}
@Volatile private var syncWatcher: android.os.FileObserver? = null
private fun watchSyncWanted(root: File) {
+9
View File
@@ -38,6 +38,8 @@ export MOZ_SHM_NO_SEALS=1
# to the app as a Steam session - SteamOS's "Return to Gaming Mode". The session's own GL choice
# (zink on Turnip, for the Steam client) is put back by droiddeck-gpu.
export KWIN_COMPOSE=Q QT_QUICK_BACKEND=software LIBGL_ALWAYS_SOFTWARE=1 GALLIUM_DRIVER=llvmpipe
# mpv refuses that software GL and then fails its X11 fallback (an assertion): allowed it, it plays.
[ -e /etc/mpv/mpv.conf ] || { mkdir -p /etc/mpv && echo gpu-sw=yes >/etc/mpv/mpv.conf; } 2>/dev/null
export DROIDDECK_NESTED_SCALE=1
export DROIDDECK_GLTHREAD=${mesa_glthread:-}
unset MESA_LOADER_DRIVER_OVERRIDE LIBGL_KOPPER_DRI2 mesa_glthread
@@ -75,6 +77,13 @@ python3 /usr/local/bin/droiddeck-pad-defaults 2>&1 | sed 's/^/[pads] /'
python3 /usr/local/bin/droiddeck-desktop-gpu 2>&1 | sed 's/^/[gpu] /'
# The windowed action still opens melonDS at the size/placement the LXQt desktop used.
/usr/local/bin/droiddeck-desktop-window-rules 2>&1 | sed 's/^/[rules] /'
# Windows programs, Linux programs, AppImages and scripts open with droiddeck-open (Steam's Proton,
# FEX for x86); an exec the kernel refuses is handed to it as well (libblsession's binfmt.c). An
# AppImage run directly cannot mount itself under proot, so it unpacks itself to run.
python3 /usr/local/bin/droiddeck-open --install 2>&1 | sed 's/^/[open] /'
export APPIMAGE_EXTRACT_AND_RUN=1
# A click runs an executable script or program without KIO asking first; Dolphin's settings can still ask.
[ -e /etc/xdg/kiorc ] || printf '[Executable scripts]\nbehaviourOnLaunch=execute\n' >/etc/xdg/kiorc 2>/dev/null
# SteamOS's desktop icons: Return to Gaming Mode and Steam, put there once - what the user moves
# or deletes stays that way.
desktop_dir=$(xdg-user-dir DESKTOP 2>/dev/null); [ -n "$desktop_dir" ] && [ "$desktop_dir" != "$HOME" ] || desktop_dir=$HOME/Desktop
@@ -12,12 +12,14 @@ import sys
HOME = os.environ.get("HOME", "/root")
STEAM_ROOT = sys.argv[1] if len(sys.argv) > 1 else os.path.join(HOME, ".local/share/Steam")
LIBRARIES = ((STEAM_ROOT, "Steam Games"), ("/mnt/droiddeck-sd", "SD Games"))
# The client's own tools and runtimes live in steamapps beside the games; they are not titles.
NOT_GAMES = {228980, 1493710, 3127680, 4183110, 4427310, 4185400}
# The client's own tools and runtimes live in steamapps beside the games; they are not titles. A
# depot with a toolmanifest.vdf is one too (a runtime or a Proton), listed here or not.
NOT_GAMES = {228980, 1070560, 1391110, 1493710, 1628350, 3127680, 4183110, 4185400, 4427310, 4628740}
APPS_DIR = os.path.join(HOME, ".local/share/applications")
PREFIX = "droiddeck-game-"
NAME = re.compile(r'^\s*"name"\s*"([^"]*)"', re.M)
STATE = re.compile(r'^\s*"StateFlags"\s*"(\d+)"', re.M)
INSTALLDIR = re.compile(r'^\s*"installdir"\s*"([^"]*)"', re.M)
def games(library):
@@ -42,6 +44,9 @@ def games(library):
# StateFlags 4 = fully installed; anything else is downloading, updating or broken.
if not name or not state or (int(state.group(1)) & 4) == 0:
continue
installdir = INSTALLDIR.search(text)
if installdir and os.path.isfile(os.path.join(steamapps, "common", installdir.group(1), "toolmanifest.vdf")):
continue
out.append((appid, name.group(1)))
return out
@@ -406,12 +406,22 @@ def split_paths(value):
return [part for part in re.split(r"[:\s]+", value or "") if part]
def data_dir(env):
# FEXServer's socket lives here. Under ~/.local its host path passes sun_path's 108 bytes, and
# proot then binds it to a temporary name only the server sees: no client ever reaches it.
runtime = env.get("XDG_RUNTIME_DIR")
if runtime and os.path.isdir(runtime) and os.access(runtime, os.W_OK):
return os.path.join(runtime, "fex")
return DATA_DIR
def guest_environment(fex, env):
result = dict(env)
for directory in (CONFIG_DIR, DATA_DIR, CACHE_DIR):
data = data_dir(env)
for directory in (CONFIG_DIR, data, CACHE_DIR):
os.makedirs(directory, exist_ok=True)
result["FEX_APP_CONFIG_LOCATION"] = CONFIG_DIR + "/"
result["FEX_APP_DATA_LOCATION"] = DATA_DIR + "/"
result["FEX_APP_DATA_LOCATION"] = data + "/"
result["FEX_APP_CACHE_LOCATION"] = CACHE_DIR + "/"
result.setdefault("FEX_SILENTLOG", "0")
result.setdefault("FEX_OUTPUTLOG", "stderr")
@@ -0,0 +1,434 @@
#!/usr/bin/python3
"""Runs a program from the desktop: a Windows program through the Proton Steam has chosen, a Linux
program (x86 ones under FEX), an AppImage, or a script (under FEX beside x86 programs, as a game's
start script is). Shared storage is noexec: FEX runs x86 programs there where they lie (libblsession's
noexec.c), while a native one is copied - alone into the cache, or with its folder by the app once (as
Add an app does) when that holds native programs or libraries. A Steam library's games are Steam's.
droiddeck-open [--window | --terminal] FILE [ARGS...] the file manager's Run entries
droiddeck-open --exec FILE [ARGS...] an exec the kernel refused (libblsession)
droiddeck-open --install the Run entries and default handlers
The desktop composites in software, so a program started from it goes full screen on the GPU
(droiddeck-gpu), as the menu's games do: --window keeps it in a desktop window instead, --terminal
runs it in Konsole. A plain script stays on the desktop, and an exec whose output is read (a terminal,
a pipe, a file) stays where it was started.
"""
import fcntl
import hashlib
import os
import runpy
import shutil
import stat
import subprocess
import sys
import time
HOME = os.environ.get("HOME") or "/root"
BIN = "/usr/local/bin"
FEX = os.path.join(BIN, "droiddeck-fex")
GPU = os.path.join(BIN, "droiddeck-gpu")
PROTON_RUN = os.path.join(BIN, "droiddeck-proton-run")
APPIMAGE_RUN = os.path.join(BIN, "droiddeck-appimage-run")
SCRIPT_RUN = os.path.join(BIN, "droiddeck-script-run")
STEAM_LAUNCH = os.path.join(BIN, "droiddeck-steam-launch")
CACHE = os.path.join(HOME, ".cache", "droiddeck")
# Where the app adds each kind, and what an added one holds.
IMPORTS = {"appimage": ("/opt/appimages/user", "app"), "folder": ("/opt/droiddeck-apps", "entry")}
# Android's own folders on shared storage: never copied in whole.
SHARED_FOLDERS = ("alarms", "android", "audiobooks", "dcim", "documents", "download", "movies", "music",
"notifications", "pictures", "podcasts", "recordings", "ringtones")
SCAN_LIMIT = 4000
APPS_DIR = "/usr/local/share/applications"
WINDOWS = (".exe", ".bat", ".cmd", ".com", ".msi")
NOT_PROGRAMS = (".dll", ".sys", ".drv", ".cpl", ".ocx", ".efi", ".scr", ".mui")
SCRIPTS = (".sh", ".bash")
MACHINES = {3: "i386", 62: "x86_64", 183: "arm64"}
EMULATED = (3, 62)
QUICK_FAILURE = 20
IMPORT_WAIT = 1800
WINDOWS_TYPES = ("application/x-ms-dos-executable", "application/x-msdownload",
"application/vnd.microsoft.portable-executable", "application/x-msi", "application/x-bat")
SCRIPT_TYPES = ("application/x-shellscript", "text/x-shellscript")
LINUX_TYPES = ("application/x-executable", "application/x-pie-executable", "application/vnd.appimage")
ENTRIES = (
("droiddeck-open.desktop", "Run", "", WINDOWS_TYPES + SCRIPT_TYPES + LINUX_TYPES),
("droiddeck-open-window.desktop", "Run in a Desktop Window", "--window ", WINDOWS_TYPES + SCRIPT_TYPES + LINUX_TYPES),
("droiddeck-open-terminal.desktop", "Run in Konsole", "--terminal ", SCRIPT_TYPES + LINUX_TYPES),
)
def say(message):
print("droiddeck-open: %s" % message, file=sys.stderr, flush=True)
def notify(title, body, icon="dialog-error"):
if not os.environ.get("DBUS_SESSION_BUS_ADDRESS") or not shutil.which("notify-send"):
return
try:
subprocess.run(["notify-send", "-a", "DroidDeck", "-i", icon, title, body],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5)
except (OSError, subprocess.SubprocessError):
pass
def head(path, size=64):
try:
with open(path, "rb") as f:
return f.read(size)
except OSError:
return b""
def machine(data):
if len(data) < 20 or data[:4] != b"\x7fELF":
return None
return int.from_bytes(data[18:20], "big" if data[5] == 2 else "little")
def kind(path):
data = head(path)
ext = os.path.splitext(path)[1].lower()
if data[:4] == b"\x7fELF":
return "appimage" if data[8:11] == b"AI\x02" or ext == ".appimage" else "elf"
if ext in NOT_PROGRAMS:
return None
if ext in WINDOWS or data[:2] == b"MZ":
return "windows"
if data[:2] == b"#!" or ext in SCRIPTS:
return "script"
return None
def noexec(path):
try:
return bool(os.statvfs(path).f_flag & os.ST_NOEXEC)
except OSError:
return False
def interpreter(path):
line = head(path, 256).split(b"\n", 1)[0].rstrip(b"\r")
if line.startswith(b"#!"):
words = line[2:].decode("utf-8", "replace").split()
if words:
return words
return ["/bin/bash"]
def for_x86(script):
return subprocess.call([FEX, "needs", os.path.dirname(script)], stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL) == 0
def runnable(path):
"""The program itself, or a copy in the cache of a native one on noexec storage."""
if not noexec(path) or machine(head(path)) in EMULATED:
return path
st = os.stat(path)
key = hashlib.sha1(os.path.realpath(path).encode("utf-8")).hexdigest()[:16]
copy = os.path.join(CACHE, "exec", key, os.path.basename(path))
try:
same = os.stat(copy)
if same.st_size == st.st_size and same.st_mtime_ns == st.st_mtime_ns:
return copy
except OSError:
pass
os.makedirs(os.path.dirname(copy), exist_ok=True)
shutil.copyfile(path, copy + ".tmp")
os.utime(copy + ".tmp", ns=(st.st_atime_ns, st.st_mtime_ns))
os.chmod(copy + ".tmp", 0o755)
os.replace(copy + ".tmp", copy)
return copy
def appimage(path):
"""The image unpacked once into the cache, laid out as droiddeck-appimage-run expects."""
st = os.stat(path)
real = os.path.realpath(path)
key = hashlib.sha1(("%s\0%d\0%d" % (real, st.st_size, st.st_mtime_ns)).encode("utf-8")).hexdigest()[:16]
images = os.path.join(CACHE, "appimages")
return once(key, lambda: unpacked(path, real, key, images, os.path.join(images, key)))
def once(key, make):
"""[make]() for [key] one start at a time: a second start waits for the first, then finds it done."""
locks = os.path.join(CACHE, "locks")
os.makedirs(locks, exist_ok=True)
with open(os.path.join(locks, key), "w") as lock:
fcntl.flock(lock, fcntl.LOCK_EX)
return make()
def unpacked(path, real, key, images, folder):
if os.path.exists(os.path.join(folder, "app", "AppRun")):
return folder
earlier = import_of(key, "appimage")
if earlier:
return earlier
name = os.path.splitext(os.path.basename(path))[0]
notify("Preparing %s" % name, "Unpacking it for its first start.", "system-run")
work = folder + ".tmp"
shutil.rmtree(work, ignore_errors=True)
os.makedirs(work)
status = subprocess.call([FEX, "extract", real], cwd=work, stdout=subprocess.DEVNULL)
unpacked = os.path.realpath(os.path.join(work, "squashfs-root"))
if status != 0 or not os.path.isdir(unpacked):
shutil.rmtree(work, ignore_errors=True)
return imported("appimage", real, key, name)
os.replace(unpacked, os.path.join(work, "app"))
if os.path.islink(os.path.join(work, "squashfs-root")):
os.unlink(os.path.join(work, "squashfs-root"))
for file, value in (("name", name), ("arch", MACHINES.get(machine(head(path)), "")), ("source", real)):
with open(os.path.join(work, file), "w", encoding="utf-8") as f:
f.write(value + "\n")
for other in os.listdir(images):
old = os.path.join(images, other)
if other != key + ".tmp" and (head(os.path.join(old, "source"), 4096).decode("utf-8", "replace").strip() == real):
shutil.rmtree(old, ignore_errors=True)
os.replace(work, folder)
return folder
def import_of(key, what):
root, made = IMPORTS[what]
try:
entries = sorted(os.listdir(root))
except OSError:
return None
for entry in entries:
folder = os.path.join(root, entry)
if head(os.path.join(folder, "desktop-import"), 64).split(b"\n", 1)[0] == key.encode() \
and os.path.exists(os.path.join(folder, made)):
return folder
return None
def imported(what, real, key, name):
"""[real] added by the app instead, as Add an app adds it (to the menu too): an AppImage whose hard
links (sharun's) only the app's own proot can make, or a folder of native programs off noexec storage."""
launch = os.environ.get("BL_LAUNCH_DIR")
if not launch or not os.path.isdir(launch):
raise RuntimeError("%s cannot run from here; add it in DroidDeck with Add an app." % name)
result = os.path.join(launch, "desktop-import-%s.result" % key)
request = os.path.join(launch, "desktop-import-%s.request" % key)
with open(request + ".tmp", "w", encoding="utf-8") as f:
f.write("id=%s\nkind=%s\npath=%s\n" % (key, what, real))
os.replace(request + ".tmp", request)
if what == "appimage":
notify("Adding %s to DroidDeck" % name, "It is unpacked once, then it starts, and it stays in the menu.", "system-run")
else:
notify("Adding %s to DroidDeck" % name, "Shared storage cannot run its programs, so the folder is copied in "
"once; then it starts, and it stays in the menu.", "system-run")
deadline = time.monotonic() + IMPORT_WAIT
while time.monotonic() < deadline:
try:
with open(result, encoding="utf-8") as f:
answer = f.read().strip()
except OSError:
time.sleep(1)
continue
os.unlink(result)
verdict, _, detail = answer.partition(" ")
if verdict == "ok" and os.path.exists(os.path.join(detail, IMPORTS[what][1])):
return detail
raise RuntimeError(detail or "DroidDeck could not add %s." % name)
raise RuntimeError("DroidDeck did not finish adding %s." % name)
def game_folder(path):
"""The folder of [path] when it runs only from a copy: on noexec storage, holding native programs or
libraries (besides [path]) that it starts or loads. Not the storage itself, one of Android's folders
or a Steam library."""
real = os.path.realpath(path)
folder = os.path.dirname(real)
if in_steam_library(real) or not noexec(folder):
return None
top = storage_root(folder)
parts = os.path.relpath(folder, top).split(os.sep)
if folder == top or (len(parts) == 1 and parts[0].lower() in SHARED_FOLDERS):
return None
seen = 0
for root, dirs, files in os.walk(folder):
if root[len(folder):].count(os.sep) >= 3:
dirs[:] = []
for name in files:
child = os.path.join(root, name)
if child == real:
continue
seen += 1
if seen > SCAN_LIMIT:
return None
if machine(head(child)) not in (None,) + EMULATED:
return folder
return None
def in_steam_library(path):
"""Whether [path] is in a Steam library: its games are Steam's to start, never copied in."""
return "steamapps" in os.path.realpath(path).lower().split(os.sep)
def storage_root(folder):
"""The top of the storage [folder] is on: the last folder above it on the same file system."""
top = folder
try:
dev = os.stat(folder).st_dev
while os.path.dirname(top) != top and os.stat(os.path.dirname(top)).st_dev == dev:
top = os.path.dirname(top)
except OSError:
pass
return top
def copied_in(path):
"""The app's copy of [path]'s folder, added (once) for it."""
real = os.path.realpath(path)
key = hashlib.sha1(("folder\0%s" % real).encode("utf-8")).hexdigest()[:16]
name = os.path.basename(os.path.dirname(real))
return once(key, lambda: import_of(key, "folder") or imported("folder", real, key, name))
def steam_title(path):
"""The Steam game [path] is the program of: Steam starts it, with the client most games need."""
try:
run = runpy.run_path(PROTON_RUN)
root = os.environ.get("STEAM_COMPAT_CLIENT_INSTALL_PATH") or run["STEAM_ROOT"]
target = run["locate"](path, root)
return target.appid if run["steam_launches"](target, path, root) else None
except (OSError, ValueError, KeyError, IndexError):
return None
def command(path, args, what, window=False, appid=None):
"""(command, working directory) that runs [path] in place."""
folder = os.path.dirname(path)
if what == "steam":
return [STEAM_LAUNCH, "desktop", appid], folder
if what == "windows":
return [PROTON_RUN] + (["--window"] if window else []) + [path] + args, folder
if what == "x86 script":
return [FEX, "run", "--mode", "on", "--for", folder, "--"] + interpreter(path) + [path] + args, folder
if what == "script":
return interpreter(path) + [path] + args, folder
if what == "appimage":
return [APPIMAGE_RUN, appimage(path)] + args, folder
if what == "folder":
return [SCRIPT_RUN, copied_in(path)] + args, folder
return [FEX, "run", "--mode", "auto", "--", runnable(path)] + args, folder
def on_desktop(env):
return env.get("BL_DESKTOP") == "1" and not env.get("GAMESCOPE_WAYLAND_DISPLAY") and os.access(GPU, os.X_OK)
def unattended(env):
"""Nobody reads the output: it goes to /dev/null or the desktop's log, as from Dolphin or the menu."""
try:
out = os.fstat(1)
except OSError:
return True
if stat.S_ISCHR(out.st_mode):
return not os.isatty(1)
folder = env.get("BL_DEBUG_DIR")
try:
log = os.stat(os.path.join(folder, "desktop.log")) if folder else None
except OSError:
log = None
return log is not None and (log.st_dev, log.st_ino) == (out.st_dev, out.st_ino)
def placed(cmd, what, mode, env):
"""[cmd] where it shows: full screen on the GPU from the desktop, else where it was started."""
if mode == "terminal" and what not in ("windows", "steam") and shutil.which("konsole"):
return ["konsole", "--hold", "-e"] + cmd
if mode == "window" or what in ("script", "steam") or not on_desktop(env):
return cmd
if mode == "exec" and what != "windows" and not unattended(env):
return cmd
return [GPU] + cmd
def install(directory=APPS_DIR):
os.makedirs(directory, exist_ok=True)
defaults = []
for file, name, option, types in ENTRIES:
body = ("[Desktop Entry]\nType=Application\nName=%s\nIcon=system-run\nExec=%s %s%%f\nTerminal=false\n"
"NoDisplay=true\nMimeType=%s;\n" % (name, os.path.join(BIN, "droiddeck-open"), option, ";".join(types)))
write(os.path.join(directory, file), body)
for mime in WINDOWS_TYPES + SCRIPT_TYPES + LINUX_TYPES:
defaults.append("%s=%s" % (mime, ENTRIES[0][0]))
added = ["%s=%s" % (mime, ";".join(f for f, _, _, types in ENTRIES if mime in types) + ";")
for mime in WINDOWS_TYPES + SCRIPT_TYPES + LINUX_TYPES]
write(os.path.join(directory, "mimeapps.list"),
"[Default Applications]\n%s\n\n[Added Associations]\n%s\n" % ("\n".join(defaults), "\n".join(added)))
return 0
def write(path, text):
try:
with open(path, encoding="utf-8") as f:
if f.read() == text:
return
except OSError:
pass
with open(path + ".tmp", "w", encoding="utf-8") as f:
f.write(text)
os.replace(path + ".tmp", path)
def main(argv):
mode = "run"
if argv and argv[0] == "--install":
return install()
if argv and argv[0] in ("--window", "--terminal", "--exec"):
mode = argv.pop(0)[2:]
if argv and argv[0] == "--":
argv.pop(0)
if not argv:
say("usage: droiddeck-open [--window | --terminal | --exec] FILE [ARGS...] | --install")
return 2
path = os.path.abspath(argv[0])
what = kind(path)
if what is None:
say("%s is not a program this desktop can run" % path)
if mode != "exec":
notify("Cannot run %s" % os.path.basename(path), "It is not a Windows or Linux program or a script.")
return 126
linux_game = what in ("script", "elf") and mode == "run" and in_steam_library(path)
appid = steam_title(path) if (what == "windows" and mode != "window") or linux_game else None
if appid:
what = "steam"
elif what == "script" and for_x86(path):
what = "x86 script"
elif what in ("script", "elf") and game_folder(path):
what = "folder"
try:
cmd, folder = command(path, argv[1:], what, mode == "window", appid)
except (OSError, RuntimeError) as error:
say(str(error))
notify("Cannot run %s" % os.path.basename(path), str(error))
return 126
cmd = placed(cmd, what, mode, os.environ)
if mode != "exec" or not sys.stderr.isatty():
say("%s (%s): %s" % (path, what, " ".join(cmd[:4])))
try:
os.chdir(folder)
except OSError:
pass
if mode == "exec":
os.execv(cmd[0], cmd)
started = time.monotonic()
try:
status = subprocess.call(cmd)
except OSError as error:
status = 127
say("%s: %s" % (cmd[0], error))
if status != 0 and time.monotonic() - started < QUICK_FAILURE:
notify("%s did not start" % os.path.basename(path),
"It exited with status %d; the session's desktop.log says why." % status)
return status
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))
@@ -0,0 +1,331 @@
#!/usr/bin/python3
"""Runs a Windows program with the Proton the Steam client has chosen for it.
droiddeck-proton-run [--window] PROGRAM [ARGS...]
The program belongs to an installed Steam game when it sits in one's folder, to a prefix when it
sits in one's drive_c, or to a non-Steam shortcut that names it; any other program gets a prefix of
its own, under the id the client gives a new shortcut. That title's CompatToolMapping entry in
config.vdf picks the tool, else the client's default for all other titles ("0"). The tool is
started as the client starts it - its toolmanifest's command line, the waitforexitandrun verb and
the STEAM_COMPAT_* environment - so its wrapper applies the game's settings, components and audio.
--window keeps the program in the desktop's own window, drawn with WineD3D.
"""
import json
import os
import re
import runpy
import struct
import sys
import zlib
HOME = os.environ.get("HOME") or "/root"
STEAM_ROOT = os.path.join(HOME, ".local", "share", "Steam")
BIN_DIR = os.path.dirname(os.path.abspath(__file__))
STATE = os.path.join(HOME, ".local", "share", "droiddeck-compat", "state.json")
EXTRA_LIBRARIES = ("/mnt/droiddeck", "/mnt/droiddeck-sd")
SYSTEM_TOOLS = ("/usr/share/steam/compatibilitytools.d", "/usr/local/share/steam/compatibilitytools.d")
VERB = "waitforexitandrun"
SHORTCUT_BIT = 0x80000000
STEAM_APP_LIMIT = 2 ** 31
_compat = None
def say(message):
print("droiddeck-proton-run: %s" % message, file=sys.stderr, flush=True)
def compat():
global _compat
if _compat is None:
_compat = runpy.run_path(os.path.join(BIN_DIR, "steam-compatibility"))
return _compat
def read_text(path):
try:
with open(path, encoding="utf-8", errors="replace") as f:
return f.read()
except OSError:
return None
def libraries(root):
found, seen = [], set()
for library in compat()["libraries"](root) + list(EXTRA_LIBRARIES):
try:
key = os.stat(os.path.join(library, "steamapps"))
except OSError:
continue
if (key.st_dev, key.st_ino) not in seen:
seen.add((key.st_dev, key.st_ino))
found.append(library)
return found
def mapping(root):
text = read_text(os.path.join(root, "config", "config.vdf"))
if not text:
return {}
c = compat()
tokens = c["tokenize"](text)
steam = c["find_block"](tokens, ["InstallConfigStore", "Software", "Valve", "Steam"])
if steam is None:
return {}
block = c["find_block"](tokens[steam + 1:], ["CompatToolMapping"])
return {} if block is None else c["read_mapping"](tokens, block + steam + 1)
def tools(root):
"""Windows compatibility tools this runtime can start, by the name config.vdf uses."""
found = {}
for base in [os.path.join(root, "compatibilitytools.d")] + list(SYSTEM_TOOLS):
try:
entries = sorted(os.listdir(base))
except OSError:
continue
for entry in entries:
folder = os.path.join(base, entry)
vdf = os.path.join(folder, "compatibilitytool.vdf")
if not re.search(r'"from_oslist"\s+"[^"]*windows', read_text(vdf) or "", re.I):
continue
command = compat()["manifest_entry_point"](os.path.join(folder, "toolmanifest.vdf"))
if not command or not os.access(os.path.join(folder, command), os.X_OK):
continue
name = compat()["first_key"](vdf, ["compatibilitytools", "compat_tools"]) or entry
found.setdefault(name, (folder, command))
return found
def state_default():
try:
with open(STATE, encoding="utf-8") as f:
value = json.load(f).get("default")
return value if isinstance(value, str) else None
except (OSError, ValueError, AttributeError):
return None
def choose_tool(appid, current, available):
"""(tool name, why) from the client's mapping, else the session's own default."""
for key, why in ((appid, "Steam's choice for this title"), ("0", "Steam's default")):
name = current.get(key)
if name in available:
return name, why
c = compat()
for name in (state_default(), c["TOOL"], c["TOOL_11"]):
if name in available:
return name, "the default ARM64 Proton"
return (sorted(available)[0], "the only one installed") if available else (None, None)
class Target:
def __init__(self, appid, compatdata, install, library, kind, start=None):
self.appid = str(appid)
self.compatdata = compatdata
self.install = install
self.library = library
self.kind = kind
self.start = start or install
def ancestors(path):
"""(folder, the entry below it on the way to path) from the program's folder up to /."""
child, folder = path, os.path.dirname(path)
while True:
yield folder, os.path.basename(child)
if folder == os.path.dirname(folder):
return
child, folder = folder, os.path.dirname(folder)
def identity(path):
try:
st = os.stat(path)
return st.st_dev, st.st_ino
except OSError:
return None
def installed_app(steamapps, installdir):
for name in sorted(os.listdir(steamapps)):
m = re.fullmatch(r"appmanifest_(\d+)\.acf", name)
if not m:
continue
found = re.search(r'"installdir"\s+"([^"]+)"', read_text(os.path.join(steamapps, name)) or "")
if found and found.group(1).lower() == installdir.lower():
return m.group(1)
return None
def steam_game(program, libs):
commons = {}
for library in libs:
key = identity(os.path.join(library, "steamapps", "common"))
if key:
commons.setdefault(key, library)
for folder, below in ancestors(program):
library = commons.get(identity(folder))
if library is None:
continue
appid = installed_app(os.path.join(library, "steamapps"), below)
if appid is None or appid in compat()["NOT_GAMES"]:
return None
return Target(appid, os.path.join(library, "steamapps", "compatdata", appid),
os.path.join(folder, below), library, "steam")
return None
def prefix_of(program):
for folder, below in ancestors(program):
parent = os.path.dirname(folder)
if below == "pfx" and os.path.basename(parent) == "compatdata" and os.path.basename(folder).isdigit():
library = os.path.dirname(os.path.dirname(parent))
return Target(os.path.basename(folder), folder, os.path.dirname(program), library, "prefix")
return None
def unquoted(value):
return value.strip().strip('"') if isinstance(value, str) else ""
def same_file(a, b):
try:
return bool(b) and os.path.samefile(a, b)
except OSError:
return False
def shortcut_entries(root):
"""(appid, exe, start dir) of the client's non-Steam shortcuts and the app's added games."""
entries = []
try:
tool = runpy.run_path(os.path.join(BIN_DIR, "droiddeck-steam-shortcuts"))
account = tool["account_dir"](root)
if account:
with open(os.path.join(account, "config", "shortcuts.vdf"), "rb") as f:
shortcuts = tool["parse"](f.read()).get("shortcuts", {})
for value in shortcuts.values():
if not isinstance(value, dict):
continue
appid = value.get("appid", value.get("AppId"))
exe = value.get("Exe", value.get("exe"))
if isinstance(appid, int) and exe:
entries.append((appid & 0xFFFFFFFF, unquoted(exe), unquoted(value.get("StartDir", ""))))
except (OSError, ValueError, IndexError, KeyError, struct.error):
pass
try:
with open(os.environ.get("BL_ADDED_GAMES", ""), encoding="utf-8") as f:
for game in json.load(f):
if isinstance(game, dict) and game.get("exe") and str(game.get("appid", "")).isdigit():
entries.append((int(game["appid"]) & 0xFFFFFFFF, game["exe"], game.get("dir") or ""))
except (OSError, ValueError, TypeError):
pass
return entries
def shortcut(program, root):
for appid, exe, start in shortcut_entries(root):
if same_file(program, exe):
return Target(appid, os.path.join(root, "steamapps", "compatdata", str(appid)),
os.path.dirname(program), root, "shortcut", start if os.path.isdir(start) else None)
return None
def own_id(program):
"""The id the client gives a new shortcut: the CRC of its quoted target and its name."""
name = os.path.splitext(os.path.basename(program))[0]
return zlib.crc32(('"%s"%s' % (program, name)).encode("utf-8")) | SHORTCUT_BIT
def locate(program, root):
libs = libraries(root)
found = steam_game(program, libs) or prefix_of(program) or shortcut(program, root)
if found:
return found
appid = own_id(program)
return Target(appid, os.path.join(root, "steamapps", "compatdata", str(appid)),
os.path.dirname(program), root, "own")
def steam_launches(target, program, root):
"""Whether a launch option of the Steam game starts [program]: then Steam should start it."""
if target.kind != "steam":
return False
entry = compat()["appinfo_entries"](os.path.join(root, "appcache", "appinfo.vdf")).get(target.appid, {})
return os.path.relpath(program, target.install).replace(os.sep, "/").lower() in entry.get("launch", ())
def environment(env, target, tool_folder, root, window=False):
env = dict(env)
steamapps = [os.path.join(lib, "steamapps") for lib in libraries(root)]
env.update({
"STEAM_COMPAT_DATA_PATH": target.compatdata,
"STEAM_COMPAT_CLIENT_INSTALL_PATH": root,
"STEAM_COMPAT_INSTALL_PATH": target.install,
"STEAM_COMPAT_LIBRARY_PATHS": ":".join(steamapps),
"STEAM_COMPAT_TOOL_PATHS": tool_folder,
"STEAM_COMPAT_SHADER_PATH": os.path.join(target.library, "steamapps", "shadercache", target.appid),
})
number = int(target.appid)
if number < STEAM_APP_LIMIT:
env.update(SteamAppId=target.appid, SteamGameId=target.appid, SteamOverlayGameId=target.appid,
STEAM_COMPAT_APP_ID=target.appid)
else:
game = str(number << 32 | 0x02000000)
env.update(SteamAppId="0", STEAM_COMPAT_APP_ID="0", SteamGameId=game, SteamOverlayGameId=game)
if window:
env.setdefault("PROTON_USE_WINED3D", "1")
return env
def windows_path(path):
return "Z:" + path.replace("/", "\\")
def arguments(program, args):
if program.lower().endswith(".msi"):
return ["msiexec", "/i", windows_path(program)] + args
return [program] + args
def main(argv):
window = False
while argv and argv[0].startswith("--"):
option = argv.pop(0)
if option == "--window":
window = True
elif option == "--":
break
else:
say("unknown option %s" % option)
return 2
if not argv:
say("usage: droiddeck-proton-run [--window] PROGRAM [ARGS...]")
return 2
program = os.path.abspath(argv[0])
if not os.path.isfile(program):
say("%s does not exist" % program)
return 2
root = os.environ.get("STEAM_COMPAT_CLIENT_INSTALL_PATH") or STEAM_ROOT
target = locate(program, root)
available = tools(root)
name, why = choose_tool(target.appid, mapping(root), available)
if name is None:
say("no Proton is installed yet; open Steam once and it is downloaded for you")
return 69
folder, command = available[name]
os.makedirs(target.compatdata, exist_ok=True)
env = environment(os.environ, target, folder, root, window)
say("%s with %s (%s), %s prefix %s" % (program, name, why, target.kind, target.compatdata))
try:
os.chdir(target.start)
except OSError:
os.chdir(os.path.dirname(program))
entry = os.path.join(folder, command)
os.execve(entry, [entry, VERB] + arguments(program, argv[1:]), env)
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))
@@ -12,14 +12,15 @@ first=
IFS= read -r first < "$entry"
first=${first%$'\r'}
interp=(/bin/bash)
case $first in '#!'*) read -r -a interp <<< "${first#'#!'}" ;; esac
# A program as the entry (the desktop's copy of a game's folder) runs itself.
case $first in '#!'*) read -r -a interp <<< "${first#'#!'}" ;; $'\x7f'ELF*) interp=() ;; esac
fex=$(cat "$dir/fex" 2>/dev/null)
case $fex in on|off) ;; *) fex=auto ;; esac
run=("${interp[@]}" "./$entry")
if [ -x /usr/local/bin/droiddeck-fex ]; then
run=(/usr/local/bin/droiddeck-fex run --mode "$fex" --for "$folder" -- "${interp[@]}" "./$entry")
fi
echo "== script $entry in $folder (${interp[*]}, FEX $fex)"
echo "== script $entry in $folder (${interp[*]:-a program}, FEX $fex)"
case ${DBUS_SESSION_BUS_ADDRESS:-} in
unix:*) "${run[@]}" "$@" ;;
*) unset DBUS_SESSION_BUS_ADDRESS
@@ -8,6 +8,7 @@ import select
import signal
import socket
import struct
import subprocess
import sys
import time
@@ -125,6 +126,37 @@ INSTALL_JS = r"""
return "installed";
})()
""".replace("__BINDING__", BINDING)
# Big Picture's "Switch to Desktop" asks the client's SteamOS manager, which the client has only when it
# runs as SteamOS (-steamos3, Deck mode); otherwise the request fails and the screen stays on "Switching
# to Desktop..." for good. The request is caught on its way to the client and answered as SteamOS's
# own path does, with steamos-session-select.
SWITCH_BINDING = "droiddeckSwitchToDesktop"
SWITCH_JS = r"""
(() => {
if (window.__droiddeckSwitch) return "present";
const name = new TextEncoder().encode("SteamOSManager.SwitchToDesktop#");
const send = WebSocket.prototype.send;
const asks = (data) => {
const bytes = data instanceof ArrayBuffer ? new Uint8Array(data)
: ArrayBuffer.isView(data) ? new Uint8Array(data.buffer, data.byteOffset, data.byteLength) : null;
if (!bytes) return false;
const end = Math.min(bytes.length, 256) - name.length;
for (let i = 8; i <= end; i++) {
let j = 0;
while (j < name.length && bytes[i + j] === name[j]) j++;
if (j === name.length) return true;
}
return false;
};
WebSocket.prototype.send = function (data) {
if (typeof window.__BINDING__ === "function" && asks(data)) window.__BINDING__("");
return send.call(this, data);
};
window.__droiddeckSwitch = true;
return "installed";
})()
""".replace("__BINDING__", SWITCH_BINDING)
SESSION_SELECT = "/usr/bin/steamos-session-select"
class Closed(Exception):
@@ -135,6 +167,18 @@ def say(message):
print("droiddeck-steam-compat: %s" % message, flush=True)
def called(event, name):
return event.get("method") == "Runtime.bindingCalled" and event.get("params", {}).get("name") == name
def switch_to_desktop():
say("Switch to Desktop: asking for the desktop")
try:
subprocess.call([SESSION_SELECT, "plasma"])
except OSError as e:
say("Switch to Desktop failed: %s" % e)
def connect(port, timeout=5):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
@@ -357,9 +401,10 @@ class Watcher:
class Helper:
def __init__(self, steam_root, port):
def __init__(self, steam_root, port, steamos=False):
self.steam_root = steam_root
self.port = port
self.steamos = steamos
self.page = None
self.known = set()
self.installed = None
@@ -374,6 +419,9 @@ class Helper:
if page.evaluate("typeof SteamClient === 'object' && typeof SteamClient.WebUITransport === 'object'") is not True:
raise LookupError("SteamClient is not ready")
page.call("Runtime.addBinding", {"name": BINDING})
if not self.steamos:
page.call("Runtime.addBinding", {"name": SWITCH_BINDING})
page.evaluate(SWITCH_JS)
page.compat("get(0)")
except Exception:
page.ws.close()
@@ -422,6 +470,13 @@ class Helper:
state = COMPAT["read_json"](COMPAT["state_path"]("state.json"))
return isinstance(request.get("seq"), int) and request["seq"] > int(state.get("applied", 0) or 0)
def drain(self):
"""The page's events; a switch to the desktop is acted on as it comes."""
events = self.page.drain()
if any(called(e, SWITCH_BINDING) for e in events):
switch_to_desktop()
return events
def serve(self, watcher):
with open(COMPAT["state_path"]("live.pid"), "w") as f:
f.write("%d\n" % os.getpid())
@@ -432,15 +487,14 @@ class Helper:
ready = [self.page.ws] if self.page.ws.pending() else select.select(sources, [], [], 30 if watcher.fileno() >= 0 else 2)[0]
due = False
if self.page.ws in ready:
due = any(e.get("method") == "Runtime.bindingCalled" and e.get("params", {}).get("name") == BINDING
for e in self.page.drain())
due = any(called(e, BINDING) for e in self.drain())
if (watcher in ready or watcher.fileno() < 0) and watcher.changed():
due = self.installed != COMPAT["installed_apps"](self.steam_root) or self.request_waiting()
if due:
time.sleep(0.3)
if watcher.fileno() >= 0 and select.select([watcher], [], [], 0)[0]:
watcher.changed()
self.page.drain()
self.drain()
self.reconcile()
@@ -466,7 +520,9 @@ def main():
return 0
os.makedirs(COMPAT["COMPAT_DIR"], exist_ok=True)
watcher = Watcher([COMPAT["state_path"]("request.json")] + [os.path.join(lib, "steamapps", "libraryfolders.vdf") for lib in (steam_root, COMPAT["LIBRARY"] + "-sd")])
helper = Helper(steam_root, port)
# The client has its own SteamOS manager when started as SteamOS (droiddeck-session's Deck mode).
steamos = os.environ.get("BL_STEAMDECK") == "1" and os.environ.get("BL_STEAM_UI") != "desktop"
helper = Helper(steam_root, port, steamos)
failures = 0
last = None
while True:
@@ -754,6 +754,21 @@ APPINFO_CACHE = {}
STEAM_PLAY_MANIFESTS = 891390
def launch_programs(config):
"""The Windows and Linux programs an app's launch options start, relative to its folder (lowercase, "/")."""
launch = config.get("launch") if isinstance(config, dict) else None
found = []
for option in launch.values() if isinstance(launch, dict) else ():
if not isinstance(option, dict) or not option.get("executable"):
continue
settings = option.get("config") if isinstance(option.get("config"), dict) else {}
oslist = str(settings.get("oslist", "")).lower()
if oslist and "windows" not in oslist and "linux" not in oslist:
continue
found.append(os.path.normpath(str(option["executable"]).replace("\\", "/")).lstrip("/").lower())
return found
def appinfo_entries(path):
import struct
try:
@@ -844,6 +859,7 @@ def appinfo_entries(path):
"osarch": str(common.get("osarch", "")).lower(),
"type": str(common.get("type", "")).lower(),
"recommended": str(recommended),
"launch": launch_programs(info.get("config")),
}
if app == STEAM_PLAY_MANIFESTS:
extended = info.get("extended", {})
+206
View File
@@ -0,0 +1,206 @@
/*
* binfmt_misc for the desktop, which proot does not have: an exec of an x86 Linux program, a Windows
* program, or a program or script on noexec shared storage goes to droiddeck-open --exec, which runs
* it under FEX, Proton or its interpreter. x86 and Windows programs are caught before the exec: proot
* loads a static x86 program as a native one, which dies on its first instruction, and execvp runs a
* Windows program as a shell script. A program on noexec storage also passes access(X_OK), which KIO
* and Qt ask before they start one. Only in the desktop session.
*/
#define _GNU_SOURCE
#include <dlfcn.h>
#include <elf.h>
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <spawn.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/statvfs.h>
#include <unistd.h>
#ifndef BINFMT_HANDLER
#define BINFMT_HANDLER "/usr/local/bin/droiddeck-open"
#endif
typedef int (*spawn_fn)(pid_t *, const char *, const posix_spawn_file_actions_t *, const posix_spawnattr_t *,
char *const[], char *const[]);
/* BINFMT_TEST: tools/tests builds it for the host, whose own x86 programs stand in for foreign ones. */
#if defined(__aarch64__) || defined(BINFMT_TEST)
static int desktop(void) {
const char *value = getenv("BL_DESKTOP");
return value && value[0] == '1' && value[1] == 0;
}
/* 1: x86 program or noexec storage, 2: Windows program, 3: script; 0 for anything else. */
static int foreign(const char *path, int error) {
if (!path || !strcmp(path, BINFMT_HANDLER)) return 0;
int fd = open(path, O_RDONLY | O_CLOEXEC);
if (fd < 0) return 0;
struct stat st;
unsigned char head[20];
ssize_t n = fstat(fd, &st) == 0 && S_ISREG(st.st_mode) ? read(fd, head, sizeof head) : -1;
int kind = 0;
if (n >= 20 && !memcmp(head, ELFMAG, SELFMAG)) {
unsigned machine = head[EI_DATA] == ELFDATA2MSB ? (unsigned)head[18] << 8 | head[19] : head[18] | (unsigned)head[19] << 8;
kind = machine == EM_X86_64 || machine == EM_386 || error == EACCES ? 1 : 0;
} else if (n >= 2 && head[0] == 'M' && head[1] == 'Z') {
kind = 2;
} else if (n >= 2 && head[0] == '#' && head[1] == '!' && error == EACCES) {
kind = 3;
}
/* What the kernel refuses for its permissions stays refused; noexec storage has no execute bits to give. */
struct statvfs vfs;
int allowed = kind && (((st.st_mode & 0111) && error != EACCES) || (fstatvfs(fd, &vfs) == 0 && (vfs.f_flag & ST_NOEXEC)));
close(fd);
return allowed ? kind : 0;
}
static int wanted(const char *path, int error) {
return (error == ENOEXEC || error == ENOENT || error == EACCES) && desktop() && foreign(path, error);
}
/* The program on PATH as execvp would find it, into [out]. */
static int lookup(const char *file, char *out, size_t size) {
if (!file || !*file) return 0;
if (strchr(file, '/')) return snprintf(out, size, "%s", file) < (int)size;
const char *paths = getenv("PATH");
if (!paths) paths = "/bin:/usr/bin";
for (const char *p = paths;; ) {
const char *end = strchrnul(p, ':');
int n = end > p ? snprintf(out, size, "%.*s/%s", (int)(end - p), p, file) : snprintf(out, size, "%s", file);
struct stat st;
if (n > 0 && n < (int)size && stat(out, &st) == 0 && S_ISREG(st.st_mode)) return 1;
if (!*end) return 0;
p = end + 1;
}
}
static int count(char *const argv[]) {
int n = 0;
while (argv && argv[n]) n++;
return n;
}
static void handler_args(char **args, const char *path, char *const argv[], int n) {
args[0] = "droiddeck-open";
args[1] = "--exec";
args[2] = (char *)path;
int i = 3;
for (int k = 1; k < n; k++) args[i++] = argv[k];
args[i] = NULL;
}
static void hand_off(const char *path, char *const argv[], char *const envp[]) {
static int (*real)(const char *, char *const[], char *const[]);
if (!real) real = (int (*)(const char *, char *const[], char *const[]))dlsym(RTLD_NEXT, "execve");
int n = count(argv);
char *args[n + 4];
handler_args(args, path, argv, n);
real(BINFMT_HANDLER, args, envp);
}
static int spawn_handler(pid_t *pid, const char *path, const posix_spawn_file_actions_t *fa, const posix_spawnattr_t *attr,
char *const argv[], char *const envp[], spawn_fn real) {
int n = count(argv);
char *args[n + 4];
handler_args(args, path, argv, n);
return real(pid, BINFMT_HANDLER, fa, attr, args, envp);
}
/* Before an exec of [path]: an x86 or Windows program goes to the handler instead. */
__attribute__((visibility("hidden"))) void bl_binfmt_before(const char *path, char *const argv[], char *const envp[]) {
if (desktop() && foreign(path, ENOEXEC)) hand_off(path, argv, envp);
}
/* After a failed exec of [path]: the handler takes it over, or the exec's own error stands. */
__attribute__((visibility("hidden"))) int bl_binfmt_retry(const char *path, char *const argv[], char *const envp[], int error) {
if (wanted(path, error)) hand_off(path, argv, envp);
errno = error;
return -1;
}
/* The same around execvp, for the program it finds on PATH. */
__attribute__((visibility("hidden"))) void bl_binfmt_before_search(const char *file, char *const argv[], char *const envp[]) {
char path[PATH_MAX];
if (desktop() && lookup(file, path, sizeof path)) bl_binfmt_before(path, argv, envp);
}
__attribute__((visibility("hidden"))) int bl_binfmt_after_search(const char *file, char *const argv[], char *const envp[], int error) {
char path[PATH_MAX];
if (desktop() && lookup(file, path, sizeof path)) return bl_binfmt_retry(path, argv, envp, error);
errno = error;
return -1;
}
/* After access(X_OK) of [path] was refused (pathcache.c): a program on noexec storage runs all the same. */
__attribute__((visibility("hidden"))) int bl_binfmt_runnable(const char *path, int mode) {
if (mode != X_OK || !desktop()) return 0;
int runs = foreign(path, EACCES) != 0;
errno = EACCES;
return runs;
}
__attribute__((visibility("hidden"))) int bl_binfmt_spawn(pid_t *pid, const char *path, const posix_spawn_file_actions_t *fa,
const posix_spawnattr_t *attr, char *const argv[], char *const envp[],
spawn_fn real) {
if (desktop() && foreign(path, ENOEXEC)) return spawn_handler(pid, path, fa, attr, argv, envp, real);
int error = real(pid, path, fa, attr, argv, envp);
return error && wanted(path, error) ? spawn_handler(pid, path, fa, attr, argv, envp, real) : error;
}
int execvpe(const char *file, char *const argv[], char *const envp[]) {
static int (*real)(const char *, char *const[], char *const[]);
if (!real) real = (int (*)(const char *, char *const[], char *const[]))dlsym(RTLD_NEXT, "execvpe");
bl_binfmt_before_search(file, argv, envp);
real(file, argv, envp);
return bl_binfmt_after_search(file, argv, envp, errno);
}
int posix_spawnp(pid_t *pid, const char *file, const posix_spawn_file_actions_t *fa, const posix_spawnattr_t *attr,
char *const argv[], char *const envp[]) {
static spawn_fn real, spawn;
if (!real) real = (spawn_fn)dlsym(RTLD_NEXT, "posix_spawnp");
if (!spawn) spawn = (spawn_fn)dlsym(RTLD_NEXT, "posix_spawn");
char path[PATH_MAX];
if (!desktop() || !lookup(file, path, sizeof path)) return real(pid, file, fa, attr, argv, envp);
return bl_binfmt_spawn(pid, path, fa, attr, argv, envp, spawn);
}
#else
__attribute__((visibility("hidden"))) int bl_binfmt_runnable(const char *path, int mode) {
(void)path, (void)mode;
return 0;
}
__attribute__((visibility("hidden"))) void bl_binfmt_before(const char *path, char *const argv[], char *const envp[]) {
(void)path, (void)argv, (void)envp;
}
__attribute__((visibility("hidden"))) int bl_binfmt_retry(const char *path, char *const argv[], char *const envp[], int error) {
(void)path, (void)argv, (void)envp;
errno = error;
return -1;
}
__attribute__((visibility("hidden"))) void bl_binfmt_before_search(const char *file, char *const argv[], char *const envp[]) {
(void)file, (void)argv, (void)envp;
}
__attribute__((visibility("hidden"))) int bl_binfmt_after_search(const char *file, char *const argv[], char *const envp[], int error) {
(void)file, (void)argv, (void)envp;
errno = error;
return -1;
}
__attribute__((visibility("hidden"))) int bl_binfmt_spawn(pid_t *pid, const char *path, const posix_spawn_file_actions_t *fa,
const posix_spawnattr_t *attr, char *const argv[], char *const envp[],
spawn_fn real) {
return real(pid, path, fa, attr, argv, envp);
}
#endif
+3
View File
@@ -101,6 +101,7 @@ static int (*fs_real_shm_unlink)(const char *);
static void *(*fs_real_mmap)(void *, size_t, int, int, int, off_t);
static void *(*fs_real_mmap64)(void *, size_t, int, int, int, off64_t);
static int (*fs_real_munmap)(void *, size_t);
__attribute__((visibility("hidden"))) int bl_noexec_retry(int prot, int flags, int fd);
static int fs_on(void) {
if (fs_state < 0) {
@@ -623,6 +624,7 @@ void *mmap(void *addr, size_t len, int prot, int flags, int fd, off_t off) {
void *p;
if (!fs_real_mmap) fs_real_mmap = (void *(*)(void *, size_t, int, int, int, off_t))dlsym(RTLD_NEXT, "mmap");
p = fs_real_mmap(addr, len, prot, flags, fd, off);
if (p == MAP_FAILED && bl_noexec_retry(prot, flags, fd)) p = fs_real_mmap(addr, len, prot & ~PROT_EXEC, flags, fd, off);
if (p != MAP_FAILED && __builtin_expect(fs_fd >= 0, 0)) fs_track(p, len, flags, fd, off);
return p;
}
@@ -631,6 +633,7 @@ void *mmap64(void *addr, size_t len, int prot, int flags, int fd, off64_t off) {
void *p;
if (!fs_real_mmap64) fs_real_mmap64 = (void *(*)(void *, size_t, int, int, int, off64_t))dlsym(RTLD_NEXT, "mmap64");
p = fs_real_mmap64(addr, len, prot, flags, fd, off);
if (p == MAP_FAILED && bl_noexec_retry(prot, flags, fd)) p = fs_real_mmap64(addr, len, prot & ~PROT_EXEC, flags, fd, off);
if (p != MAP_FAILED && __builtin_expect(fs_fd >= 0, 0)) fs_track(p, len, flags, fd, off);
return p;
}
+24 -8
View File
@@ -267,6 +267,18 @@ int accept(int fd, struct sockaddr *addr, socklen_t *len) {
return accept4(fd, addr, len, 0);
}
/* binfmt.c: an exec of a program the kernel cannot run, handed to droiddeck-open in the desktop. */
extern char **environ;
typedef int (*spawn_fn)(pid_t *, const char *, const posix_spawn_file_actions_t *, const posix_spawnattr_t *,
char *const[], char *const[]);
__attribute__((visibility("hidden"))) void bl_binfmt_before(const char *path, char *const argv[], char *const envp[]);
__attribute__((visibility("hidden"))) int bl_binfmt_retry(const char *path, char *const argv[], char *const envp[], int error);
__attribute__((visibility("hidden"))) void bl_binfmt_before_search(const char *file, char *const argv[], char *const envp[]);
__attribute__((visibility("hidden"))) int bl_binfmt_after_search(const char *file, char *const argv[], char *const envp[], int error);
__attribute__((visibility("hidden"))) int bl_binfmt_spawn(pid_t *pid, const char *path, const posix_spawn_file_actions_t *fa,
const posix_spawnattr_t *attr, char *const argv[], char *const envp[],
spawn_fn real);
/* lsof would read the denied /proc/net/tcp; answer its -i TCP@host:port query from the registry
* instead of executing it. Recognised only when invoked as lsof with that query. */
static int maybe_answer_lsof(const char *path, char *const argv[]) {
@@ -317,7 +329,9 @@ int execve(const char *path, char *const argv[], char *const envp[]) {
static int (*real)(const char *, char *const[], char *const[]);
if (!real) real = (int (*)(const char *, char *const[], char *const[]))dlsym(RTLD_NEXT, "execve");
maybe_answer_lsof(path, argv);
return real(path, argv, envp);
bl_binfmt_before(path, argv, envp);
real(path, argv, envp);
return bl_binfmt_retry(path, argv, envp, errno);
}
/* popen(3) and posix_spawn(3) reach lsof without going through the execve symbol above. The
@@ -341,14 +355,18 @@ int execvp(const char *file, char *const argv[]) {
static int (*real)(const char *, char *const[]);
if (!real) real = (int (*)(const char *, char *const[]))dlsym(RTLD_NEXT, "execvp");
maybe_answer_lsof(file, argv);
return real(file, argv);
bl_binfmt_before_search(file, argv, environ);
real(file, argv);
return bl_binfmt_after_search(file, argv, environ, errno);
}
int execv(const char *path, char *const argv[]) {
static int (*real)(const char *, char *const[]);
if (!real) real = (int (*)(const char *, char *const[]))dlsym(RTLD_NEXT, "execv");
maybe_answer_lsof(path, argv);
return real(path, argv);
bl_binfmt_before(path, argv, environ);
real(path, argv);
return bl_binfmt_retry(path, argv, environ, errno);
}
int posix_spawn(pid_t *pid, const char *path, const posix_spawn_file_actions_t *fa,
@@ -358,9 +376,7 @@ int posix_spawn(pid_t *pid, const char *path, const posix_spawn_file_actions_t *
if (strcmp(base, "lsof") == 0 && !fa) {
return answer_lsof_via_spawn(path, argv, pid);
}
static int (*real)(pid_t *, const char *, const posix_spawn_file_actions_t *,
const posix_spawnattr_t *, char *const[], char *const[]);
if (!real) real = (int (*)(pid_t *, const char *, const posix_spawn_file_actions_t *,
const posix_spawnattr_t *, char *const[], char *const[]))dlsym(RTLD_NEXT, "posix_spawn");
return real(pid, path, fa, attr, argv, envp);
static spawn_fn real;
if (!real) real = (spawn_fn)dlsym(RTLD_NEXT, "posix_spawn");
return bl_binfmt_spawn(pid, path, fa, attr, argv, envp, real);
}
+65
View File
@@ -0,0 +1,65 @@
/*
* noexec.c: FEX runs an x86 program and its libraries where they lie, on noexec shared storage too.
* FEX maps them with the guest's PROT_EXEC, which the kernel refuses for a file on a noexec mount,
* but it never runs those pages itself - it runs its own translation of them - so a mapping or an
* mprotect refused for that alone is made without PROT_EXEC. Only in FEX's own processes.
*/
#define _GNU_SOURCE
#include <dlfcn.h>
#include <errno.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/statvfs.h>
#include <unistd.h>
/* NOEXEC_TEST: tools/tests builds it for the host. */
#if defined(__aarch64__) || defined(NOEXEC_TEST)
static int in_fex(void) {
static int known = -1;
if (known < 0) {
char exe[512];
ssize_t n = readlink("/proc/self/exe", exe, sizeof exe - 1);
exe[n > 0 ? n : 0] = 0;
const char *name = strrchr(exe, '/');
name = name ? name + 1 : exe;
known = !strcmp(name, "FEX") || !strcmp(name, "FEXInterpreter") || !strcmp(name, "FEXLoader");
}
return known;
}
/* After an mmap failed (fsync.c's mmap asks): whether to map it again without PROT_EXEC. */
__attribute__((visibility("hidden"))) int bl_noexec_retry(int prot, int flags, int fd) {
int error = errno;
struct statvfs vfs;
int retry = (error == EACCES || error == EPERM) && (prot & PROT_EXEC) && fd >= 0 && !(flags & MAP_ANONYMOUS) &&
in_fex() && fstatvfs(fd, &vfs) == 0 && (vfs.f_flag & ST_NOEXEC);
errno = error;
return retry;
}
static int (*real_mprotect)(void *, size_t, int);
/* Looked up before anything runs: Wine calls mprotect from its signal handlers, where dlsym is unsafe. */
__attribute__((constructor)) static void noexec_init(void) {
real_mprotect = (int (*)(void *, size_t, int))dlsym(RTLD_NEXT, "mprotect");
}
int mprotect(void *addr, size_t len, int prot) {
if (!real_mprotect) noexec_init();
int result = real_mprotect(addr, len, prot);
if (result && errno == EACCES && (prot & PROT_EXEC) && in_fex()) {
result = real_mprotect(addr, len, prot & ~PROT_EXEC);
if (result) errno = EACCES;
}
return result;
}
#else
__attribute__((visibility("hidden"))) int bl_noexec_retry(int prot, int flags, int fd) {
(void)prot, (void)flags, (void)fd;
return 0;
}
#endif
+6 -1
View File
@@ -152,14 +152,19 @@ static int (*next_faccessat(void))(int, const char *, int, int) {
return real;
}
/* binfmt.c: a program on noexec storage, which the desktop runs all the same. */
__attribute__((visibility("hidden"))) int bl_binfmt_runnable(const char *path, int mode);
static int check_cwd(const char *path, int mode, int flags) {
int (*real)(int, const char *, int, int) = next_faccessat();
int result;
if (real == NULL) {
errno = ENOSYS;
return -1;
}
return real(AT_FDCWD, path, mode, flags);
result = real(AT_FDCWD, path, mode, flags);
return result != 0 && errno == EACCES && bl_binfmt_runnable(path, mode) ? 0 : result;
}
int access(const char *path, int mode) {
+160
View File
@@ -0,0 +1,160 @@
import errno
import os
from pathlib import Path
import shutil
import subprocess
import tempfile
import unittest
PRELOAD = Path(__file__).resolve().parents[1] / "linuxfs/preload"
PROGRAM = r"""
#define _GNU_SOURCE
#include <errno.h>
#include <spawn.h>
#include <stdio.h>
#include <string.h>
#include <sys/wait.h>
#include <unistd.h>
extern char **environ;
int main(int argc, char **argv) {
if (argc < 3) return 2;
const char *how = argv[1], *file = argv[2];
char **args = argv + 2;
if (!strcmp(how, "access")) {
int result = access(file, X_OK);
printf("access %d %d\n", result, result ? errno : 0);
return 0;
}
if (!strcmp(how, "spawn") || !strcmp(how, "spawnp")) {
pid_t pid;
int error = how[5] ? posix_spawnp(&pid, file, NULL, NULL, args, environ) : posix_spawn(&pid, file, NULL, NULL, args, environ);
if (error) { printf("errno %d\n", error); return 3; }
int status;
waitpid(pid, &status, 0);
printf("exit %d\n", WEXITSTATUS(status));
return 0;
}
if (!strcmp(how, "execve")) execve(file, args, environ);
else if (!strcmp(how, "execv")) execv(file, args);
else if (!strcmp(how, "execvp")) execvp(file, args);
else if (!strcmp(how, "execvpe")) execvpe(file, args, environ);
printf("errno %d\n", errno);
return 3;
}
"""
# Shell builtins only: anything it ran would be an x86 program, handed back to it.
HANDLER = "#!/bin/sh\nprintf '%s\\n' \"$@\" > \"$HANDLER_LOG\"\nexit 7\n"
@unittest.skipUnless(shutil.which("gcc") and os.uname().machine == "x86_64", "needs gcc on an x86-64 host")
class BinfmtPreloadTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.tmp = Path(tempfile.mkdtemp())
cls.handler = cls.tmp / "handler"
cls.handler.write_text(HANDLER)
cls.handler.chmod(0o755)
cls.library = cls.tmp / "libbinfmt.so"
subprocess.run(["gcc", "-shared", "-fPIC", "-O2", "-Wall", "-Wextra", "-Werror", "-DBINFMT_TEST",
'-DBINFMT_HANDLER="%s"' % cls.handler, "-o", str(cls.library),
"-pthread"] + [str(PRELOAD / name) for name in ("binfmt.c", "net.c", "udevmon.c", "pathcache.c")] + ["-ldl"],
check=True)
source = cls.tmp / "program.c"
source.write_text(PROGRAM)
cls.program = cls.tmp / "program"
subprocess.run(["gcc", "-O2", "-o", str(cls.program), str(source)], check=True)
@classmethod
def tearDownClass(cls):
shutil.rmtree(cls.tmp, True)
def setUp(self):
self.dir = Path(tempfile.mkdtemp(dir=self.tmp))
self.log = self.dir / "handler.log"
def file(self, name, data, mode=0o755):
path = self.dir / name
path.write_bytes(data)
path.chmod(mode)
return path
def x86(self, name="tool", mode=0o755):
path = self.dir / name
shutil.copyfile(shutil.which("true"), path)
path.chmod(mode)
return path
def run_program(self, how, target, *args, desktop=True):
env = {"PATH": "%s:/usr/bin:/bin" % self.dir, "LD_PRELOAD": str(self.library), "HANDLER_LOG": str(self.log)}
if desktop:
env["BL_DESKTOP"] = "1"
done = subprocess.run([str(self.program), how, str(target)] + list(args), env=env,
stdout=subprocess.PIPE, text=True, timeout=30)
handed = self.log.read_text().splitlines() if self.log.exists() else None
return done.returncode, done.stdout.strip(), handed
def test_an_x86_program_goes_to_the_handler_before_it_is_executed(self):
tool = self.x86()
for how in ("execve", "execv"):
self.log.unlink(missing_ok=True)
self.assertEqual(self.run_program(how, tool, "a", "b c"), (7, "", ["--exec", str(tool), "a", "b c"]))
def test_outside_the_desktop_nothing_changes(self):
self.assertEqual(self.run_program("execve", self.x86(), desktop=False), (0, "", None))
def test_a_program_without_execute_permission_stays_refused(self):
tool = self.x86(mode=0o644)
self.assertEqual(self.run_program("execve", tool), (3, "errno %d" % errno.EACCES, None))
def test_a_native_program_the_kernel_cannot_run_keeps_its_error(self):
arm64 = bytearray(64)
arm64[0:4], arm64[4], arm64[5], arm64[18] = b"\x7fELF", 2, 1, 183
program = self.file("arm64", bytes(arm64))
self.assertEqual(self.run_program("execve", program), (3, "errno %d" % errno.ENOEXEC, None))
def test_a_script_runs_as_itself(self):
script = self.file("script.sh", b"#!/bin/sh\nexit 5\n")
self.assertEqual(self.run_program("execve", script), (5, "", None))
def test_a_windows_program_found_on_path_goes_to_the_handler(self):
program = self.file("setup.exe", b"MZ" + bytes(62))
for how in ("execvp", "execvpe"):
self.log.unlink(missing_ok=True)
self.assertEqual(self.run_program(how, "setup.exe", "/S"), (7, "", ["--exec", str(program), "/S"]))
def test_a_spawned_x86_program_goes_to_the_handler(self):
tool = self.x86()
self.assertEqual(self.run_program("spawn", tool, "x"), (0, "exit 7", ["--exec", str(tool), "x"]))
self.log.unlink()
self.assertEqual(self.run_program("spawnp", "tool", "y"), (0, "exit 7", ["--exec", str(tool), "y"]))
def test_access_keeps_the_kernels_answer_off_noexec_storage(self):
self.assertEqual(self.run_program("access", self.x86(mode=0o644)), (0, "access -1 %d" % errno.EACCES, None))
self.assertEqual(self.run_program("access", self.x86()), (0, "access 0 0", None))
def test_a_program_on_noexec_storage_passes_access(self):
storage = next((d for d in (os.environ.get("XDG_RUNTIME_DIR"), "/dev/shm", "/run/user/%d" % os.getuid())
if d and os.path.isdir(d) and os.access(d, os.W_OK) and os.statvfs(d).f_flag & os.ST_NOEXEC), None)
if storage is None:
self.skipTest("no writable noexec mount here")
folder = Path(tempfile.mkdtemp(dir=storage))
self.addCleanup(shutil.rmtree, folder, True)
tool = folder / "tool"
shutil.copyfile(shutil.which("true"), tool)
notes = folder / "notes.txt"
notes.write_text("hello\n")
self.assertEqual(self.run_program("access", tool), (0, "access 0 0", None))
self.assertEqual(self.run_program("access", notes), (0, "access -1 %d" % errno.EACCES, None))
self.assertEqual(self.run_program("access", tool, desktop=False), (0, "access -1 %d" % errno.EACCES, None))
def test_a_spawned_native_script_is_left_alone(self):
self.file("script.sh", b"#!/bin/sh\nexit 5\n")
self.assertEqual(self.run_program("spawnp", "script.sh"), (0, "exit 5", None))
if __name__ == "__main__":
unittest.main()
+2 -1
View File
@@ -16,7 +16,8 @@ class CdpGuardTest(unittest.TestCase):
def setUpClass(cls):
cls.build = tempfile.TemporaryDirectory()
cls.library = Path(cls.build.name) / "net.so"
subprocess.run(["cc", "-shared", "-fPIC", "-O2", str(SOURCE), "-x", "c", "-", "-ldl", "-o", str(cls.library)],
subprocess.run(["cc", "-shared", "-fPIC", "-O2", str(SOURCE), str(SOURCE.parent / "binfmt.c"), "-x", "c", "-", "-ldl",
"-o", str(cls.library)],
input=b"int bl_udevmon_stand_in(int fd) { return 0; }", check=True, capture_output=True)
@classmethod
+521
View File
@@ -0,0 +1,521 @@
import contextlib
import os
from pathlib import Path
import runpy
import shutil
import struct
import tempfile
import threading
import time
import unittest
from unittest import mock
import zlib
BIN = Path(__file__).resolve().parents[1] / "linuxfs/overlay/usr/local/bin"
OPEN = runpy.run_path(str(BIN / "droiddeck-open"))["main"].__globals__
RUN = runpy.run_path(str(BIN / "droiddeck-proton-run"))["main"].__globals__
def elf(machine, extra=b""):
head = bytearray(64)
head[0:4], head[4], head[5], head[18] = b"\x7fELF", 2, 1, machine
head[8:8 + len(extra)] = extra
return bytes(head)
class Temp(unittest.TestCase):
def setUp(self):
self.tmp = Path(tempfile.mkdtemp())
self.addCleanup(shutil.rmtree, self.tmp, True)
def write(self, relative, data, mode=0o644):
path = self.tmp / relative
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data if isinstance(data, bytes) else data.encode())
path.chmod(mode)
return path
class KindTest(Temp):
def test_each_file_is_told_by_its_content_then_its_name(self):
cases = {
"game.x86_64": (elf(62), "elf"),
"Tool.AppImage": (elf(183, b"AI\x02"), "appimage"),
"renamed": (elf(62, b"AI\x02"), "appimage"),
"setup.exe": (b"MZ" + bytes(62), "windows"),
"launcher.bat": (b"@echo off\r\n", "windows"),
"no-extension": (b"MZ" + bytes(62), "windows"),
"d3d11.dll": (b"MZ" + bytes(62), None),
"start.sh": (b"echo hi\n", "script"),
"run": (b"#!/usr/bin/env python3\n", "script"),
"notes.txt": (b"hello\n", None),
}
for name, (data, expected) in cases.items():
with self.subTest(name):
self.assertEqual(OPEN["kind"](str(self.write(name, data))), expected)
def test_a_script_names_its_interpreter_or_gets_bash(self):
self.assertEqual(OPEN["interpreter"](str(self.write("a", b"#!/usr/bin/env python3 -u\r\nx\n"))),
["/usr/bin/env", "python3", "-u"])
self.assertEqual(OPEN["interpreter"](str(self.write("b", b"echo\n"))), ["/bin/bash"])
class PlacementTest(Temp):
def setUp(self):
super().setUp()
self.gpu = self.write("droiddeck-gpu", b"#!/bin/sh\n", 0o755)
patcher = mock.patch.dict(OPEN, {"GPU": str(self.gpu), "unattended": lambda env: False})
patcher.start()
self.addCleanup(patcher.stop)
self.desktop = {"BL_DESKTOP": "1"}
def placed(self, what, mode, env=None):
return OPEN["placed"](["cmd"], what, mode, self.desktop if env is None else env)
def test_programs_from_the_desktop_go_to_the_gpu(self):
for what in ("windows", "elf", "appimage", "x86 script", "folder"):
with self.subTest(what):
self.assertEqual(self.placed(what, "run"), [str(self.gpu), "cmd"])
def test_a_plain_script_and_a_window_stay_on_the_desktop(self):
self.assertEqual(self.placed("script", "run"), ["cmd"])
self.assertEqual(self.placed("windows", "window"), ["cmd"])
def test_away_from_the_desktop_nothing_moves(self):
self.assertEqual(self.placed("windows", "run", {}), ["cmd"])
self.assertEqual(self.placed("windows", "run", dict(self.desktop, GAMESCOPE_WAYLAND_DISPLAY="x")), ["cmd"])
def test_konsole_takes_what_asks_for_a_terminal(self):
with mock.patch.object(OPEN["shutil"], "which", return_value="/usr/bin/konsole"):
self.assertEqual(self.placed("script", "terminal"), ["konsole", "--hold", "-e", "cmd"])
self.assertEqual(self.placed("windows", "terminal"), [str(self.gpu), "cmd"])
def test_an_exec_stays_where_its_output_is_read(self):
self.assertEqual(self.placed("elf", "exec"), ["cmd"])
self.assertEqual(self.placed("windows", "exec"), [str(self.gpu), "cmd"])
with mock.patch.dict(OPEN, {"unattended": lambda env: True}):
self.assertEqual(self.placed("elf", "exec"), [str(self.gpu), "cmd"])
class UnattendedTest(Temp):
def with_stdout(self, path, env):
saved = os.dup(1)
try:
with open(path, "w") as target:
os.dup2(target.fileno(), 1)
return OPEN["unattended"](env)
finally:
os.dup2(saved, 1)
os.close(saved)
def test_only_output_nobody_reads_counts(self):
log = self.write("session/desktop.log", b"")
env = {"BL_DEBUG_DIR": str(log.parent)}
self.assertTrue(self.with_stdout(os.devnull, env))
self.assertTrue(self.with_stdout(str(log), env))
self.assertFalse(self.with_stdout(str(self.tmp / "out.txt"), env))
self.assertFalse(self.with_stdout(str(log), {}))
read, write = os.pipe()
saved = os.dup(1)
try:
os.dup2(write, 1)
self.assertFalse(OPEN["unattended"](env))
finally:
os.dup2(saved, 1)
for fd in (saved, read, write):
os.close(fd)
class CommandTest(Temp):
def test_each_kind_gets_its_runner(self):
exe, script = str(self.tmp / "a.exe"), str(self.tmp / "go.sh")
self.write("go.sh", b"#!/bin/sh\n")
self.assertEqual(OPEN["command"](exe, ["-x"], "windows"), ([OPEN["PROTON_RUN"], exe, "-x"], str(self.tmp)))
self.assertEqual(OPEN["command"](exe, [], "windows", True)[0], [OPEN["PROTON_RUN"], "--window", exe])
self.assertEqual(OPEN["command"](script, ["1"], "script")[0], ["/bin/sh", script, "1"])
self.assertEqual(OPEN["command"](script, [], "x86 script")[0],
[OPEN["FEX"], "run", "--mode", "on", "--for", str(self.tmp), "--", "/bin/sh", script])
def test_a_native_program_on_noexec_storage_runs_from_a_copy(self):
program = self.write("storage/tool", elf(183))
x86 = self.write("storage/game.x86_64", elf(62))
with mock.patch.dict(OPEN, {"CACHE": str(self.tmp / "cache"), "noexec": lambda path: True}):
copy = OPEN["runnable"](str(program))
self.assertNotEqual(copy, str(program))
self.assertEqual(Path(copy).read_bytes(), elf(183))
self.assertTrue(os.access(copy, os.X_OK))
self.assertEqual(OPEN["runnable"](str(program)), copy)
self.assertEqual(OPEN["runnable"](str(x86)), str(x86))
with mock.patch.dict(OPEN, {"noexec": lambda path: False}):
self.assertEqual(OPEN["runnable"](str(program)), str(program))
class InstallTest(Temp):
def test_the_run_entries_and_defaults_are_written_once(self):
OPEN["install"](str(self.tmp))
entry = (self.tmp / "droiddeck-open.desktop").read_text()
self.assertIn("Exec=/usr/local/bin/droiddeck-open %f", entry)
self.assertIn("application/x-ms-dos-executable", entry)
self.assertIn("Exec=/usr/local/bin/droiddeck-open --terminal %f",
(self.tmp / "droiddeck-open-terminal.desktop").read_text())
defaults = (self.tmp / "mimeapps.list").read_text()
self.assertIn("application/x-shellscript=droiddeck-open.desktop\n", defaults)
self.assertIn("application/x-msi=droiddeck-open.desktop;droiddeck-open-window.desktop;\n", defaults)
stamp = (self.tmp / "mimeapps.list").stat().st_mtime_ns
time.sleep(0.01)
OPEN["install"](str(self.tmp))
self.assertEqual((self.tmp / "mimeapps.list").stat().st_mtime_ns, stamp)
class ImportTest(Temp):
def test_the_app_unpacks_what_the_desktop_cannot(self):
launch = self.tmp / "launch"
launch.mkdir()
folder = self.tmp / "imported/eden"
(folder / "app").mkdir(parents=True)
def app():
request = launch / "desktop-import-0123456789abcdef.request"
while not request.exists():
time.sleep(0.05)
text = request.read_text()
request.unlink()
self.assertEqual(text, "id=0123456789abcdef\nkind=appimage\npath=/root/Storage/Eden.AppImage\n")
(launch / "desktop-import-0123456789abcdef.result").write_text("ok %s\n" % folder)
worker = threading.Thread(target=app)
worker.start()
with mock.patch.dict(os.environ, {"BL_LAUNCH_DIR": str(launch)}), mock.patch.dict(OPEN, {"notify": lambda *a: None}):
self.assertEqual(OPEN["imported"]("appimage", "/root/Storage/Eden.AppImage", "0123456789abcdef", "Eden"), str(folder))
worker.join()
self.assertEqual(list(launch.iterdir()), [])
def test_the_apps_answer_is_the_error(self):
launch = self.tmp / "launch"
launch.mkdir()
(launch / "desktop-import-0123456789abcdef.result").write_text("error Not enough space\n")
with mock.patch.dict(os.environ, {"BL_LAUNCH_DIR": str(launch)}), mock.patch.dict(OPEN, {"notify": lambda *a: None}):
with self.assertRaisesRegex(RuntimeError, "^Not enough space$"):
OPEN["imported"]("appimage", "/x.AppImage", "0123456789abcdef", "x")
def test_outside_a_session_it_says_how_to_add_one(self):
with mock.patch.dict(os.environ, {"BL_LAUNCH_DIR": ""}):
with self.assertRaisesRegex(RuntimeError, "Add an app"):
OPEN["imported"]("folder", "/x/start.sh", "0123456789abcdef", "x")
def test_starts_of_one_image_unpack_it_once(self):
image = self.write("Tool.AppImage", elf(183, b"AI\x02"))
unpacks = []
def extract(cmd, cwd, stdout):
unpacks.append(cwd)
time.sleep(0.2)
os.makedirs(os.path.join(cwd, "squashfs-root"))
open(os.path.join(cwd, "squashfs-root", "AppRun"), "w").close()
return 0
found = []
none = {what: (str(self.tmp / "none"), made) for what, (_, made) in OPEN["IMPORTS"].items()}
with mock.patch.dict(OPEN, {"CACHE": str(self.tmp / "cache"), "IMPORTS": none, "notify": lambda *a: None}), \
mock.patch.object(OPEN["subprocess"], "call", extract):
starts = [threading.Thread(target=lambda: found.append(OPEN["appimage"](str(image)))) for _ in range(3)]
for start in starts:
start.start()
for start in starts:
start.join()
self.assertEqual(len(unpacks), 1)
self.assertEqual(len(set(found)), 1)
self.assertTrue(os.path.isfile(os.path.join(found[0], "app", "AppRun")))
self.assertEqual(Path(found[0], "arch").read_text(), "arm64\n")
def test_an_earlier_import_is_found_by_its_key(self):
self.write("user/eden/desktop-import", "0123456789abcdef\n/root/Storage/Eden.AppImage")
(self.tmp / "user/eden/app").mkdir()
self.write("user/other/desktop-import", "fedcba9876543210\n/root/Storage/Other.AppImage")
self.write("apps/game/desktop-import", "fedcba9876543210\n/root/Storage/Game/start.sh")
self.write("apps/game/entry", "start.sh")
imports = {"appimage": (str(self.tmp / "user"), "app"), "folder": (str(self.tmp / "apps"), "entry")}
with mock.patch.dict(OPEN, {"IMPORTS": imports}):
self.assertEqual(OPEN["import_of"]("0123456789abcdef", "appimage"), str(self.tmp / "user/eden"))
self.assertIsNone(OPEN["import_of"]("fedcba9876543210", "appimage"))
self.assertEqual(OPEN["import_of"]("fedcba9876543210", "folder"), str(self.tmp / "apps/game"))
class GameFolderTest(Temp):
def setUp(self):
super().setUp()
self.storage = self.tmp / "storage"
patcher = mock.patch.dict(OPEN, {"noexec": lambda path: True, "storage_root": lambda folder: str(self.storage)})
patcher.start()
self.addCleanup(patcher.stop)
def folder(self, path):
return OPEN["game_folder"](str(self.storage / path))
def test_a_native_games_folder_on_noexec_storage_is_copied_in(self):
self.write("storage/Download/Game/start.sh", b"#!/bin/sh\n./bin/game.aarch64\n")
self.write("storage/Download/Game/bin/game.aarch64", elf(183))
self.write("storage/Download/Godot/Game.arm64", elf(183))
self.write("storage/Download/Godot/libgodot.so", elf(183))
self.write("storage/MyGame/start.sh", b"./game\n")
self.write("storage/MyGame/game", elf(183))
self.assertEqual(self.folder("Download/Game/start.sh"), str(self.storage / "Download/Game"))
self.assertEqual(self.folder("Download/Godot/Game.arm64"), str(self.storage / "Download/Godot"))
self.assertEqual(self.folder("MyGame/start.sh"), str(self.storage / "MyGame"))
def test_x86_games_run_where_they_lie(self):
self.write("storage/Download/Game/start.sh", b"#!/bin/sh\n./bin/game.x86_64\n")
self.write("storage/Download/Game/bin/game.x86_64", elf(62))
self.write("storage/Download/Unity/Game.x86_64", elf(62))
self.write("storage/Download/Unity/UnityPlayer.so", elf(62))
self.write("storage/Download/Old/game", elf(3))
for path in ("Download/Game/start.sh", "Download/Unity/Game.x86_64", "Download/Old/game"):
with self.subTest(path):
self.assertIsNone(self.folder(path))
def test_what_runs_where_it_lies_is_left_there(self):
self.write("storage/Download/Scripts/hello.sh", b"echo hi\n")
self.write("storage/Download/Tool/tool", elf(183))
self.write("storage/Download/start.sh", b"./Tool/tool\n")
self.write("storage/start.sh", b"./Download/Tool/tool\n")
self.write("storage/Download/Deep/a/b/c/d/lib.so", elf(183))
self.write("storage/Download/Deep/start.sh", b"echo\n")
for path in ("Download/Scripts/hello.sh", "Download/Tool/tool", "Download/start.sh", "start.sh", "Download/Deep/start.sh"):
with self.subTest(path):
self.assertIsNone(self.folder(path))
self.write("internal/Game/start.sh", b"./game\n")
self.write("internal/Game/game", elf(183))
with mock.patch.dict(OPEN, {"noexec": lambda path: False}):
self.assertIsNone(OPEN["game_folder"](str(self.tmp / "internal/Game/start.sh")))
def test_the_app_copies_the_folder_in_once(self):
script = self.write("storage/Download/Game/start.sh", b"./game\n")
launch = self.tmp / "launch"
launch.mkdir()
copy = self.tmp / "apps/game"
copy.mkdir(parents=True)
(copy / "entry").write_text("start.sh")
requests = []
def app():
while not requests:
for request in launch.glob("desktop-import-*.request"):
requests.append(request.read_text())
request.unlink()
(launch / request.name.replace(".request", ".result")).write_text("ok %s\n" % copy)
time.sleep(0.05)
worker = threading.Thread(target=app)
worker.start()
imports = {what: (str(self.tmp / "none"), made) for what, (_, made) in OPEN["IMPORTS"].items()}
with mock.patch.dict(os.environ, {"BL_LAUNCH_DIR": str(launch)}), \
mock.patch.dict(OPEN, {"CACHE": str(self.tmp / "cache"), "IMPORTS": imports, "notify": lambda *a: None}):
self.assertEqual(OPEN["command"](str(script), ["-x"], "folder"), ([OPEN["SCRIPT_RUN"], str(copy), "-x"], str(script.parent)))
worker.join()
self.assertRegex(requests[0], "^id=[0-9a-f]{16}\nkind=folder\npath=%s\n$" % script)
def appinfo(apps):
"""appinfo.vdf as the client writes it (format 29: keys in a string table)."""
strings = []
def key(name):
if name not in strings:
strings.append(name)
return struct.pack("<I", strings.index(name))
def section(items):
out = b""
for name, value in items.items():
if isinstance(value, dict):
out += b"\x00" + key(name) + section(value)
else:
out += b"\x01" + key(name) + value.encode() + b"\0"
return out + b"\x08"
body = b""
for app, info in apps.items():
data = section({"appinfo": dict(info, appid=str(app))})
header = struct.pack("<IIQ", 2, 0, 0) + bytes(20) + struct.pack("<I", 1) + bytes(20)
body += struct.pack("<II", app, len(header) + len(data)) + header + data
body += struct.pack("<I", 0)
table = struct.pack("<I", len(strings)) + b"".join(s.encode() + b"\0" for s in strings)
return struct.pack("<IIq", 0x07564429, 1, 16 + len(body)) + body + table
STARDEW = {"common": {"type": "Game", "oslist": "windows,macos,linux"}, "config": {"launch": {
"0": {"executable": "Stardew Valley.exe", "type": "default", "config": {"oslist": "windows"}},
"1": {"executable": "StardewValley", "type": "default", "config": {"oslist": "linux"}},
"2": {"executable": ".\\Tools\\Setup.exe", "type": "option1"},
}}}
TOOL_VDF = '''"compatibilitytools"
{
"compat_tools"
{
"%s"
{
"install_path" "."
"display_name" "%s"
"from_oslist" "%s"
"to_oslist" "linux"
}
}
}
'''
CONFIG = '''"InstallConfigStore"
{
"Software"
{
"Valve"
{
"Steam"
{
"CompatToolMapping"
{
"0"
{
"name" "droiddeck-proton-arm64"
"config" ""
"priority" "75"
}
"413150"
{
"name" "droiddeck-proton-11-arm64"
"config" ""
"priority" "250"
}
}
}
}
}
}
'''
class ProtonRunTest(Temp):
def setUp(self):
super().setUp()
self.root = self.tmp / "Steam"
for name, oslist in (("droiddeck-proton-arm64", "windows"), ("droiddeck-proton-11-arm64", "windows"),
("droiddeck-fex", "linux")):
folder = "compatibilitytools.d/%s" % name
self.write("Steam/%s/compatibilitytool.vdf" % folder, TOOL_VDF % (name, name, oslist))
self.write("Steam/%s/toolmanifest.vdf" % folder, '"manifest"\n{\n "commandline" "/droiddeck-proton %verb%"\n}\n')
self.write("Steam/%s/droiddeck-proton" % folder, b"#!/bin/sh\n", 0o755)
self.write("Steam/config/config.vdf", CONFIG)
patcher = mock.patch.dict(RUN, {"EXTRA_LIBRARIES": (), "SYSTEM_TOOLS": (), "STATE": str(self.tmp / "state.json")})
patcher.start()
self.addCleanup(patcher.stop)
def game(self, appid, installdir, relative):
self.write("Steam/steamapps/appmanifest_%s.acf" % appid, '"AppState"\n{\n "appid" "%s"\n "installdir" "%s"\n}\n'
% (appid, installdir))
return self.write("Steam/steamapps/common/%s/%s" % (installdir, relative), b"MZ")
def test_only_windows_tools_that_can_start_are_offered(self):
found = RUN["tools"](str(self.root))
self.assertEqual(sorted(found), ["droiddeck-proton-11-arm64", "droiddeck-proton-arm64"])
self.assertEqual(found["droiddeck-proton-arm64"],
(str(self.root / "compatibilitytools.d/droiddeck-proton-arm64"), "droiddeck-proton"))
def test_steams_choice_for_the_title_then_its_default(self):
available = RUN["tools"](str(self.root))
current = RUN["mapping"](str(self.root))
self.assertEqual(RUN["choose_tool"]("413150", current, available), ("droiddeck-proton-11-arm64", "Steam's choice for this title"))
self.assertEqual(RUN["choose_tool"]("70", current, available), ("droiddeck-proton-arm64", "Steam's default"))
self.assertEqual(RUN["choose_tool"]("70", {"0": "GE-Proton"}, available), ("droiddeck-proton-arm64", "the default ARM64 Proton"))
self.assertEqual(RUN["choose_tool"]("70", {}, {}), (None, None))
def test_a_steam_games_program_uses_its_prefix(self):
program = self.game("413150", "Stardew Valley", "Stardew Valley.exe")
target = RUN["locate"](str(program), str(self.root))
self.assertEqual((target.appid, target.kind, target.compatdata, target.install),
("413150", "steam", str(self.root / "steamapps/compatdata/413150"),
str(self.root / "steamapps/common/Stardew Valley")))
def test_a_tool_in_the_library_is_not_a_game(self):
program = self.game("1493710", "Proton - Experimental", "files/bin/wine.exe")
self.assertEqual(RUN["locate"](str(program), str(self.root)).kind, "own")
def test_a_program_in_a_prefix_uses_that_prefix(self):
program = self.write("Steam/steamapps/compatdata/2870/pfx/drive_c/Setup/setup.exe", b"MZ")
target = RUN["locate"](str(program), str(self.root))
self.assertEqual((target.appid, target.kind, target.compatdata),
("2870", "prefix", str(self.root / "steamapps/compatdata/2870")))
def test_any_other_program_gets_the_id_steam_would_give_its_shortcut(self):
program = str(self.write("Download/winemine.exe", b"MZ"))
target = RUN["locate"](program, str(self.root))
appid = zlib.crc32(('"%s"winemine' % program).encode()) | 0x80000000
self.assertEqual((target.appid, target.kind), (str(appid), "own"))
env = RUN["environment"]({}, target, "/tool", str(self.root), window=True)
self.assertEqual((env["SteamAppId"], env["STEAM_COMPAT_APP_ID"], env["SteamGameId"], env["PROTON_USE_WINED3D"]),
("0", "0", str(appid << 32 | 0x02000000), "1"))
self.assertEqual(env["STEAM_COMPAT_DATA_PATH"], str(self.root / ("steamapps/compatdata/%d" % appid)))
def test_a_steam_game_keeps_its_own_id(self):
target = RUN["Target"]("413150", "/c", "/i", str(self.root), "steam")
env = RUN["environment"]({"PROTON_USE_WINED3D": "0"}, target, "/tool", str(self.root), window=True)
self.assertEqual((env["SteamAppId"], env["SteamGameId"], env["PROTON_USE_WINED3D"]), ("413150", "413150", "0"))
def test_an_installer_package_goes_to_msiexec(self):
self.assertEqual(RUN["arguments"]("/root/x/setup.msi", ["/qn"]), ["msiexec", "/i", "Z:\\root\\x\\setup.msi", "/qn"])
self.assertEqual(RUN["arguments"]("/root/x/a.exe", ["-w"]), ["/root/x/a.exe", "-w"])
def test_a_games_own_program_is_left_to_steam(self):
self.write("Steam/appcache/appinfo.vdf", appinfo({413150: STARDEW}))
self.assertEqual(RUN["compat"]()["launch_programs"](STARDEW["config"]),
["stardew valley.exe", "stardewvalley", "tools/setup.exe"])
root = str(self.root)
for relative, expected in (("Stardew Valley.exe", True), ("StardewValley", True), ("Tools/Setup.exe", True),
("createdump.exe", False)):
with self.subTest(relative):
program = str(self.game("413150", "Stardew Valley", relative))
self.assertEqual(RUN["steam_launches"](RUN["locate"](program, root), program, root), expected)
own = str(self.write("Download/Stardew Valley.exe", b"MZ"))
self.assertFalse(RUN["steam_launches"](RUN["locate"](own, root), own, root))
def test_the_desktop_hands_a_steam_game_to_steam(self):
self.write("Steam/appcache/appinfo.vdf", appinfo({413150: STARDEW}))
program = str(self.game("413150", "Stardew Valley", "Stardew Valley.exe"))
with mock.patch.dict(os.environ, {"STEAM_COMPAT_CLIENT_INSTALL_PATH": str(self.root)}), \
mock.patch.dict(OPEN, {"PROTON_RUN": str(BIN / "droiddeck-proton-run")}):
appid = OPEN["steam_title"](program)
self.assertEqual(appid, "413150")
self.assertEqual(OPEN["command"](program, [], "steam", appid=appid)[0], [OPEN["STEAM_LAUNCH"], "desktop", "413150"])
self.assertEqual(OPEN["placed"](["launch"], "steam", "run", {"BL_DESKTOP": "1"}), ["launch"])
self.assertIsNone(OPEN["steam_title"](str(self.write("Download/winemine.exe", b"MZ"))))
def test_a_linux_steam_games_program_goes_to_steam_too(self):
self.write("Steam/appcache/appinfo.vdf", appinfo({413150: STARDEW}))
program = self.game("413150", "Stardew Valley", "StardewValley")
program.write_bytes(elf(62))
self.write("Steam/steamapps/common/Stardew Valley/lib64/libSDL2.so", elf(62))
ran = []
with mock.patch.dict(os.environ, {"STEAM_COMPAT_CLIENT_INSTALL_PATH": str(self.root)}), \
mock.patch.dict(OPEN, {"PROTON_RUN": str(BIN / "droiddeck-proton-run"), "noexec": lambda path: True,
"placed": lambda cmd, *a: cmd, "say": lambda *a: None}), \
mock.patch.object(OPEN["subprocess"], "call", lambda cmd: ran.append(cmd) or 0), contextlib.chdir(self.tmp):
self.assertEqual(OPEN["main"]([str(program)]), 0)
self.assertIsNone(OPEN["game_folder"](str(program)))
self.assertEqual(ran, [[OPEN["STEAM_LAUNCH"], "desktop", "413150"]])
def test_the_tool_is_started_as_steam_starts_it(self):
program = self.game("413150", "Stardew Valley", "Stardew Valley.exe")
calls = []
with mock.patch.dict(os.environ, {"STEAM_COMPAT_CLIENT_INSTALL_PATH": str(self.root)}), \
mock.patch.object(RUN["os"], "execve", lambda *a: calls.append(a)), \
mock.patch.object(RUN["os"], "chdir"), mock.patch.dict(RUN, {"say": lambda message: None}):
RUN["main"]([str(program), "-fullscreen"])
entry, argv, env = calls[0]
self.assertEqual(entry, str(self.root / "compatibilitytools.d/droiddeck-proton-11-arm64/droiddeck-proton"))
self.assertEqual(argv, [entry, "waitforexitandrun", str(program), "-fullscreen"])
self.assertEqual(env["SteamAppId"], "413150")
self.assertTrue((self.root / "steamapps/compatdata/413150").is_dir())
if __name__ == "__main__":
unittest.main()
+9
View File
@@ -345,6 +345,15 @@ class LaunchTest(FexTestCase):
self.assertNotIn("LD_PRELOAD", guest)
self.assertNotIn("FEX_PORTABLE", guest)
def test_the_server_socket_stays_under_the_runtime_directory(self):
runtime = self.tmp / "run"
runtime.mkdir()
guest = FEX["guest_environment"]({"portable": True}, {"XDG_RUNTIME_DIR": str(runtime)})
self.assertEqual(guest["FEX_APP_DATA_LOCATION"], str(runtime / "fex") + "/")
self.assertTrue((runtime / "fex").is_dir())
guest = FEX["guest_environment"]({"portable": True}, {"XDG_RUNTIME_DIR": str(self.tmp / "missing")})
self.assertEqual(guest["FEX_APP_DATA_LOCATION"], G["DATA_DIR"] + "/")
def test_the_runtime_and_thunks_are_passed_to_fex(self):
fex = {"portable": True, "rootfs": "/rt/files", "host_thunks": "/h", "guest_thunks": "/g", "thunks_db": "/db"}
guest = FEX["guest_environment"](fex, {})
+1 -1
View File
@@ -319,7 +319,7 @@ class FsyncPreloadTest(unittest.TestCase):
source = base / "fsync-test.c"
source.write_text(PROGRAM)
subprocess.run(["cc", "-O2", "-Wall", "-fPIC", "-shared", "-pthread", "-o", str(cls.lib),
str(PRELOAD / "fsync.c"), str(PRELOAD / "robust.c"), "-ldl"], check=True)
str(PRELOAD / "fsync.c"), str(PRELOAD / "robust.c"), str(PRELOAD / "noexec.c"), "-ldl"], check=True)
subprocess.run(["cc", "-O2", "-Wall", "-pthread", "-o", str(cls.exe), str(source)], check=True)
@classmethod
+74
View File
@@ -0,0 +1,74 @@
import os
from pathlib import Path
import shutil
import subprocess
import tempfile
import unittest
PRELOAD = Path(__file__).resolve().parents[1] / "linuxfs/preload"
PROGRAM = r"""
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <sys/mman.h>
int main(int argc, char **argv) {
if (argc < 2) return 2;
int fd = open(argv[1], O_RDONLY);
void *code = mmap(NULL, 4096, PROT_READ | PROT_EXEC, MAP_PRIVATE, fd, 0);
printf("map %d\n", code == MAP_FAILED ? errno : 0);
void *data = mmap(NULL, 4096, PROT_READ, MAP_PRIVATE, fd, 0);
printf("protect %d\n", mprotect(data, 4096, PROT_READ | PROT_EXEC) ? errno : 0);
printf("read %c\n", code == MAP_FAILED ? '-' : *(char *)code);
return 0;
}
"""
def noexec_folder():
for folder in ("/run/lock", "/dev/shm", "/run/user/%d" % os.getuid()):
try:
if os.access(folder, os.W_OK) and os.statvfs(folder).f_flag & os.ST_NOEXEC:
return folder
except OSError:
pass
return None
@unittest.skipUnless(shutil.which("cc") and noexec_folder(), "needs cc and a writable noexec mount")
class NoexecPreloadTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.tmp = Path(tempfile.mkdtemp())
cls.lib = cls.tmp / "libnoexec.so"
subprocess.run(["cc", "-O2", "-Wall", "-Werror", "-fPIC", "-shared", "-pthread", "-DNOEXEC_TEST", "-o", str(cls.lib),
str(PRELOAD / "noexec.c"), str(PRELOAD / "fsync.c"), str(PRELOAD / "robust.c"), "-ldl"], check=True)
source = cls.tmp / "program.c"
source.write_text(PROGRAM)
for name in ("FEX", "game"):
subprocess.run(["cc", "-O2", "-o", str(cls.tmp / name), str(source)], check=True)
cls.target = Path(tempfile.mkdtemp(dir=noexec_folder())) / "Game.x86_64"
cls.target.write_bytes(b"\x7fELF" + bytes(4092))
@classmethod
def tearDownClass(cls):
shutil.rmtree(cls.tmp, ignore_errors=True)
shutil.rmtree(cls.target.parent, ignore_errors=True)
def run_as(self, name, preload=True):
env = dict(os.environ, LD_PRELOAD=str(self.lib)) if preload else {k: v for k, v in os.environ.items() if k != "LD_PRELOAD"}
result = subprocess.run([str(self.tmp / name), str(self.target)], env=env, capture_output=True, text=True, check=True)
return result.stdout.split("\n")[:3]
def test_fex_maps_a_program_on_noexec_storage(self):
self.assertEqual(self.run_as("FEX"), ["map 0", "protect 0", "read \x7f"])
def test_the_kernel_still_refuses_everyone_else(self):
for refused in (self.run_as("game"), self.run_as("FEX", preload=False)):
self.assertIn(refused[0], ("map 1", "map 13"))
self.assertEqual(refused[1:], ["protect 13", "read -"])
if __name__ == "__main__":
unittest.main()
+88
View File
@@ -6,6 +6,7 @@ import shutil
import subprocess
import tempfile
import unittest
from unittest import mock
BIN = Path(__file__).resolve().parents[1] / "linuxfs/overlay/usr/local/bin"
LIVE = runpy.run_path(str(BIN / "droiddeck-steam-compat"))
@@ -214,5 +215,92 @@ class ReconcileTest(unittest.TestCase):
self.assertFalse(COMPAT["live_helper_running"]())
SWITCH_HARNESS = r"""
const sent = [], asked = [];
class FakeSocket { send(data) { sent.push(data); } }
globalThis.window = globalThis;
globalThis.WebSocket = FakeSocket;
window.droiddeckSwitchToDesktop = (x) => asked.push(x);
const install = require("fs").readFileSync(process.argv[2], "utf8");
const frame = (method) => {
const name = new TextEncoder().encode(method), h = [0x51, 1, 0, 0, 0, 0, 0, 0, 0, 0x62, name.length, ...name];
const out = new Uint8Array(8 + h.length), dv = new DataView(out.buffer);
dv.setUint32(0, (0x80000000 | 146) >>> 0, true); dv.setUint32(4, h.length, true); out.set(h, 8);
return out;
};
if (eval(install) !== "installed" || eval(install) !== "present") throw new Error("install is not idempotent");
const ws = new FakeSocket();
ws.send(frame("CompatManager.GetCompatTools#1"));
ws.send(frame("SteamOSManager.SwitchToDesktop#1").buffer);
ws.send(frame("SteamOSManager.SwitchToDesktop#1"));
ws.send("SteamOSManager.SwitchToDesktop#1");
console.log(JSON.stringify({ sent: sent.length, asked: asked.length }));
"""
@unittest.skipUnless(shutil.which("node"), "node is not installed")
class SwitchScriptTest(unittest.TestCase):
def test_the_request_is_reported_and_still_sent(self):
with tempfile.TemporaryDirectory() as tmp:
script = Path(tmp) / "switch.js"
script.write_text(LIVE["SWITCH_JS"])
harness = Path(tmp) / "harness.js"
harness.write_text(SWITCH_HARNESS)
result = subprocess.run(["node", str(harness), str(script)], capture_output=True, text=True, timeout=60)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(json.loads(result.stdout), {"sent": 4, "asked": 2})
class AttachPage:
def __init__(self, port, path):
self.calls, self.scripts = [], []
self.ws = mock.Mock()
def evaluate(self, expression, timeout=30):
self.scripts.append(expression)
return True
def call(self, method, params=None, timeout=30):
self.calls.append((method, (params or {}).get("name")))
return {}
def compat(self, call):
return {"selected": "", "fallback": "", "tools": []}
class SwitchToDesktopTest(unittest.TestCase):
def setUp(self):
self.globals = LIVE["Helper"].attach.__globals__
patcher = mock.patch.dict(self.globals, {"shared_context": lambda port: "/devtools/page/1", "Page": AttachPage})
patcher.start()
self.addCleanup(patcher.stop)
def test_only_a_client_without_its_steamos_manager_is_watched(self):
helper = LIVE["Helper"]("/steam", 1, steamos=False)
helper.attach()
self.assertIn(("Runtime.addBinding", LIVE["SWITCH_BINDING"]), helper.page.calls)
self.assertIn(LIVE["SWITCH_JS"], helper.page.scripts)
helper = LIVE["Helper"]("/steam", 1, steamos=True)
helper.attach()
self.assertNotIn(("Runtime.addBinding", LIVE["SWITCH_BINDING"]), helper.page.calls)
self.assertNotIn(LIVE["SWITCH_JS"], helper.page.scripts)
def test_the_request_goes_to_steamos_session_select(self):
helper = LIVE["Helper"]("/steam", 1)
helper.page = mock.Mock()
helper.page.drain.return_value = [
{"method": "Runtime.bindingCalled", "params": {"name": LIVE["BINDING"], "payload": "1"}},
{"method": "Runtime.bindingCalled", "params": {"name": LIVE["SWITCH_BINDING"], "payload": ""}},
]
with mock.patch.object(self.globals["subprocess"], "call", return_value=0) as call:
events = helper.drain()
call.assert_called_once_with(["/usr/bin/steamos-session-select", "plasma"])
self.assertTrue(any(LIVE["called"](e, LIVE["BINDING"]) for e in events))
helper.page.drain.return_value = [{"method": "Runtime.bindingCalled", "params": {"name": LIVE["BINDING"]}}]
with mock.patch.object(self.globals["subprocess"], "call") as call:
helper.drain()
call.assert_not_called()
if __name__ == "__main__":
unittest.main()
+13
View File
@@ -147,6 +147,19 @@ class DesktopGamesTest(unittest.TestCase):
patch.object(desktop, "LIBRARIES", ((str(steam), "Steam Games"), (str(root / "card"), "SD Games"))):
self.assertEqual({40: "Internal", 50: "Outside"}, desktop.write_entries())
def test_runtimes_and_compatibility_tools_get_none(self):
with tempfile.TemporaryDirectory() as temp:
root = Path(temp)
steam = root / "Steam"
manifest(steam / "steamapps", 40, name="Game")
manifest(steam / "steamapps", 1628350, name="Steam Linux Runtime 3.0 (sniper)", installdir="SteamLinuxRuntime_sniper")
manifest(steam / "steamapps", 5000, name="Some Proton", installdir="Some Proton")
(steam / "steamapps/common/Some Proton").mkdir(parents=True)
(steam / "steamapps/common/Some Proton/toolmanifest.vdf").write_text('"manifest"\n{\n}\n')
with patch.object(desktop, "STEAM_ROOT", str(steam)), patch.object(desktop, "APPS_DIR", str(root / "apps")), \
patch.object(desktop, "LIBRARIES", ((str(steam), "Steam Games"),)):
self.assertEqual({40: "Game"}, desktop.write_entries())
if __name__ == "__main__":
unittest.main()