#!/usr/bin/env bash # Official Linux installer. Safe to source: definitions only; main runs in a subshell. # No network/install occurs with --dry-run. No downloaded code is used as a verifier. aixtab_usage() { printf '%s\n' 'Usage: bash scripts/install-linux.sh [--SAND_BOX_AIXTAB_DIR[=DIR]] [--dry-run]' \ 'Linux x86_64 only. Without sandbox markers, installs the official stable DEB.' \ 'Sandbox: install/update latest signed stable AppImage under DIR/aixtab (DIR defaults to /workspace).' \ '--SAND_BOX_AIXTAB_DIR DIR is also accepted; absent/empty/non-absolute values default to /workspace.' \ 'Requires python3, curl and minisign; normal also needs apt-get/dpkg-deb, sandbox needs flock. Never auto-installs tools.' } aixtab_parse_args() { AIXTAB_CLI_DIR_SET=0 AIXTAB_CLI_DIR='' AIXTAB_DRY_RUN=0 AIXTAB_HELP=0 while (($#)); do case "$1" in --SAND_BOX_AIXTAB_DIR) AIXTAB_CLI_DIR_SET=1 if (($# > 1)) && [[ "$2" != --* ]]; then AIXTAB_CLI_DIR=$2; shift; fi ;; --SAND_BOX_AIXTAB_DIR=*) AIXTAB_CLI_DIR_SET=1; AIXTAB_CLI_DIR=${1#*=} ;; --dry-run) AIXTAB_DRY_RUN=1 ;; -h|--help) AIXTAB_HELP=1 ;; *) printf 'Error: unknown argument: %s\n' "$1" >&2; return 1 ;; esac shift done } # One isolated, trusted interpreter entry point. CWD is never a download directory. # Paths and untrusted JSON are arguments/data, never evaluated as code. aixtab_python() ( local interpreter caller_cwd caller_cwd=$(pwd -P) || exit 1 interpreter=$(type -P python3) || { printf '%s\n' 'Error: python3 executable is required.' >&2; exit 1; } case "$interpreter" in /*) ;; *) interpreter="$caller_cwd/$interpreter" ;; esac cd / || exit 1 "$interpreter" -I - "$@" <<'PY' import base64 import configparser import contextlib import datetime import hashlib import json import os import pathlib import re import shlex import shutil import signal import stat import subprocess import sys import tempfile import urllib.parse P = pathlib.Path PROC_ROOT = P('/proc') # Tests inject a fake proc tree in a trusted copy, never via environment. ATOMIC_IDENTITIES = {} # Candidate inode recorded before replace, including signal/error windows. FIXED = ('app', 'bin', 'xdg', 'xdg/config', 'xdg/data', 'xdg/cache', 'state', 'state/aixautobot', 'state/keep_alive', 'state/sdk-update', 'tmp', 'integration') LEAVES = ('xdg/data/aix-browser', 'xdg/data/aix-browser/profiles', 'xdg/config/com.aixnew.studio', 'xdg/data/com.aixnew.studio', 'xdg/cache/com.aixnew.studio') ALLOWED = set(LEAVES) VERSION = re.compile(r'^(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)(?:-([0-9A-Za-z.-]+))?(?:\+[0-9A-Za-z.-]+)?$') def die(message): raise ValueError(message) def absolute(value): return value.startswith('/') and not any(ord(c) < 32 or ord(c) == 127 for c in value) def normal(value): # Resolve each existing component BEFORE processing '..'. Lexical normpath # changes OS semantics for e.g. work/link/../chosen when link is a symlink. cursor = P('/') for part in value.split('/'): if part in ('', '.'): continue if part == '..': if not cursor.is_dir(): die('cannot traverse .. after a missing/non-directory path component: ' + str(cursor)) cursor = cursor.parent continue if exists(cursor) and not cursor.is_dir(): die('non-directory path component: ' + str(cursor)) candidate = cursor / part if candidate.is_symlink(): if not candidate.exists(): die('broken symlink path component: ' + str(candidate)) cursor = candidate.resolve(strict=True) else: cursor = candidate return str(cursor) def within(path, root): return path == root or root in path.parents def exists(path): return os.path.lexists(path) def owned(path): if path.lstat().st_uid != os.geteuid(): die('not owned by the installing user: ' + str(path)) def directory(path, create=False, private=True): if exists(path): if path.is_symlink() or not path.is_dir(): die('expected a real directory (no symlink/file): ' + str(path)) if private: owned(path) if not os.access(path, os.R_OK | os.W_OK | os.X_OK): die('directory is not writable/searchable: ' + str(path)) # os.access alone would accept a root-owned read-only fixture as root. if not path.stat().st_mode & 0o222: die('directory has no write permission: ' + str(path)) if create and private and stat.S_IMODE(path.stat().st_mode) != 0o700: path.chmod(0o700) elif create: path.mkdir(mode=0o700) elif not os.access(path.parent, os.W_OK | os.X_OK) and path.parent.exists(): die('parent is not writable/searchable: ' + str(path.parent)) def regular(path, optional=True): if not exists(path): if optional: return die('required file missing: ' + str(path)) if path.is_symlink() or not path.is_file(): die('expected a real file (no symlink): ' + str(path)) owned(path) def base_check(base, create=False): # Never chmod/chown B or its ancestors. Existing ancestor symlinks are fine; # the selected B itself and all managed directories must be real directories. missing = [] cursor = base while not exists(cursor): missing.append(cursor) cursor = cursor.parent if not cursor.is_dir(): die('base ancestor is a file/broken symlink: ' + str(cursor)) if not os.access(cursor, os.W_OK | os.X_OK) or not cursor.stat().st_mode & 0o222: die('base ancestor is not writable: ' + str(cursor)) if exists(base): directory(base, private=False) if create: for item in reversed(missing): item.mkdir(mode=0o700) def session_roots(): home = os.environ.get('HOME', '') if not absolute(home): die('HOME must be an absolute path for desktop session registration') result = {} for kind, suffix in (('config', '.config'), ('data', '.local/share'), ('cache', '.cache')): value = os.environ.get('XDG_' + kind.upper() + '_HOME', '') result[kind] = P(value) if absolute(value) else P(home) / suffix return result def desktop_value(value): return value.replace('\\', '\\\\').replace(' ', '\\s').replace('\n', '\\n').replace('\r', '\\r').replace('\t', '\\t') def desktop_exec(value): # Desktop Entry has string-unescaping followed by Exec-argument unescaping. value = value.replace('%', '%%') value = ''.join('\\' + c if c in '\\"`$' else c for c in value) return '"' + desktop_value(value) + '"' def launcher(base): app = base / 'aixtab' # Type-2 extract-and-run executes AppRun immediately, before we can repair # the 0.1.50 payload. Extract the verified image explicitly into a private, # content-addressed cache instead. Never patch the signed image itself. program = r"""import hashlib, os, pathlib, stat, subprocess, sys, tempfile # Guardians/relaunches may inherit this switch; it must not override extraction. os.environ.pop('APPIMAGE_EXTRACT_AND_RUN', None) image = pathlib.Path(sys.argv[1]) base = sys.argv[2] with image.open('rb') as stream: digest = hashlib.file_digest(stream, 'sha256').hexdigest() if hasattr(hashlib, 'file_digest') else None if digest is None: h = hashlib.sha256() with image.open('rb') as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b''): h.update(chunk) digest = h.hexdigest() cache = pathlib.Path(os.environ['TMPDIR']) / ('appimage_extracted_' + digest) # Serialize first extraction/repair across launchers without holding the native # application's .install.lock (the application acquires that lock itself). import fcntl with open(str(cache) + '.lock', 'a') as lock: fcntl.flock(lock, fcntl.LOCK_EX) if not cache.exists(): stage = pathlib.Path(tempfile.mkdtemp(prefix='appimage-stage-', dir=os.environ['TMPDIR'])) try: subprocess.run([str(image), '--appimage-extract'], cwd=stage, check=True, stdout=subprocess.DEVNULL) (stage / 'squashfs-root').rename(cache) finally: # Retain failed/ambiguous extraction contents for diagnosis. if not any(stage.iterdir()): stage.rmdir() if cache.is_symlink() or cache.stat().st_uid != os.geteuid(): raise RuntimeError('unsafe AppImage extraction cache') binary = cache / 'usr/bin/aix-browser' resolved = binary.resolve(strict=True) if cache.resolve() not in resolved.parents or not stat.S_ISREG(resolved.stat().st_mode): raise RuntimeError('unsafe AppImage main binary') resolved.chmod(resolved.stat().st_mode | stat.S_IXUSR) os.environ.update(APPIMAGE=str(image), APPDIR=str(cache), OWD=os.getcwd()) os.execv(str(cache / 'AppRun'), [str(cache / 'AppRun'), '--SAND_BOX_AIXTAB_DIR', base, *sys.argv[3:]]) """ return ('#!/bin/sh\nset -eu\n' 'export SAND_BOX_AIXTAB_DIR=' + shlex.quote(str(base)) + '\n' 'export TMPDIR=' + shlex.quote(str(app / 'tmp')) + '\n' 'exec python3 -I -c ' + shlex.quote(program) + ' ' + shlex.quote(str(app / 'app/AixTab.AppImage')) + ' ' + shlex.quote(str(base)) + ' "$@"\n') def desktop(base, autostart=False, flags=None): app = base / 'aixtab' lines = ['[Desktop Entry]', 'Type=Application', 'Name=AixTab', 'Exec=' + desktop_exec(str(app / 'bin/aixtab')) + ' %U', 'Terminal=false', 'Categories=Network;Utility;', 'MimeType=x-scheme-handler/http;x-scheme-handler/https;', 'X-AixTab-Sandbox-Managed=true', 'X-AixTab-Sandbox-BaseDir=' + desktop_value(str(base))] if autostart: flags = flags or {} lines.append('X-GNOME-Autostart-enabled=' + flags.get('X-GNOME-Autostart-enabled', 'true')) if 'Hidden' in flags: lines.append('Hidden=' + flags['Hidden']) return '\n'.join(lines) + '\n' def desktop_unescape(value): result = [] escapes = {'s': ' ', 'n': '\n', 'r': '\r', 't': '\t', '\\': '\\'} cursor = 0 while cursor < len(value): char = value[cursor] if char == '\\': cursor += 1 if cursor >= len(value) or value[cursor] not in escapes: die('invalid desktop entry string escape') char = escapes[value[cursor]] result.append(char) cursor += 1 return ''.join(result) def managed_entry(path, base): if not exists(path): return False regular(path) parser = configparser.ConfigParser(interpolation=None, strict=True) parser.optionxform = str parser.read_string(path.read_text()) section = parser['Desktop Entry'] actual = desktop_unescape(section.get('X-AixTab-Sandbox-BaseDir', '')) if section.get('X-AixTab-Sandbox-Managed') != 'true' or actual != str(base): die('refusing to overwrite an unknown desktop entry: ' + str(path)) return True def autostart_flags(path): parser = configparser.ConfigParser(interpolation=None, strict=True) parser.optionxform = str parser.read_string(path.read_text()) section = parser['Desktop Entry'] result = {} for key in ('Hidden', 'X-GNOME-Autostart-enabled'): if key in section: value = section[key] if value not in ('true', 'false'): die('invalid boolean in managed autostart source: ' + key) result[key] = value return result def registration(base): roots = session_roots() menu = roots['data'] / 'applications/com.aixnew.studio.desktop' auto = roots['config'] / 'autostart/com.aixnew.studio.desktop' source = base / 'aixtab/integration/autostart.desktop' managed_entry(menu, base) enabled = managed_entry(source, base) if enabled: managed_entry(auto, base) return menu, auto, enabled def semver(value): match = VERSION.fullmatch(value) if not match: die('invalid release version') major, minor, patch, pre = match.groups() if pre is not None: ids = pre.split('.') if any(not x or (x.isdigit() and len(x) > 1 and x.startswith('0')) for x in ids): die('invalid prerelease version') prekey = (0, tuple((0, int(x)) if x.isdigit() else (1, x) for x in ids)) else: prekey = (1, ()) return (int(major), int(minor), int(patch), prekey) def bootstrap(base): app = base / 'aixtab' path = app / 'bootstrap.json' if not exists(path): return None regular(path, False) data = json.loads(path.read_text()) if not isinstance(data, dict): die('bootstrap must be a JSON object') if (type(data.get('schema_version')) is not int or data['schema_version'] != 1 or data.get('mode') != 'sandbox' or data.get('base_dir') != str(base) or data.get('package_type') != 'appimage'): die('existing bootstrap identity does not match this sandbox base') semver(data.get('version', '')) mappings = data.get('legacy_mappings') if not isinstance(mappings, dict) or set(mappings) - ALLOWED: die('unknown legacy mapping in bootstrap') if 'xdg/data/aix-browser' in mappings and 'xdg/data/aix-browser/profiles' in mappings: die('overlapping browser mappings') for key, source in mappings.items(): if not isinstance(source, str) or not absolute(source): die('mapping source is not absolute') src = P(source) if not src.is_dir() or src.is_symlink() or src.resolve() != src: die('mapping source is missing/indirect: ' + str(src)) legacy_source(src) if key == 'xdg/data/aix-browser': legacy_source(src / 'profiles') allowed_root = base / ('aix-browser-data' if 'aix-browser' in key else 'aix-studio-data') if not within(src, allowed_root.resolve()) or not within(src.resolve(), base.resolve()): die('mapping source is outside the confirmed persistent legacy root') target = app / key if not target.is_symlink() or target.resolve() != src.resolve(): die('mapping target conflicts or is missing: ' + str(target)) for name in FIXED: if not (app / name).is_dir() or (app / name).is_symlink(): die('established layout is missing/conflicting: ' + str(app / name)) for name in LEAVES: target = app / name if not target.is_dir(): die('established data leaf is missing: ' + str(target)) if target.is_symlink() and name not in mappings: die('unrecorded data symlink: ' + str(target)) regular(app / 'app/AixTab.AppImage', False) if not os.access(app / 'app/AixTab.AppImage', os.X_OK): die('established AppImage is not executable; refusing to guess/replace it') return data def unmanaged_tree_check(base, staging=()): app = base / 'aixtab' if not exists(app): return directory(app) allowed = set(FIXED) | set(LEAVES) stages = set(staging) for stage in stages: if stage.parent != app / 'tmp': die('transaction staging must be a direct child of A/tmp') directory(stage) if stat.S_IMODE(stage.stat().st_mode) & 0o077: die('transaction staging must be private') def visit(parent): for child in parent.iterdir(): relative = child.relative_to(app).as_posix() if child in stages: continue # Exact paths created by this invocation, never a prefix rule. if child.is_symlink(): die('unmanaged installation contains an unknown link: ' + str(child)) if child.is_dir() and relative in allowed: owned(child) visit(child) elif relative == '.install.lock' and child.is_file(): owned(child) if child.stat().st_size or stat.S_IMODE(child.stat().st_mode) != 0o600: die('unmanaged lock is not an empty private installer lock: ' + str(child)) else: die('unmanaged nonempty install root; refusing to overwrite/chmod: ' + str(child)) visit(app) def legacy_source(path): if not path.is_dir(): die('legacy source is not a directory: ' + str(path)) owned(path) mode = stat.S_IMODE(path.stat().st_mode) if mode & 0o700 != 0o700 or not os.access(path, os.R_OK | os.W_OK | os.X_OK): die('legacy source is not readable/writable/searchable by its owner: ' + str(path)) def nonempty(path): return path.is_dir() and next(path.iterdir(), None) is not None def legacy_plan(base, existing): if existing: return existing['legacy_mappings'] app = base / 'aixtab' roots = session_roots() candidates = [('xdg/data/aix-browser', roots['data'] / 'aix-browser', 'data'), ('xdg/data/aix-browser/profiles', roots['data'] / 'aix-browser/profiles', 'data')] candidates += [('xdg/' + kind + '/com.aixnew.studio', root / 'com.aixnew.studio', kind) for kind, root in roots.items()] mappings = {} for key, origin, kind in candidates: # A root browser symlink already identifies all of its children. if key.endswith('/profiles') and 'xdg/data/aix-browser' in mappings: continue family = base / ('aix-browser-data' if 'aix-browser' in key else 'aix-studio-data') cursor, linked = origin, False while True: linked = linked or cursor.is_symlink() if cursor == roots[kind] or cursor == cursor.parent: break cursor = cursor.parent explicit = os.environ.get('XDG_' + kind.upper() + '_HOME', '') explicit_persistent_root = absolute(explicit) and within(roots[kind].resolve(), family.resolve()) if exists(origin) and (linked or explicit_persistent_root): if not origin.is_dir(): die('broken/non-directory original XDG consumer: ' + str(origin)) src = origin.resolve() if not family.is_dir() or family.is_symlink() or not within(src, family.resolve()): die('original XDG consumer points outside a confirmed B legacy directory: ' + str(origin)) if not within(src, base.resolve()): die('legacy source escaped the persistent base') legacy_source(src) if key == 'xdg/data/aix-browser' and (not (src / 'profiles').is_dir() or (src / 'profiles').is_symlink()): die('browser root source does not contain real profiles; refusing to guess its level') if key == 'xdg/data/aix-browser': legacy_source(src / 'profiles') if key.endswith('/profiles') and (src / 'profiles').exists(): die('ambiguous nested profiles source') mappings[key] = str(src) elif exists(origin) and not origin.is_dir(): die('original XDG consumer is not a directory: ' + str(origin)) elif nonempty(origin): # A real browser parent can contain just a confirmed profiles symlink. if key == 'xdg/data/aix-browser' and set(p.name for p in origin.iterdir()) == {'profiles'} and (origin / 'profiles').is_symlink(): continue die('existing original XDG data has no confirmed persistent symlink: ' + str(origin)) if len(set(mappings.values())) != len(mappings): die('one legacy source is claimed by multiple consumer types; refusing ambiguous identity') for family, token in (('aix-browser-data', 'aix-browser'), ('aix-studio-data', 'com.aixnew.studio')): source = base / family if exists(source): if source.is_symlink() or not source.is_dir(): die('legacy root is not a real persistent directory: ' + str(source)) if not any(token in key for key in mappings): die('legacy directory exists but its XDG identity/source is unknown: ' + str(source)) for name in LEAVES: target = app / name parent_mapping = next((key for key in mappings if name.startswith(key + '/')), None) if name in mappings: if exists(target) and (not target.is_symlink() or target.resolve() != P(mappings[name])): die('legacy mapping conflicts with an independent target: ' + str(target)) elif not parent_mapping and exists(target): empty_browser_wrapper = False if name == 'xdg/data/aix-browser' and target.is_dir() and not target.is_symlink(): profiles = target / 'profiles' if set(p.name for p in target.iterdir()) == {'profiles'}: empty_browser_wrapper = ( profiles.is_dir() and not profiles.is_symlink() and not nonempty(profiles)) mapped = mappings.get('xdg/data/aix-browser/profiles') empty_browser_wrapper = empty_browser_wrapper or ( bool(mapped) and profiles.is_symlink() and profiles.resolve() == P(mapped)) if target.is_symlink() or not target.is_dir() or (nonempty(target) and not empty_browser_wrapper): die('unidentified pre-existing data target: ' + str(target)) return mappings def check_processes(sources): if not sources: return if not PROC_ROOT.is_dir(): die('cannot inspect legacy writers: /proc is unavailable; use a Linux host with readable procfs') # References to these sources matter across all displays/sessions. A process # basename alone does not prove it can write this legacy data. roots = [P(src) for src in sources] needles = [os.fsencode(src) for src in sources] def dead(proc): # comm may contain spaces and ')'; state follows its final ')'. try: fields = (proc / 'stat').read_text().rsplit(')', 1)[1].split() return fields[0] in ('Z', 'X', 'x') except (FileNotFoundError, ProcessLookupError): return not proc.exists() except (OSError, IndexError): return False # Unknown live state must still be inspected. def referenced(proc, message): # A matching argv/handle may have been read just before process exit. if dead(proc): raise ProcessLookupError() die(message) for proc in PROC_ROOT.iterdir(): if not proc.name.isdigit() or int(proc.name) == os.getpid(): continue try: if proc.stat().st_uid != os.geteuid(): continue if dead(proc): continue args = (proc / 'cmdline').read_bytes().split(b'\0') if any(re.search(re.escape(needle) + rb'(?=$|[/\s,:\x22\x27])', arg) for needle in needles for arg in args): referenced(proc, 'a process still references legacy data; close it before installing') handles = list((proc / 'fd').iterdir()) handles += [proc / name for name in ('cwd', 'exe') if exists(proc / name)] for fd in handles: try: raw = os.readlink(fd) except FileNotFoundError: continue # An individual descriptor/process can close during inspection. dest = P(raw[:-10] if raw.endswith(' (deleted)') else raw) if dest.is_absolute() and any(within(dest, root) for root in roots): referenced(proc, 'a process has legacy data/WAL open; close it before installing') maps = proc / 'maps' if maps.exists(): for line in maps.read_bytes().splitlines(): fields = line.split(None, 5) if len(fields) == 6: raw = os.fsdecode(fields[5]) dest = P(raw[:-10] if raw.endswith(' (deleted)') else raw) if dest.is_absolute() and any(within(dest, root) for root in roots): referenced(proc, 'a process has legacy data mapped; close it before installing') except (FileNotFoundError, ProcessLookupError): if not dead(proc): die('cannot completely inspect a live process for legacy data use: ' + str(proc)) except OSError as error: if dead(proc): continue die('cannot inspect a process for legacy data use: ' + str(proc) + ': ' + str(error)) def host_source(base): home = os.environ.get('HOME', '') if not absolute(home): die('HOME must be absolute to inspect legacy host state') source = P(home) / '.aixtab' real = P(normal(str(source))) app = base / 'aixtab' archive = app / 'legacy/home-aixtab' # Check real and physical ancestry before creating B/A/lock. Bind mounts can # alias the source or any child without realpath revealing that relationship. mounts = source_mount_table() if exists(source) else [] if any(physical_within(path, real, mounts) for path in (base, app, archive)): die('persistent destination is physically inside/aliases legacy HOME/.aixtab; choose a non-overlapping base: ' + str(source)) if physical_within(real, archive, mounts) or physical_within(real, app / 'tmp', mounts): die('legacy source overlaps archive/transaction staging: ' + str(source)) persistent = physical_within(real, base, mounts) if source.is_symlink() and not persistent: die('legacy HOME/.aixtab is an unsafe external symlink; review manually: ' + str(source)) return source, real, persistent def source_identity(metadata, relative): return {'path': relative, 'type': 'directory' if stat.S_ISDIR(metadata.st_mode) else 'file', 'dev': metadata.st_dev, 'ino': metadata.st_ino, 'uid': metadata.st_uid, 'gid': metadata.st_gid, 'mode': stat.S_IMODE(metadata.st_mode), 'size': metadata.st_size, 'nlink': metadata.st_nlink, 'mtime_ns': metadata.st_mtime_ns, 'ctime_ns': metadata.st_ctime_ns} def source_mount_table(): path = PROC_ROOT / 'self/mountinfo' try: lines = path.read_text().splitlines() except OSError: die('cannot inspect mount boundaries for legacy snapshot: /proc/self/mountinfo') mounts = [] for line in lines: fields = line.split() if len(fields) < 7 or ' - ' not in line: die('invalid mountinfo; cannot establish safe legacy snapshot boundaries') def decode(value): return P(re.sub(r'\\([0-7]{3})', lambda m: chr(int(m[1], 8)), value)) root, mountpoint = decode(fields[3]), decode(fields[4]) if not root.is_absolute() or not mountpoint.is_absolute(): die('non-absolute mountinfo path') mounts.append((fields[2], root, mountpoint)) if not mounts: die('empty mountinfo; cannot establish safe legacy snapshot boundaries') return mounts def source_mounts(): return {mountpoint for _, _, mountpoint in source_mount_table()} def inode_identity(metadata): return (metadata.st_dev, metadata.st_ino) @contextlib.contextmanager def defer_cancellation(): # Only creation + trusted identity/undo registration belongs in this short # section. Copying, hashing and downloads must stay interruptible. previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGTERM, signal.SIGINT}) try: yield finally: signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask) def source_write_check(base, old, mappings, imports): source, real, _ = host_source(base) if not exists(source): return mounts = source_mount_table() app = base / 'aixtab' menu, autostart, enabled = registration(base) files = [app / name for name in ('bin/aixtab', 'integration/aixtab.desktop', 'bootstrap.json')] files += [app / key for key in imports] files.append(menu) if not old: files.append(app / 'app/AixTab.AppImage') if enabled: files += [app / 'integration/autostart.desktop', autostart] for path in files: if physical_within(path, real, mounts): die('installation/desktop registration would modify legacy source: ' + str(path)) # Correct-mode existing managed directories must be true no-ops. If making # one private/creating a leaf would touch an already-persistent source, stop. directories = [app] + [app / name for name in FIXED] directories += [app / name for name in LEAVES if not any(name == key or name.startswith(key + '/') for key in mappings)] for path in directories: if physical_within(path, real, mounts) and (not exists(path) or stat.S_IMODE(path.stat().st_mode) != 0o700): die('creating/chmodding a managed directory would modify legacy source: ' + str(path)) def physical_within(path, root, mounts): """realpath cannot expose bind aliases. Compare mount roots AND inode ancestry.""" path, root = P(normal(str(path))), P(normal(str(root))) if within(path, root): return True def location(value): covering = [item for item in mounts if within(value, item[2])] if not covering: die('mountinfo does not cover path: ' + str(value)) device, origin, mountpoint = max(covering, key=lambda item: len(item[2].parts)) return device, origin / value.relative_to(mountpoint) if mounts: device, origin = location(path) root_device, root_origin = location(root) if device == root_device and within(origin, root_origin): return True if root.exists(): identity = inode_identity(root.stat()) cursor = path while True: if cursor.exists() and inode_identity(cursor.stat()) == identity: return True if cursor == cursor.parent: break cursor = cursor.parent return False def scan_host_tree(root, copy_to=None, archive=False): """FD-anchored enumeration/copy: never follow child links or open special files. Record identities exclude atime (our reads can update it), but include ctime, mtime, inode, mode, owner, size and link count. Every file is streamed/hashed; directory listings and fstat identities are checked again after each read. """ mounts = source_mounts() entries = [] flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW descriptor = os.open(root, flags) device = os.fstat(descriptor).st_dev def visit(parent_fd, name, relative, destination=None): metadata = os.fstat(parent_fd) if name is None else os.stat(name, dir_fd=parent_fd, follow_symlinks=False) path = root if relative == '.' else root / relative mode = stat.S_IMODE(metadata.st_mode) isdir = stat.S_ISDIR(metadata.st_mode) if not isdir and not stat.S_ISREG(metadata.st_mode): die('unsafe symlink/special file in legacy tree: ' + str(path)) if metadata.st_uid != os.geteuid() or mode & 0o022: die('unsafe ownership/write permissions in legacy tree: ' + str(path)) if not mode & 0o400 or (isdir and not mode & 0o100): die('unreadable/unsearchable legacy tree entry: ' + str(path)) if metadata.st_dev != device or (relative != '.' and (path in mounts or os.path.ismount(path))): die('legacy tree crosses a mount boundary: ' + str(path)) if archive and mode != (0o700 if isdir else 0o600): die('snapshot permissions are not private: ' + str(path)) before = source_identity(metadata, relative) if name is None: fd = os.dup(parent_fd) else: fd = os.open(name, flags if isdir else os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent_fd) try: if source_identity(os.fstat(fd), relative) != before: die('legacy entry was replaced before reading: ' + str(path)) if isdir: entries.append(before) children = sorted(os.listdir(fd)) if destination is not None: destination.mkdir(mode=0o700) for child in children: child_rel = child if relative == '.' else relative + '/' + child visit(fd, child, child_rel, destination / child if destination is not None else None) if sorted(os.listdir(fd)) != children: die('legacy directory changed during snapshot: ' + str(path)) if destination is not None: sync_directory(destination) else: digest = hashlib.sha256() output = None try: if destination is not None: out_fd = os.open(destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) output = os.fdopen(out_fd, 'wb') with os.fdopen(os.dup(fd), 'rb') as stream: while True: chunk = stream.read(1024 * 1024) if not chunk: break digest.update(chunk) if output is not None: output.write(chunk) if output is not None: output.flush() os.fsync(output.fileno()) finally: if output is not None: output.close() before['sha256'] = digest.hexdigest() entries.append(before) after = source_identity(os.fstat(fd), relative) if after != {key: value for key, value in before.items() if key != 'sha256'}: die('legacy entry changed during snapshot: ' + str(path)) current = root.lstat() if name is None else os.stat(name, dir_fd=parent_fd, follow_symlinks=False) if source_identity(current, relative) != after: die('legacy entry was replaced during snapshot: ' + str(path)) finally: os.close(fd) try: visit(descriptor, None, '.', copy_to) finally: os.close(descriptor) return entries def sync_directory(path): fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: os.fsync(fd) finally: os.close(fd) def safe_remove_tree(path, identity): """Remove only this invocation's same-filesystem tree, never through mounts. Preflight the whole tree before deleting anything. Then use directory FDs and the captured identities again while unlinking; changed/mounted/unknown entries are retained for manual inspection, not traversed with rmtree/rm -rf. """ if not exists(path): return if identity is None: die('cleanup has no captured identity; retained for manual inspection: ' + str(path)) if path.is_symlink(): die('cleanup root became a symlink; retained for manual inspection: ' + str(path)) path = P(normal(str(path.parent))) / path.name inventory, children = {}, {} flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW def checked(metadata, relative): if metadata.st_uid != os.geteuid() or metadata.st_dev != identity[0]: die('cleanup ownership/filesystem changed: ' + str(path / relative)) kind = stat.S_IFMT(metadata.st_mode) if kind not in (stat.S_IFDIR, stat.S_IFREG, stat.S_IFLNK): die('cleanup found an unknown special entry: ' + str(path / relative)) if kind == stat.S_IFREG and metadata.st_nlink != 1: die('cleanup refuses an unknown/shared hard link: ' + str(path / relative)) if kind == stat.S_IFDIR and stat.S_IMODE(metadata.st_mode) != 0o700: die('cleanup directory is no longer private: ' + str(path / relative)) return inode_identity(metadata) + (kind,) def boundary(): if any(within(mount, path) for mount in source_mounts()) or os.path.ismount(path): die('cleanup refuses a mounted tree; retained, unmount/review manually (never rm -rf): ' + str(path)) boundary() if path.is_symlink() or inode_identity(path.lstat()) != identity: die('cleanup root identity changed; retained for manual inspection: ' + str(path)) root_fd = os.open(path, flags) def collect(fd, relative): metadata = os.fstat(fd) key = checked(metadata, relative) if relative == '.' and key[:2] != identity: die('cleanup root was replaced: ' + str(path)) inventory[relative] = key children[relative] = set(os.listdir(fd)) for name in sorted(children[relative]): child_rel = name if relative == '.' else relative + '/' + name info = os.stat(name, dir_fd=fd, follow_symlinks=False) inventory[child_rel] = checked(info, child_rel) if stat.S_ISDIR(info.st_mode): child_fd = os.open(name, flags, dir_fd=fd) try: if checked(os.fstat(child_fd), child_rel) != inventory[child_rel]: die('cleanup directory was replaced: ' + child_rel) collect(child_fd, child_rel) finally: os.close(child_fd) def remove(fd, relative): boundary() if checked(os.fstat(fd), relative) != inventory[relative]: die('cleanup directory identity changed: ' + relative) expected = children[relative] if set(os.listdir(fd)) != expected: die('cleanup tree changed; retained for manual inspection: ' + str(path / relative)) for name in sorted(expected): child_rel = name if relative == '.' else relative + '/' + name info = os.stat(name, dir_fd=fd, follow_symlinks=False) if checked(info, child_rel) != inventory[child_rel]: die('cleanup entry identity changed: ' + child_rel) if stat.S_ISDIR(info.st_mode): child_fd = os.open(name, flags, dir_fd=fd) try: remove(child_fd, child_rel) finally: os.close(child_fd) if checked(os.stat(name, dir_fd=fd, follow_symlinks=False), child_rel) != inventory[child_rel]: die('cleanup directory changed before removal: ' + child_rel) os.rmdir(name, dir_fd=fd) else: os.unlink(name, dir_fd=fd) try: collect(root_fd, '.') boundary() remove(root_fd, '.') if path.is_symlink() or inode_identity(path.lstat()) != identity: die('cleanup root changed before removal: ' + str(path)) path.rmdir() finally: os.close(root_fd) def retain_interrupted_snapshot(path, identity): # Only the package created by THIS attempt may be moved off the completion # path. Never rename through a mounted/replaced package root. if (exists(path) and not path.is_symlink() and inode_identity(path.lstat()) == identity and path not in source_mounts() and not os.path.ismount(path)): quarantine = path.with_name('.interrupted-' + path.name + '-' + str(identity[0]) + '-' + str(identity[1])) if not exists(quarantine): os.rename(path, quarantine) return quarantine return path def private_snapshot_directory(path): if path.is_symlink() or not path.is_dir(): die('snapshot directory is missing/indirect: ' + str(path)) owned(path) if stat.S_IMODE(path.stat().st_mode) != 0o700: die('snapshot directory is not private 0700: ' + str(path)) def read_snapshot(base, source): package = base / 'aixtab/legacy/home-aixtab' legacy = package.parent if exists(legacy): private_snapshot_directory(legacy) if not exists(package): return None private_snapshot_directory(package) mounts = source_mounts() for path in (legacy, package): if path.stat().st_dev != (base / 'aixtab').stat().st_dev or path in mounts or os.path.ismount(path): die('snapshot package crosses a mount/filesystem boundary: ' + str(path)) record = package / 'snapshot.json' regular(record, False) if stat.S_IMODE(record.stat().st_mode) != 0o600: die('snapshot record is not private 0600: ' + str(record)) data = json.loads(record.read_text()) if (not isinstance(data, dict) or type(data.get('schema_version')) is not int or data['schema_version'] != 1 or data.get('kind') not in ('snapshot', 'already_persistent') or data.get('complete') is not True or data.get('base_dir') != str(base) or data.get('source') != str(source) or not isinstance(data.get('source_realpath'), str) or not absolute(data['source_realpath']) or not isinstance(data.get('created_at'), str)): die('invalid/incomplete snapshot record: ' + str(record)) try: datetime.datetime.fromisoformat(data['created_at'].replace('Z', '+00:00')) except ValueError: die('invalid snapshot creation time: ' + str(record)) entries = data.get('entries') if not isinstance(entries, list) or not entries: die('snapshot record has no root/entry inventory: ' + str(record)) known = {} identity_keys = ('dev', 'ino', 'uid', 'gid', 'mode', 'size', 'nlink', 'mtime_ns', 'ctime_ns') for item in entries: if not isinstance(item, dict): die('invalid snapshot entry') relative = item.get('path') if (not isinstance(relative, str) or not relative or (relative != '.' and (relative.startswith('/') or any(part in ('', '.', '..') for part in relative.split('/')))) or relative in known or item.get('type') not in ('directory', 'file') or any(type(item.get(key)) is not int or item[key] < 0 for key in identity_keys)): die('invalid snapshot path/identity metadata: ' + str(record)) if relative == '.': if known or item['type'] != 'directory': die('snapshot root inventory is invalid') elif known.get(P(relative).parent.as_posix(), {}).get('type') != 'directory': die('snapshot entry has no inventoried parent: ' + relative) if (item['uid'] != os.geteuid() or item['mode'] > 0o7777 or item['mode'] & 0o022 or not item['mode'] & 0o400 or (item['type'] == 'directory' and not item['mode'] & 0o100)): die('snapshot source metadata records unsafe ownership/permissions: ' + relative) if item['type'] == 'file' and not re.fullmatch(r'[0-9a-f]{64}', item.get('sha256', '')): die('snapshot file has no valid content hash: ' + relative) known[relative] = item expected = {'snapshot.json'} if data['kind'] == 'snapshot': expected.add('tree') tree = package / 'tree' private_snapshot_directory(tree) if tree.stat().st_dev != package.stat().st_dev or tree in mounts or os.path.ismount(tree): die('snapshot tree root crosses a mount/filesystem boundary: ' + str(tree)) actual = scan_host_tree(tree, archive=True) if len(actual) != len(entries): die('snapshot tree conflicts with its record inventory') for item in actual: saved = known.get(item['path']) if (saved is None or saved['type'] != item['type'] or (item['type'] == 'file' and (saved['size'] != item['size'] or saved['sha256'] != item['sha256']))): die('snapshot tree content conflicts with its record: ' + item['path']) elif not physical_within(P(data['source_realpath']), base, source_mount_table()): die('already_persistent record points outside its declared base') if set(path.name for path in package.iterdir()) != expected: die('snapshot package has missing/unknown entries: ' + str(package)) return data def snapshot_plan(base): source, real, persistent = host_source(base) previous = read_snapshot(base, source) if not exists(source): if previous and previous['kind'] != 'snapshot': die('previously persistent legacy source is missing; no archived tree exists: ' + str(source)) return {'status': 'retained' if previous else 'absent', 'record': previous, 'source': source, 'real': real, 'entries': None} # HOME itself and each source ancestor in the persistent range must not let # another user replace the source. No chmod/chown is performed on any source. parents = {P(os.environ['HOME']).resolve()} if persistent: cursor = real.parent while within(cursor, base): parents.add(cursor) if cursor == base: break cursor = cursor.parent for parent in parents: if not parent.is_dir() or parent.stat().st_uid != os.geteuid() or stat.S_IMODE(parent.stat().st_mode) & 0o022: die('unsafe legacy source parent ownership/permissions: ' + str(parent)) check_processes([str(real)]) entries = scan_host_tree(real) if previous: if (previous['source_realpath'] != str(real) or (previous['kind'] == 'snapshot' and previous['entries'] != entries)): die('legacy source changed since the completed snapshot; refusing to overwrite history: ' + str(source)) if previous['kind'] == 'already_persistent' and not persistent: die('legacy source is no longer inside the declared persistent base') # already_persistent acknowledges a mutable location, NOT an immutable # archive. Keep its original observation record, inspect current safety, # and compare only this attempt's inventory during writes. return {'status': 'reuse', 'record': previous, 'source': source, 'real': real, 'entries': entries} if len(entries) == 1: return {'status': 'empty', 'record': None, 'source': source, 'real': real, 'entries': entries} kind = 'already_persistent' if persistent else 'snapshot' nearest = base / 'aixtab/tmp' while not nearest.exists(): nearest = nearest.parent required = sum(item['size'] for item in entries if item['type'] == 'file') if kind == 'snapshot' else 0 required += len(entries) * 4096 + 3 * 1048576 # two <=1 MiB token imports, record/directories and filesystem overhead if shutil.disk_usage(nearest).free < required: die('insufficient free space for a complete private legacy snapshot: ' + str(base)) record = {'schema_version': 1, 'kind': kind, 'complete': True, 'base_dir': str(base), 'source': str(source), 'source_realpath': str(real), 'created_at': datetime.datetime.now(datetime.timezone.utc).isoformat(), 'entries': entries} return {'status': 'new', 'record': record, 'source': source, 'real': real, 'entries': entries} def verify_snapshot_source(plan): if plan['entries'] is None: if exists(plan['source']): die('legacy source appeared during installation; retry to preserve it') return if not exists(plan['source']) or P(normal(str(plan['source']))) != plan['real']: die('legacy source disappeared/was redirected during installation: ' + str(plan['source'])) check_processes([str(plan['real'])]) if scan_host_tree(plan['real']) != plan['entries']: die('legacy source changed during installation: ' + str(plan['source'])) def stage_snapshot(base, plan, tracked=None): app = base / 'aixtab' if (app / 'tmp').stat().st_dev != app.stat().st_dev: die('snapshot staging must be on the installation filesystem') stage = stage_identity = None try: with defer_cancellation(): stage = P(tempfile.mkdtemp(prefix='snapshot-', dir=app / 'tmp')) stage_identity = inode_identity(stage.lstat()) if tracked is not None: tracked.append((stage, stage_identity)) # Caller owns it BEFORE copy/return. if plan['record']['kind'] == 'snapshot': copied = scan_host_tree(plan['real'], stage / 'tree') if copied != plan['entries']: die('legacy source changed before/during snapshot copy') verify_snapshot_source(plan) atomic(stage / 'snapshot.json', (json.dumps(plan['record'], ensure_ascii=True, indent=2) + '\n').encode(), 0o600) sync_directory(stage) return stage except BaseException: if stage is not None: try: safe_remove_tree(stage, stage_identity) except (ValueError, OSError) as cleanup_error: print('Error: incomplete staging retained; review mounts/identity manually: ' + str(cleanup_error), file=sys.stderr) raise def host_state_plan(base, old): if old: return {} # Never re-import HOME caches into an established account/state. _, root, _ = host_source(base) result = {} def safe_parent(path): if path.is_symlink() or not path.is_dir(): die('legacy host state has an indirect/non-directory source; review manually: ' + str(path)) owned(path) if stat.S_IMODE(path.stat().st_mode) & 0o022 or not os.access(path, os.R_OK | os.X_OK): die('legacy host state parent is unsafe/inaccessible; close AixTab and secure its ownership/permissions: ' + str(path)) if not exists(root): return result safe_parent(root) for relative in ('aixautobot/node-token', 'keep_alive/token.json'): source = root / relative if exists(source.parent): safe_parent(source.parent) if not exists(source): continue regular(source, False) metadata = source.stat() if (stat.S_IMODE(metadata.st_mode) & 0o022 or not metadata.st_mode & 0o400 or not os.access(source, os.R_OK) or metadata.st_size > 1048576): die('legacy credential needs manual review; close AixTab and use an owner-readable regular file with no group/other write permission, then rerun: ' + str(source)) if not metadata.st_size: continue target = base / 'aixtab/state' / relative if exists(target): die('legacy credential conflicts with existing target; do not merge/delete credentials: ' + str(target)) result['state/' + relative] = source return result def read_host_state(source): descriptor = os.open(source, os.O_RDONLY | os.O_NOFOLLOW) with os.fdopen(descriptor, 'rb') as stream: metadata = os.fstat(stream.fileno()) if (not stat.S_ISREG(metadata.st_mode) or metadata.st_uid != os.geteuid() or stat.S_IMODE(metadata.st_mode) & 0o022 or not metadata.st_mode & 0o400): die('legacy credential changed or is unsafe: ' + str(source)) content = stream.read(1048577) if (not content or len(content) > 1048576 or source_identity(os.fstat(stream.fileno()), '.') != source_identity(metadata, '.')): die('legacy credential changed/has unexpected size: ' + str(source)) return content def layout(base, create=False, staging=()): app = base / 'aixtab' host_source(base) # Topology must be resolved before creating even B/A/lock. base_check(base, False) directory(app, False) for name in FIXED: # Checking the parent chain separately prevents following managed symlinks. path = app / name if exists(path): directory(path) for name in ('bootstrap.json', '.install.lock', 'app/AixTab.AppImage', 'bin/aixtab', 'integration/aixtab.desktop', 'integration/autostart.desktop'): regular(app / name) old = bootstrap(base) if not old: unmanaged_tree_check(base, staging) mappings = legacy_plan(base, old) snapshot_plan(base) # Also runs on established/offline repair; never just old:return. imports = host_state_plan(base, old) managed_entry(app / 'integration/aixtab.desktop', base) registration(base) # Unknown persistent/session entries fail before any writes. source_write_check(base, old, mappings, imports) if not old: check_processes(list(mappings.values()) + [str(path) for path in imports.values()]) if create: base_check(base, True) directory(app, True) for name in FIXED: directory(app / name, True) for name in LEAVES: path = app / name if name in mappings: continue # Mapping links are created only in the signed commit transaction. elif not any(name.startswith(key + '/') for key in mappings): directory(path, True) if hasattr(os, 'ST_NOEXEC') and os.statvfs(app).f_flag & os.ST_NOEXEC: die('sandbox base is mounted noexec; choose an executable persistent base') return old, mappings def atomic(path, content, mode): fd = name = temporary_identity = stream = None try: with defer_cancellation(): fd, name = tempfile.mkstemp(prefix='.aixtab-', dir=path.parent) temporary_identity = inode_identity(os.fstat(fd)) stream = os.fdopen(fd, 'wb') fd = None # The recorded stream now owns the descriptor. with stream: stream.write(content) stream.flush() os.fsync(stream.fileno()) os.chmod(name, mode) ATOMIC_IDENTITIES[str(path)] = temporary_identity os.replace(name, path) finally: if stream is not None and not stream.closed: stream.close() if fd is not None: os.close(fd) if name is not None and os.path.lexists(name): metadata = os.lstat(name) if (not stat.S_ISREG(metadata.st_mode) or inode_identity(metadata) != temporary_identity or metadata.st_uid != os.geteuid() or metadata.st_nlink != 1): die('atomic temporary identity/shared-link changed; retained for manual inspection: ' + name) os.unlink(name) # Unlinking a mounted file fails EBUSY; never follows it. def register_parent(path, saved): # Desktop session directories are not managed data: never chmod existing # parents. New empty directories participate in the same undo sequence. missing, cursor = [], path.parent while not exists(cursor): missing.append(cursor) cursor = cursor.parent for item in reversed(missing): with defer_cancellation(): item.mkdir(mode=0o700) saved.append(('emptydir', item, None, None, inode_identity(item.lstat()))) if not path.parent.is_dir() or not os.access(path.parent, os.W_OK | os.X_OK): die('desktop registration directory is not writable: ' + str(path.parent)) def commit(base, version, payload, staging=()): # TERM and Ctrl-C enter the same undo path as publication failures. A second # cancellation must not interrupt that undo. SIGKILL/power loss cannot be # caught; unknown nonempty roots remain guarded for manual inspection. requested = [False] def cancelled(signum, frame): if requested[0]: return requested[0] = True raise InterruptedError(signal.Signals(signum).name + ' received; installation interrupted, rolling back') previous_term = signal.getsignal(signal.SIGTERM) previous_int = signal.getsignal(signal.SIGINT) try: with defer_cancellation(): signal.signal(signal.SIGTERM, cancelled) signal.signal(signal.SIGINT, cancelled) return commit_transaction(base, version, payload, staging) finally: with defer_cancellation(): signal.signal(signal.SIGTERM, previous_term) signal.signal(signal.SIGINT, previous_int) def commit_transaction(base, version, payload, staging=()): app = base / 'aixtab' old, mappings = layout(base, True, staging) history = snapshot_plan(base) imports = host_state_plan(base, old) source_write_check(base, old, mappings, imports) if old and not payload: version = old['version'] # Internal entry-only repair retains the installed package. # A nonempty payload has passed the signed stable release preflight. # Existing package replacement shares the same identity-checked undo journal. menu, autostart, enabled = registration(base) manifest = {'schema_version': 1, 'mode': 'sandbox', 'base_dir': str(base), 'package_type': 'appimage', 'version': version, 'legacy_mappings': mappings} writes = [(app / 'bin/aixtab', launcher(base).encode(), 0o700), (app / 'integration/aixtab.desktop', desktop(base).encode(), 0o600), (menu, desktop(base).encode(), 0o600)] if enabled: flags = autostart_flags(app / 'integration/autostart.desktop') content = desktop(base, True, flags).encode() writes += [(app / 'integration/autostart.desktop', content, 0o600), (autostart, content, 0o600)] writes.append((app / 'bootstrap.json', (json.dumps(manifest, ensure_ascii=False, indent=2) + '\n').encode(), 0o600)) for path, _, _ in writes: regular(path) saved = [] backup = backup_identity = None appfile = app / 'app/AixTab.AppImage' history_stage = history_identity = None history_tracking = [] undo_errors = [] completed = False try: with defer_cancellation(): backup = P(tempfile.mkdtemp(prefix='commit-', dir=app / 'tmp')) backup_identity = inode_identity(backup.lstat()) # Registration, backup and snapshot paths were all checked against the # physical source before layout() could write B/A or chmod directories. for path, _, _ in writes: if path in (menu, autostart): register_parent(path, saved) if history['status'] == 'new': history_stage = stage_snapshot(base, history, history_tracking) history_identity = history_tracking[-1][1] if (history_stage != history_tracking[-1][0] or history_stage.is_symlink() or inode_identity(history_stage.lstat()) != history_identity): die('snapshot staging identity changed during ownership handoff') parent = app / 'legacy' if not exists(parent): with defer_cancellation(): parent.mkdir(mode=0o700) saved.append(('emptydir', parent, None, None, inode_identity(parent.lstat()))) private_snapshot_directory(parent) if parent.stat().st_dev != history_stage.stat().st_dev: die('snapshot archive and staging must share a filesystem') target = parent / 'home-aixtab' if exists(target): die('snapshot target appeared during installation; refusing to overwrite: ' + str(target)) # Register the known staging inode BEFORE rename: a signal directly # after the syscall must not leave an untracked canonical package. saved.append(('snapshot', target, None, None, history_identity)) os.rename(history_stage, target) history_stage = None sync_directory(parent) for key, source in mappings.items(): target = app / key if not exists(target): link_identity = [] saved.append(('link', target, source, None, link_identity)) with defer_cancellation(): target.symlink_to(source, target_is_directory=True) link_identity.append(inode_identity(target.lstat())) for key, source in imports.items(): target = app / key if exists(target): die('legacy credential target appeared during installation: ' + str(target)) content = read_host_state(source) relative = source.relative_to(history['real']).as_posix() identity = next((item for item in history['entries'] or [] if item['path'] == relative), None) if identity is None or identity.get('sha256') != hashlib.sha256(content).hexdigest(): die('legacy credential changed after snapshot planning: ' + str(source)) saved.append(('file', target, None, None, (content, 0o600, inode_identity(target.parent.stat())))) ATOMIC_IDENTITIES.pop(str(target), None) atomic(target, content, 0o600) if payload: regular(P(payload), False) old_app = backup / 'previous.AppImage' had_app = exists(appfile) original_identity = inode_identity(appfile.lstat()) if had_app else None if had_app: os.link(appfile, old_app) saved.append(('app', appfile, old_app if had_app else None, original_identity, inode_identity(P(payload).lstat()))) os.chmod(payload, 0o700) os.replace(payload, appfile) for path, content, mode in writes: if path == app / 'bootstrap.json': verify_snapshot_source(history) if history['record'] is not None and read_snapshot(base, history['source']) != history['record']: die('snapshot completion record changed during installation') previous = path.read_bytes() if exists(path) else None previous_mode = stat.S_IMODE(path.stat().st_mode) if previous is not None else None saved.append(('file', path, previous, previous_mode, (content, mode, inode_identity(path.parent.stat())))) ATOMIC_IDENTITIES.pop(str(path), None) atomic(path, content, mode) completed = True except BaseException: # Disable both handled cancellation signals before independent undo; # no second TERM/Ctrl-C may strand later package/token/snapshot steps. with defer_cancellation(): signal.signal(signal.SIGTERM, signal.SIG_IGN) signal.signal(signal.SIGINT, signal.SIG_IGN) for kind, path, previous, mode, expected in reversed(saved): try: present = exists(path) if kind == 'file': if present: regular(path, False) current, current_mode = path.read_bytes(), stat.S_IMODE(path.stat().st_mode) if previous is not None and current == previous and current_mode == mode: continue # Failed atomic left it unchanged (e.g. immutable menu). if ((current, current_mode) != expected[:2] or inode_identity(path.lstat()) != ATOMIC_IDENTITIES.get(str(path)) or inode_identity(path.parent.stat()) != expected[2]): die('undo refuses an unexpectedly changed file/identity: ' + str(path)) if path in source_mounts() or os.path.ismount(path) or path.stat().st_nlink != 1: die('undo refuses a mounted/shared file: ' + str(path)) elif previous is None: continue else: die('undo refuses an unexpectedly missing previous file: ' + str(path)) if previous is not None: atomic(path, previous, mode) else: path.unlink() elif kind == 'app': if present and inode_identity(path.lstat()) == mode: continue # Package replacement did not happen. if present and (path.is_symlink() or inode_identity(path.lstat()) != expected or path in source_mounts()): die('undo refuses a replaced/mounted program: ' + str(path)) if previous is not None: os.replace(previous, path) elif present: path.unlink() elif kind == 'link' and present: if (not expected or not path.is_symlink() or inode_identity(path.lstat()) != expected[0] or os.readlink(path) != previous): die('undo refuses an unknown mapping/identity: ' + str(path)) path.unlink() elif kind == 'snapshot' and present: try: safe_remove_tree(path, expected) except (ValueError, OSError) as cleanup_error: retained = retain_interrupted_snapshot(path, expected) die(str(cleanup_error) + '; interrupted snapshot retained at ' + str(retained)) elif kind == 'emptydir' and present: if path.is_symlink() or inode_identity(path.lstat()) != expected or path in source_mounts(): die('undo refuses a replaced/mounted directory: ' + str(path)) path.rmdir() # Never recursively remove unknown/new contents. except (ValueError, OSError) as undo_error: undo_errors.append(str(undo_error)) for error in undo_errors: print('Error: undo retained an unsafe/unchanged residual; inspect manually, never rm -rf: ' + error, file=sys.stderr) raise finally: cleanups = [(backup, backup_identity)] if backup is not None else [] cleanups += history_tracking # Registered at allocation, not only after stage_snapshot returns. for path, identity in cleanups: try: if undo_errors and path == backup and nonempty(backup): die('rollback backup retained for manual recovery: ' + str(backup)) safe_remove_tree(path, identity) except (ValueError, OSError) as cleanup_error: print('Error: private staging retained; inspect mounts/identity manually: ' + str(cleanup_error), file=sys.stderr) if completed: die('installation was published, but private staging could not be safely cleaned; inspect the reported path') status = history['record']['kind'] if history['status'] == 'new' else history['status'] print('Legacy HOME/.aixtab history: ' + status + ' (active state is not merged).') print('Installed/repaired sandbox version ' + version + ': ' + str(app / 'bin/aixtab')) try: command, *args = sys.argv[1:] if command == 'detect': platform, cli_set, cli_dir, caller_cwd = args mode, reason = 'normal', 'none' if platform == 'Linux': if cli_set == '1': mode, reason = 'sandbox', 'cli' elif any(key.startswith('SAND_BOX_') for key in os.environ): mode, reason = 'sandbox', 'SAND_BOX_* presence' elif any('box-chrome' in (key + '=' + value).lower() for key, value in os.environ.items()): mode, reason = 'sandbox', 'box-chrome environment marker' elif any(os.path.isfile(os.path.join(caller_cwd, part, 'box-chrome')) and os.access(os.path.join(caller_cwd, part, 'box-chrome'), os.X_OK) for part in os.environ.get('PATH', '').split(os.pathsep)): mode, reason = 'sandbox', 'box-chrome PATH executable (not executed)' chosen = cli_dir if cli_set == '1' else os.environ.get('SAND_BOX_AIXTAB_DIR', '') base = normal(chosen) if absolute(chosen) else '/workspace' print('\n'.join((mode, reason, base, base.rstrip('/') + '/aixtab'))) elif command == 'metadata': path, package = args regular(P(path), False) data = json.loads(P(path).read_text()) if not isinstance(data, dict): die('update manifest must be a JSON object') version, url, signature = (data.get(key) for key in ('version', 'url', 'signature')) if not all(isinstance(x, str) and x for x in (version, url, signature)): die('manifest must contain version, url and signature') semver(version) parsed = urllib.parse.urlsplit(url) if (parsed.scheme != 'https' or not parsed.hostname or parsed.username or parsed.password or any(ord(c) < 33 for c in url) or not parsed.path.endswith('.' + package)): die('manifest URL must be an HTTPS ' + package + ' artifact without credentials') if not re.fullmatch(r'[A-Za-z0-9+/]+={0,2}', signature): die('manifest signature must be a single Tauri base64-encoded signature') print('\n'.join((version, url, signature))) elif command == 'artifact': path, package = args with open(path, 'rb') as stream: header = stream.read(64) if package == 'AppImage': if (len(header) < 64 or header[:4] != b'\x7fELF' or header[4:6] != b'\x02\x01' or header[8:11] != b'AI\x02' or header[18:20] != b'\x3e\x00'): die('verified artifact is not a Linux x86_64 type-2 AppImage') elif header[:8] != b'!\n': die('verified stable artifact is not a DEB/ar package') elif command == 'decode': public, signature, out = args key = base64.b64decode(public, validate=True) sig = base64.b64decode(signature, validate=True) key_lines, sig_lines = key.decode('ascii').splitlines(), sig.decode('ascii').splitlines() if (len(key_lines) != 2 or not key_lines[0].startswith('untrusted comment:') or len(base64.b64decode(key_lines[1], validate=True)) != 42): die('invalid embedded Tauri/minisign public key encoding') if (len(sig_lines) != 4 or not sig_lines[0].startswith('untrusted comment:') or not sig_lines[2].startswith('trusted comment:') or len(base64.b64decode(sig_lines[1], validate=True)) != 74 or len(base64.b64decode(sig_lines[3], validate=True)) != 64): die('invalid Tauri/minisign signature encoding') (P(out) / 'release.pub').write_bytes(key + (b'' if key.endswith(b'\n') else b'\n')) (P(out) / 'release.sig').write_bytes(sig + (b'' if sig.endswith(b'\n') else b'\n')) elif command == 'host': if not os.environ.get('DISPLAY') and not os.environ.get('WAYLAND_DISPLAY'): print('Preflight: no graphical session; installation will not launch AixTab.') if os.geteuid() == 0: print('Preflight warning: Chromium normally requires a non-root user and working OS sandbox; no --no-sandbox bypass is supplied.') for filename in ('/proc/sys/kernel/unprivileged_userns_clone', '/proc/sys/user/max_user_namespaces'): path = P(filename) if path.exists() and path.read_text().strip() == '0': print('Preflight warning: user namespaces are disabled by ' + filename + '; arrange an administrator-approved Chromium sandbox before launch.') ldconfig = shutil.which('ldconfig') if ldconfig: try: result = subprocess.run([ldconfig, '-p'], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, timeout=15, cwd='/') missing = [name for name in ('libgtk-3.so.0', 'libwebkit2gtk-4.1.so.0', 'libjavascriptcoregtk-4.1.so.0') if name not in result.stdout] if result.returncode or missing: print('Preflight warning: system GTK/WebKit cache is incomplete; verify AppImage/host library compatibility on Linux before launch.') except (OSError, subprocess.TimeoutExpired): print('Preflight warning: could not inspect the system library cache; validate GTK/WebKit on the host before launch.') else: print('Preflight note: ldconfig is unavailable; validate GTK/WebKit libraries on the Linux host before launch.') elif command == 'identity': print(':'.join(str(value) for value in inode_identity(P(args[0]).lstat()))) elif command == 'cleanup': path, identity = args captured = tuple(int(value) for value in identity.split(':')) if identity else None if captured is not None and len(captured) != 2: die('invalid cleanup identity; retained staging: ' + path) safe_remove_tree(P(path), captured) elif command == 'base': base = P(args[0]); layout(base, False) base_check(base, True); directory(base / 'aixtab', True) regular(base / 'aixtab/.install.lock') elif command == 'layout': base, create = args layout(P(base), create == 'prepare') elif command == 'current': old = bootstrap(P(args[0])) print(old['version'] if old else '') elif command == 'commit': commit(P(args[0]), args[1], args[2], tuple(P(value) for value in args[3:])) elif command == 'launcher': sys.stdout.write(launcher(P(args[0]))) elif command == 'desktop': sys.stdout.write(desktop(P(args[0]), len(args) > 1 and args[1] == 'autostart')) else: die('unknown internal operation') except (ValueError, OSError, KeyError, TypeError, configparser.Error) as error: print('Error: ' + str(error), file=sys.stderr) sys.exit(1) PY ) aixtab_detect() { # PLATFORM is an explicit pure-function input for macOS fixture tests, never an install override. local platform=$1 result result=$(aixtab_python detect "$platform" "$AIXTAB_CLI_DIR_SET" "$AIXTAB_CLI_DIR" "$(pwd -P)") || return 1 { IFS= read -r AIXTAB_MODE; IFS= read -r AIXTAB_REASON; IFS= read -r AIXTAB_BASE; IFS= read -r AIXTAB_INSTALL; } <<< "$result" } aixtab_render_launcher() { aixtab_python launcher "$1"; } aixtab_render_desktop() { aixtab_python desktop "$@"; } aixtab_check_layout() { aixtab_python layout "$1" check; } aixtab_prepare_layout() { aixtab_python layout "$1" prepare; } aixtab_require() { local tool for tool in "$@"; do command -v "$tool" >/dev/null 2>&1 || { printf 'Error: required tool missing: %s. Install it from your trusted OS package source; the installer will not download a verifier.\n' "$tool" >&2 return 1 } done } aixtab_cleanup_downloads() { local failed=0 if [[ -n "$manifest_dir" ]]; then aixtab_python cleanup "$manifest_dir" "$manifest_identity" || failed=1; fi if [[ -n "$payload_dir" ]]; then aixtab_python cleanup "$payload_dir" "$payload_identity" || failed=1; fi if [[ -n "$verify_dir" ]]; then aixtab_python cleanup "$verify_dir" "$verify_identity" || failed=1; fi if [[ "$failed" != 0 ]]; then printf '%s\n' 'Error: some private staging was retained because its identity/mount boundaries could not be proved. Inspect/unmount manually; never recursively delete a mounted residual.' >&2 fi return "$failed" } aixtab_fetch_manifest() { local status status=$(curl --fail --silent --show-error --location --proto '=https' --proto-redir '=https' \ --connect-timeout 20 --max-time 120 --max-filesize 1048576 --output "$1" --write-out '%{http_code}' "$AIXTAB_ENDPOINT") || return 1 case "$status" in 200) ;; 204) printf 'Error: no release available in the %s channel; not falling back to another channel/package.\n' "$AIXTAB_CHANNEL" >&2; return 1 ;; *) printf 'Error: update manifest returned HTTP %s.\n' "$status" >&2; return 1 ;; esac } aixtab_verify_download() { local url=$1 signature=$2 payload=$3 verify_dir=$4 # Synced verbatim from tauri/src-tauri/tauri.conf.json plugins.updater.pubkey. local public='dW50cnVzdGVkIGNvbW1lbnQ6IG1pbmlzaWduIHB1YmxpYyBrZXk6IDkyNjk2NEQ2NkJBREE3ODcKUldTSHA2MXIxbVJwa2wzK0l4Q3E2UlZLdlZPa1hmQVlJYVFqYzk3U2pyM1lVd3JkOTJVWCtKYUcK' aixtab_python decode "$public" "$signature" "$verify_dir" || return 1 curl --fail --silent --show-error --location --proto '=https' --proto-redir '=https' \ --connect-timeout 20 --max-time 1800 --output "$payload" "$url" || return 1 [[ -s "$payload" ]] || { printf '%s\n' 'Error: downloaded artifact is empty.' >&2; return 1; } minisign -V -m "$payload" -x "$verify_dir/release.sig" -p "$verify_dir/release.pub" || { printf '%s\n' 'Error: artifact signature verification failed; keeping the previous installation.' >&2 return 1 } } aixtab_main() ( set -euo pipefail umask 077 aixtab_parse_args "$@" || exit 1 if [[ "$AIXTAB_HELP" == 1 ]]; then aixtab_usage; exit 0; fi aixtab_require python3 local platform arch platform=$(uname -s); arch=$(uname -m) aixtab_detect "$platform" [[ "$platform" == Linux ]] || { printf '%s\n' 'Error: this installer supports Linux only.' >&2; exit 1; } [[ "$arch" == x86_64 || "$arch" == amd64 ]] || { printf '%s\n' 'Error: only Linux x86_64 is currently published by this installer.' >&2; exit 1; } AIXTAB_CHANNEL=stable if [[ "$AIXTAB_MODE" == sandbox ]]; then AIXTAB_PACKAGE=AppImage AIXTAB_FORMAT=appimage else AIXTAB_PACKAGE=deb AIXTAB_FORMAT=deb fi AIXTAB_ENDPOINT="https://aixtab.com/api/v1/updates/linux-x86_64/0.0.0/$AIXTAB_FORMAT?channel=$AIXTAB_CHANNEL" printf 'Mode: %s (%s)\nManifest: %s\n' "$AIXTAB_MODE" "$AIXTAB_REASON" "$AIXTAB_ENDPOINT" if [[ "$AIXTAB_MODE" == sandbox ]]; then printf 'Base B: %s\nInstall A: %s\n' "$AIXTAB_BASE" "$AIXTAB_INSTALL"; fi if [[ "$AIXTAB_DRY_RUN" == 1 ]]; then printf '%s\n' 'DRY RUN: no downloads, directories, locks, chmod, registration, sudo or installation.' if [[ "$AIXTAB_MODE" == sandbox ]]; then printf '%s\n' 'Would install/update the latest signed stable AppImage, preserve sandbox data and repair the persistent launcher/menu. Autostart stays opt-in.' else printf '%s\n' 'Would verify the official stable DEB, then run apt-get install (sudo only if not root).' fi exit 0 fi local manifest_dir='' payload_dir='' verify_dir='' current='' local manifest_identity='' payload_identity='' verify_identity='' trap 'aixtab_cleanup_downloads' EXIT if [[ "$AIXTAB_MODE" == sandbox ]]; then aixtab_require flock # Read-only conflict checks precede creating A, then recheck under the native primary lock. aixtab_check_layout "$AIXTAB_BASE" current=$(aixtab_python current "$AIXTAB_BASE") aixtab_require curl minisign aixtab_python base "$AIXTAB_BASE" exec 9>>"$AIXTAB_INSTALL/.install.lock" chmod 600 "$AIXTAB_INSTALL/.install.lock" flock -n 9 || { printf '%s\n' 'Error: AixTab is running or another installation holds A/.install.lock; close it and retry.' >&2; exit 1; } aixtab_prepare_layout "$AIXTAB_BASE" current=$(aixtab_python current "$AIXTAB_BASE") if [[ -n "$current" ]]; then printf 'Checking latest signed stable AppImage for managed installation (bootstrap version %s).\n' "$current" fi aixtab_python host manifest_dir=$(mktemp -d "$AIXTAB_INSTALL/tmp/install-manifest.XXXXXX") manifest_identity=$(aixtab_python identity "$manifest_dir") payload_dir=$(mktemp -d "$AIXTAB_INSTALL/tmp/install-payload.XXXXXX") payload_identity=$(aixtab_python identity "$payload_dir") verify_dir=$(mktemp -d "$AIXTAB_INSTALL/tmp/install-verifier.XXXXXX") verify_identity=$(aixtab_python identity "$verify_dir") else aixtab_require curl minisign apt-get dpkg-deb if [[ "$(id -u)" != 0 ]]; then aixtab_require sudo; fi manifest_dir=$(mktemp -d "${TMPDIR:-/tmp}/aixtab-manifest.XXXXXX") manifest_identity=$(aixtab_python identity "$manifest_dir") payload_dir=$(mktemp -d "${TMPDIR:-/tmp}/aixtab-payload.XXXXXX") payload_identity=$(aixtab_python identity "$payload_dir") verify_dir=$(mktemp -d "${TMPDIR:-/tmp}/aixtab-verifier.XXXXXX") verify_identity=$(aixtab_python identity "$verify_dir") fi aixtab_fetch_manifest "$manifest_dir/manifest.json" local metadata version url signature payload # Preserve field boundaries without evaluating JSON or allowing shell expansion. metadata=$(aixtab_python metadata "$manifest_dir/manifest.json" "$AIXTAB_PACKAGE") { IFS= read -r version; IFS= read -r url; IFS= read -r signature; } <<< "$metadata" payload="$payload_dir/AixTab.$AIXTAB_PACKAGE" aixtab_verify_download "$url" "$signature" "$payload" "$verify_dir" aixtab_python artifact "$payload" "$AIXTAB_PACKAGE" if [[ "$AIXTAB_MODE" == normal ]]; then local deb_arch deb_arch=$(dpkg-deb --field "$payload" Architecture) [[ "$deb_arch" == amd64 ]] || { printf '%s\n' 'Error: verified DEB is not Linux amd64; refusing package installation.' >&2; exit 1; } fi if [[ "$AIXTAB_MODE" == sandbox ]]; then aixtab_python commit "$AIXTAB_BASE" "$version" "$payload" "$manifest_dir" "$payload_dir" "$verify_dir" printf 'Start with: %q\n' "$AIXTAB_INSTALL/bin/aixtab" printf '%s\n' 'After bot/HOME refresh, an external platform hook must call this persistent launcher.' if [[ -z "${DISPLAY:-}" && -z "${WAYLAND_DISPLAY:-}" ]]; then printf '%s\n' 'Note: no graphical session detected; launch later in a desktop session with the required WebKit/GTK libraries and Chromium sandbox support. No --no-sandbox flag is added.' fi elif [[ "$(id -u)" == 0 ]]; then apt-get install -y "$payload" else sudo apt-get install -y "$payload" fi ) if [[ "${BASH_SOURCE[0]}" == "$0" ]]; then aixtab_main "$@" fi