Fuse filesystem that exposes runners for all installed flatpak applications

flatpak-fuse.py

#!/usr/bin/env python3

import os, errno, stat, sys
import copy
import fuse

from subprocess import run, PIPE
from time import time

fuse.fuse_python_api = (0, 2)

FLATPAK = run(['/bin/sh', '-c', 'command -p -v flatpak'], stdout=PIPE).stdout.decode().strip()
FPENV = copy.deepcopy(os.environ)
FPENV['PATH'] = '/bin:/sbin:/usr/bin:/usr/sbin:/usr/local/bin:/usr/local/sbin'

README = f'''This is a virtual filesystem created using python and fuse, to expose
dumb "runners" for flatpak applications.

The script hosting this filesystem is at SELF
'''.encode()
def file_readme(): return README

with open(sys.argv[0]) as f: SELF = f.read().encode()
def file_self(): return SELF

MORE_FILES = ['README', 'SELF', 'STATS']
NAMES = []
HUMAN_NAMES = {}
PACKAGE_NAMES = {}
PACKAGES = []
LAST_UPDATE = 0

# our extra entries need to be cached.
# this is not safe, as calls from different threads might be interleaved,
# but it's good enough for me.
CACHE = {}
def cache(fn):
    CACHE[fn] = fn()
    return CACHE[fn]

def with_cache(fn):
    x = CACHE.get(fn)
    if x: return x
    return cache(fn)

STATS = {}
def hit(what, c=1): STATS[what] = STATS.get(what, 0) + c
def file_stats():
    with open(f'/proc/{os.getpid()}/stat') as f:
        rt = '\n'.join([(lambda x, y: f'{x}: {y}')(['runtime.user', 'runtime.sys'][i], int(x) / 100) for (i, x) in enumerate(f.read().strip().split()[13:15])])
    return '\n'.join(['\n'.join(f'{x}: {STATS[x]}' for x in STATS), rt, '']).encode()

NAME_MAP = {
    'com.jgraph.drawio.desktop': 'drawio',
    'com.spotify.Client': 'spotify',
    'org.localsend.localsend_app': 'localsend',
    'io.gitlab.librewolf-community': 'librewolf',
}

def update_packages():
    def fp_list():
        r = run([FLATPAK, 'list', '--app', '--columns=application'],
                env=FPENV,
                stdout=PIPE,
                stderr=PIPE)
        assert r.returncode == 0 and r.stderr == b''
        return r.stdout.decode().splitlines()

    global LAST_UPDATE
    global NAMES
    global HUMAN_NAMES
    global PACKAGE_NAMES
    global PACKAGES

    # update only every now and then ...
    if time() - LAST_UPDATE < 10:
        return

    LAST_UPDATE = time()
    NAMES = []
    HUMAN_NAMES = {}
    PACKAGE_NAMES = {}
    PACKAGES = fp_list()

    for i in PACKAGES:
        n = i.split('.')
        n = NAME_MAP.get(i, n[-1]).lower()
        HUMAN_NAMES[i] = n
        PACKAGE_NAMES[n] = i
        NAMES.append(n)

update_packages()

def runner(p):
    return f'''#!/bin/sh
exec {FLATPAK} run {p} "$@"
'''.encode()

class FlatpakFS(fuse.Fuse):
    def getattr(self, path):
        hit('getattr.count')
        if path == '/':
            hit('getattr.successful')
            return fuse.Stat(st_mode  = stat.S_IFDIR | 0o555,
                             st_nlink = len(NAMES) + 1)
        else:
            if path[1:] in NAMES:
                hit('getattr.successful')
                return fuse.Stat(st_mode  = stat.S_IFREG | 0o555,
                                 st_nlink = 1,
                                 st_size  = len(runner(PACKAGE_NAMES[path[1:]])))
            elif path[1:] in MORE_FILES:
                hit('getattr.successful')
                return fuse.Stat(st_mode  = stat.S_IFREG | 0o444,
                                 st_nlink = 1,
                                 st_size  = len(cache(globals()['file_' + path[1:].lower()])))
        return -errno.ENOENT

    def readdir(self, path, offset):
        hit('readdir.count')
        if path == '/':
            hit('readdir.successful')
            update_packages()
            for i in NAMES + MORE_FILES + ['.', '..']:
                hit('readdir.entries')
                yield fuse.Direntry(i)

    def open(self, path, flags):
        hit('open.count')
        if path[1:] not in ['.', '..', *NAMES, *MORE_FILES]:
            return -errno.ENOENT
        if (flags & (os.O_RDWR | os.O_WRONLY | os.O_RDONLY)) != os.O_RDONLY:
            return -errno.EACCES
        hit('open.successful')

    def read(self, path, size, offset):
        hit('read.count')
        if path[1:] not in NAMES + MORE_FILES:
            return -errno.ENOENT

        hit('read.successful')
        r = ''
        if path[1:] in NAMES:
            r = runner(PACKAGE_NAMES[path[1:]])
        elif path[1:] in MORE_FILES:
            r = with_cache(globals()['file_' + path[1:].lower()])
        l = len(r)

        if offset < l:
            if offset + size > l:
                size = l - offset
            hit('read.bytes', size)
            return r[offset:offset+size]

        return b''

server = FlatpakFS(version='%prog ' + fuse.__version__, usage=fuse.Fuse.fusage, dash_s_do='setsingle')
server.parse(errex=1)
server.main()