diff --git a/README.md b/README.md index 8345619..34faf79 100644 --- a/README.md +++ b/README.md @@ -109,6 +109,10 @@ curl --fail --location --proto '=https' --proto-redir '=https' \ --output decksmith-install.sh && sh decksmith-install.sh ``` +On Arch Linux and Arch-based systems like Omarchy, the same command installs +missing packages with pacman, then builds the `main` branch from source +and installs it. There's no signed Arch download, so the build happens locally. + On Fedora 45, Debian 13, Ubuntu 26.04 or Pop!_OS 24.04, opt into the experimental package: ```sh diff --git a/apps/decksmith-daemon/src/audio_target.rs b/apps/decksmith-daemon/src/audio_target.rs index 91dfccb..c48b608 100644 --- a/apps/decksmith-daemon/src/audio_target.rs +++ b/apps/decksmith-daemon/src/audio_target.rs @@ -78,6 +78,13 @@ impl Connection { pub(crate) fn exchange( &mut self, request: &impl serde::Serialize, + ) -> Result, &'static str> { + self.exchange_with_timeout(request, Duration::from_secs(3)) + } + pub(crate) fn exchange_with_timeout( + &mut self, + request: &impl serde::Serialize, + timeout: Duration, ) -> Result, &'static str> { use std::io::Write; let bytes = serde_json::to_vec(request).map_err(|_| "audio_target_failed")?; @@ -90,7 +97,7 @@ impl Connection { .and_then(|()| self.input.flush()) .map_err(|_| "audio_target_failed")?; self.output - .recv_timeout(Duration::from_secs(3)) + .recv_timeout(timeout) .map_err(|error| match error { std::sync::mpsc::RecvTimeoutError::Timeout => "audio_target_timeout", std::sync::mpsc::RecvTimeoutError::Disconnected => "audio_target_failed", diff --git a/apps/decksmith-daemon/src/connected_editor_lab.rs b/apps/decksmith-daemon/src/connected_editor_lab.rs index d885113..d11e104 100644 --- a/apps/decksmith-daemon/src/connected_editor_lab.rs +++ b/apps/decksmith-daemon/src/connected_editor_lab.rs @@ -193,3 +193,143 @@ pub fn run(args: &[String]) -> Result<()> { } Ok(()) } + +#[cfg(test)] +mod live_lock_tests { + use super::*; + use decksmith_core::device_model::ControlAddress; + + /// Exercises the real guarded native helper through a VirtualDeck key. + /// The operator configures the disposable guest's locker before each run. + #[test] + #[ignore = "requires an explicitly prepared disposable Hyprland VM"] + fn virtual_key_runs_real_guarded_lock_action() { + assert_eq!(std::env::var("DECKSMITH_ISOLATED_TEST").as_deref(), Ok("1")); + assert!( + std::env::var("DBUS_SESSION_BUS_ADDRESS") + .is_ok_and(|value| value.starts_with("unix:path=/tmp/dbus-")) + ); + assert!(std::env::var_os("HYPRLAND_INSTANCE_SIGNATURE").is_some()); + let mode = std::env::var("DECKSMITH_LOCK_CASE").unwrap(); + assert!(matches!(mode.as_str(), "slow" | "unconfigured")); + let root = std::env::temp_dir().join(format!("decksmith-live-lock-{}", std::process::id())); + let store = Store::create(&root).unwrap(); + let discovery = Discovery::new(vec![Candidate::new( + 0, + DeviceModel::Mk2, + Some("lock-test-mk2"), + )]) + .unwrap(); + let id = random_device_id().unwrap(); + store + .enroll(0, &discovery, 0, Enrollment::new(id, "Lock test", 50, None)) + .unwrap(); + let catalog = store.load().unwrap(); + let mut value: serde_json::Value = + serde_json::from_slice(catalog.profile(id).unwrap().document().bytes()).unwrap(); + value["pages"][0]["keys"][0] = + serde_json::json!({"label":"Lock", "action":{"type":"system", "command":"lock"}}); + store + .save_profile( + id, + 1, + &serde_json::to_vec(&value).unwrap(), + 0, + "Lock test", + 50, + ) + .unwrap(); + let backend = crate::shared_native_media::NativeBackend::connect_daily().unwrap(); + let mut runtime = + Runtime::start(store.load().unwrap(), vec![Box::new(backend)], None).unwrap(); + let revision = runtime + .discover(Discovery::new(discovery.candidates().to_vec()).unwrap()) + .unwrap(); + let deck = Arc::new(Mutex::new(VirtualDeck::new(DeviceModel::Mk2))); + runtime + .attach(revision, &discovery, 0, Box::new(Probe(deck.clone()))) + .unwrap(); + let ready = Instant::now() + Duration::from_secs(10); + while Instant::now() < ready + && runtime + .statuses() + .iter() + .any(|s| !s.input_ready || !s.display_ready) + { + runtime.tick(); + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + runtime + .statuses() + .iter() + .all(|s| s.input_ready && s.display_ready) + ); + for (time, pressed) in [(1, true), (2, false)] { + deck.lock() + .unwrap() + .inject(InputEvent { + timestamp_ms: time, + event: RawEvent::Key { index: 0, pressed }, + }) + .unwrap(); + } + let started = Instant::now(); + let end = started + Duration::from_secs(27); + let result = loop { + let result = runtime + .tick() + .into_iter() + .find(|f| f.scope.device.control == ControlAddress::Key(0)); + if let Some(feedback) = result { + break feedback.result; + } + assert!(Instant::now() < end, "no lock action feedback"); + std::thread::sleep(Duration::from_millis(10)); + }; + let elapsed = started.elapsed(); + if mode == "slow" { + assert_eq!(result, Ok(())); + assert!( + elapsed >= Duration::from_secs(3), + "fixture did not exceed the old timeout" + ); + } else { + assert_eq!(result, Err("lock_unconfirmed")); + assert!( + elapsed >= Duration::from_secs(8), + "lock confirmation returned prematurely" + ); + // A lock that was not confirmed might still happen later. Preserve + // the existing fail-closed builtin fence until controls restart. + for (time, pressed) in [(3, true), (4, false)] { + deck.lock() + .unwrap() + .inject(InputEvent { + timestamp_ms: time, + event: RawEvent::Key { index: 0, pressed }, + }) + .unwrap(); + } + let end = Instant::now() + Duration::from_secs(5); + let retry = loop { + if let Some(feedback) = runtime + .tick() + .into_iter() + .find(|f| f.scope.device.control == ControlAddress::Key(0)) + { + break feedback.result; + } + assert!(Instant::now() < end, "no blocked retry feedback"); + std::thread::sleep(Duration::from_millis(10)); + }; + assert_eq!(retry, Err("native_builtin_effect_blocked")); + } + println!( + "PASS full VirtualDeck lock action: case={mode}, result={result:?}, elapsed_ms={}", + elapsed.as_millis() + ); + assert!(runtime.stop().iter().all(|(_, result)| result.is_ok())); + std::fs::remove_dir_all(root).unwrap(); + } +} diff --git a/apps/decksmith-daemon/src/feedback.rs b/apps/decksmith-daemon/src/feedback.rs index 09faaf2..a296eab 100644 --- a/apps/decksmith-daemon/src/feedback.rs +++ b/apps/decksmith-daemon/src/feedback.rs @@ -131,6 +131,12 @@ pub fn failed(check: Check, name: String, code: &str) -> Notice { "Not set", "Set brightness", ), + "lock_unconfirmed" => ( + "The desktop lock could not be confirmed", + "Check the desktop lock, fix the locker, then restart Background controls before retrying.", + "Lock unclear", + "Check lock", + ), code if code.ends_with("timeout") => ( "The action timed out; completion is unknown", "Check the target before retrying; a partial change may have occurred.", @@ -495,5 +501,19 @@ mod tests { let timed_out = failed(c, "MPZ".into(), "launch_timeout"); assert!(timed_out.detail.contains("unknown")); assert!(timed_out.hint.contains("before retrying")); + let lock = failed( + Check { + page: 0, + slot: 0, + label: "Lock".into(), + kind: "action".into(), + target: "Lock".into(), + command: String::new(), + }, + "Lock".into(), + "lock_unconfirmed", + ); + assert!(lock.detail.contains("could not be confirmed")); + assert!(lock.hint.contains("restart Background controls")); } } diff --git a/apps/decksmith-daemon/src/foreground.rs b/apps/decksmith-daemon/src/foreground.rs index d4e5fb8..4a93c02 100644 --- a/apps/decksmith-daemon/src/foreground.rs +++ b/apps/decksmith-daemon/src/foreground.rs @@ -1,5 +1,8 @@ //! Latest application identity only. No titles, focus polling, or manual override timer. use serde::Serialize; +use std::io::{BufRead, BufReader, ErrorKind, Read, Write}; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use std::time::Duration; pub const SOURCE: &str = "cc.senecal.Decksmith.Foreground"; @@ -39,6 +42,10 @@ impl Foreground { } pub fn monitor(&self) { let weak = Arc::downgrade(&self.0); + if let Some(dir) = hyprland_dir() { + std::thread::spawn(move || hyprland(weak, dir)); + return; + } std::thread::spawn(move || { while let Some(inner) = weak.upgrade() { let present = futures_lite::future::block_on(async { @@ -72,10 +79,187 @@ impl Foreground { }); } } +fn hyprland_dir() -> Option { + let signature = std::env::var_os("HYPRLAND_INSTANCE_SIGNATURE")?; + let runtime = std::env::var_os("XDG_RUNTIME_DIR")?; + Some(PathBuf::from(runtime).join("hypr").join(signature)) +} +/// Hyprland reports focus on its event socket; no extension or D-Bus source. +fn hyprland(weak: std::sync::Weak>, dir: PathBuf) { + while weak.strong_count() > 0 { + if let Ok(events) = UnixStream::connect(dir.join(".socket2.sock")) { + let _ = events.set_read_timeout(Some(Duration::from_secs(2))); + let class = active_class(&dir).unwrap_or_default(); + let report = |class: &str| match weak.upgrade() { + Some(inner) => { + // Desktop entries can be added, removed or edited while the + // daemon is running. Resolve on focus events so both misses + // and previously successful mappings can recover. + report_class(&Foreground(inner), class, &application_dirs()); + true + } + None => false, + }; + if !report(&class) { + return; + } + let mut reader = BufReader::new(events); + let mut line = String::new(); + loop { + match reader.read_line(&mut line) { + Ok(0) => break, + Ok(_) => { + let class = event_class(&line); + if class.is_some_and(|c| !report(c)) { + return; + } + line.clear(); + } + // Timeouts keep partial lines and only check for shutdown. + Err(e) if timed_out(&e) => { + if weak.strong_count() == 0 { + return; + } + } + Err(_) => break, + } + } + } + match weak.upgrade() { + Some(inner) => Foreground(inner).unavailable(), + None => return, + } + std::thread::sleep(Duration::from_secs(2)); + } +} +fn report_class(foreground: &Foreground, class: &str, dirs: &[PathBuf]) { + let _ = foreground.report(&desktop_id(class, dirs)); +} +/// `activewindow>>CLASS,TITLE` -> CLASS. +fn event_class(line: &str) -> Option<&str> { + let event = line.trim_end().strip_prefix("activewindow>>")?; + Some(event.split_once(',').map_or(event, |(class, _)| class)) +} +fn timed_out(e: &std::io::Error) -> bool { + matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) +} +fn active_class(dir: &Path) -> Option { + let mut s = UnixStream::connect(dir.join(".socket.sock")).ok()?; + s.set_read_timeout(Some(Duration::from_secs(1))).ok()?; + s.write_all(b"j/activewindow").ok()?; + let mut reply = String::new(); + s.read_to_string(&mut reply).ok()?; + let value: serde_json::Value = serde_json::from_str(&reply).ok()?; + Some(value["class"].as_str()?.to_owned()) +} +fn application_dirs() -> Vec { + let home = std::env::var_os("XDG_DATA_HOME") + .map(PathBuf::from) + .or_else(|| { + let home = PathBuf::from(std::env::var_os("HOME")?); + Some(home.join(".local/share")) + }); + let system = std::env::var("XDG_DATA_DIRS") + .ok() + .filter(|d| !d.is_empty()) + .unwrap_or_else(|| "/usr/local/share:/usr/share".into()); + home.into_iter() + .chain(system.split(':').map(PathBuf::from)) + .map(|d| d.join("applications")) + .collect() +} +/// Maps a window class to its desktop entry: `.desktop`, then +/// StartupWMClass. shortcut: top-level entries only, add subdirectory +/// ids if needed. +fn desktop_id(class: &str, dirs: &[PathBuf]) -> String { + if class.is_empty() { + return String::new(); + } + for name in [class.to_owned(), class.to_lowercase()] { + let id = format!("{name}.desktop"); + let installed = dirs.iter().any(|d| d.join(&id).is_file()); + if crate::launch::valid_application(&id) && installed { + return id; + } + } + for dir in dirs { + let Ok(entries) = std::fs::read_dir(dir) else { + continue; + }; + for entry in entries.flatten() { + let id = entry.file_name().to_string_lossy().into_owned(); + if !crate::launch::valid_application(&id) { + continue; + } + let Ok(text) = std::fs::read_to_string(entry.path()) else { + continue; + }; + if text.lines().any(|l| { + l.strip_prefix("StartupWMClass=") + .is_some_and(|v| v.trim().eq_ignore_ascii_case(class)) + }) { + return id; + } + } + } + String::new() +} #[cfg(test)] mod tests { use super::*; #[test] + fn focus_resolves_entries_again_after_install_edit_and_removal() { + let root = + std::env::temp_dir().join(format!("decksmith-focus-refresh-{}", std::process::id())); + let apps = root.join("applications"); + std::fs::create_dir_all(&apps).unwrap(); + let dirs = [apps.clone()]; + let f = Foreground::default(); + report_class(&f, "DynamicApp", &dirs); + assert_eq!(f.snapshot().application, None); + let entry = apps.join("dynamic.desktop"); + std::fs::write(&entry, "[Desktop Entry]\nStartupWMClass=DynamicApp\n").unwrap(); + report_class(&f, "DynamicApp", &dirs); + assert_eq!(f.snapshot().application.as_deref(), Some("dynamic.desktop")); + std::fs::write(&entry, "[Desktop Entry]\nStartupWMClass=Other\n").unwrap(); + report_class(&f, "DynamicApp", &dirs); + assert_eq!(f.snapshot().application, None); + std::fs::write(&entry, "[Desktop Entry]\nStartupWMClass=DynamicApp\n").unwrap(); + report_class(&f, "DynamicApp", &dirs); + std::fs::remove_file(&entry).unwrap(); + report_class(&f, "DynamicApp", &dirs); + assert_eq!(f.snapshot().application, None); + std::fs::remove_dir_all(root).unwrap(); + } + #[test] + fn hyprland_event_line_yields_window_class() { + let line = "activewindow>>google-chrome,Inbox, mail\n"; + assert_eq!(event_class(line), Some("google-chrome")); + assert_eq!(event_class("activewindow>>,\n"), Some("")); + assert_eq!(event_class("workspace>>2\n"), None); + } + #[test] + fn hyprland_class_maps_to_desktop_entry() { + let name = format!("decksmith-apps-{}", std::process::id()); + let root = std::env::temp_dir().join(name); + let apps = root.join("applications"); + std::fs::create_dir_all(&apps).unwrap(); + let entry = "[Desktop Entry]\n"; + std::fs::write(apps.join("brave-browser.desktop"), entry).unwrap(); + std::fs::write( + apps.join("code.desktop"), + "[Desktop Entry]\nStartupWMClass=Code\n", + ) + .unwrap(); + let dirs = [apps]; + assert_eq!(desktop_id("brave-browser", &dirs), "brave-browser.desktop"); + assert_eq!(desktop_id("Brave-browser", &dirs), "brave-browser.desktop"); + assert_eq!(desktop_id("Code", &dirs), "code.desktop"); + assert_eq!(desktop_id("unknown", &dirs), ""); + assert_eq!(desktop_id("", &dirs), ""); + std::fs::remove_dir_all(root).unwrap(); + } + #[test] fn identity_changes_only_and_missing_source_retains_no_stale_target() { let f = Foreground::default(); f.report("brave-browser.desktop").unwrap(); diff --git a/apps/decksmith-daemon/src/shared_builtin.rs b/apps/decksmith-daemon/src/shared_builtin.rs index 043b261..4250a0a 100644 --- a/apps/decksmith-daemon/src/shared_builtin.rs +++ b/apps/decksmith-daemon/src/shared_builtin.rs @@ -1,6 +1,7 @@ //! Ordinary built-ins use the existing adapters with an explicit per-effect //! grant after discovery. A lost reply never causes an automatic replay. use crate::{audio_target::Connection, pages::Action}; +use std::time::Duration; type Result = std::result::Result; #[derive(serde::Deserialize)] #[serde(tag = "status", rename_all = "snake_case", deny_unknown_fields)] @@ -22,8 +23,15 @@ impl Client { if self.helper.is_none() { self.helper = Some(Connection::start_helper("shared_builtin.py", &[])?); } + // Lock confirmation can take eight seconds after the side effect and + // invokes a bounded subprocess. Preserve the shorter default for all + // other built-ins and for audio helper clients. + let timeout = reply_timeout(action); let result = conversation(action, current, |request| { - self.helper.as_mut().unwrap().exchange(request) + self.helper + .as_mut() + .unwrap() + .exchange_with_timeout(request, timeout) }); if result.is_err() { self.helper = None; @@ -31,6 +39,13 @@ impl Client { result } } +fn reply_timeout(action: &Action) -> Duration { + if matches!(action, Action::System { command } if command == "lock") { + Duration::from_secs(25) + } else { + Duration::from_secs(3) + } +} fn conversation( action: &Action, current: impl Fn() -> bool, @@ -66,6 +81,7 @@ fn conversation( return Err(match code.as_str() { "stale" => "stale_action_scope", "unavailable" => "builtin_unavailable", + "lock_unconfirmed" => "lock_unconfirmed", "unsupported" => "unsupported_shared_builtin_action", _ => uncertain("helper_error", apply_attempts), }); @@ -88,6 +104,34 @@ fn uncertain(reason: &'static str, apply_attempts: usize) -> &'static str { mod tests { use super::*; #[test] + fn lock_has_bounded_longer_wait_and_unconfirmed_result_is_preserved() { + let lock = Action::System { + command: "lock".into(), + }; + let other = Action::System { + command: "dnd".into(), + }; + assert_eq!(reply_timeout(&lock), Duration::from_secs(25)); + assert_eq!(reply_timeout(&other), Duration::from_secs(3)); + let mut calls = 0; + assert_eq!( + conversation( + &lock, + || true, + |_| { + calls += 1; + Ok(if calls == 1 { + br#"{"status":"ready"}"#.to_vec() + } else { + br#"{"status":"error","code":"lock_unconfirmed"}"#.to_vec() + }) + } + ), + Err("lock_unconfirmed") + ); + assert_eq!(calls, 2); + } + #[test] fn scope_revoked_after_discovery_aborts_each_effect() { let current = std::cell::Cell::new(true); let mut requests = Vec::new(); diff --git a/apps/decksmith-daemon/src/shared_native_media.rs b/apps/decksmith-daemon/src/shared_native_media.rs index 4437ef7..a8f4cb4 100644 --- a/apps/decksmith-daemon/src/shared_native_media.rs +++ b/apps/decksmith-daemon/src/shared_native_media.rs @@ -372,7 +372,10 @@ impl Backend for NativeBackend { return Err("native_builtin_effect_blocked"); } let result = self.builtin.execute(action, || fence.accepts()); - if result == Err("native_builtin_effect_uncertain") { + if matches!( + result, + Err("native_builtin_effect_uncertain" | "lock_unconfirmed") + ) { self.blocked_builtin = true; self.uncertain.store(true, Ordering::Release); } diff --git a/apps/decksmith-studio/shared_builtin.py b/apps/decksmith-studio/shared_builtin.py index f53954d..235099e 100644 --- a/apps/decksmith-studio/shared_builtin.py +++ b/apps/decksmith-studio/shared_builtin.py @@ -40,7 +40,7 @@ def execute(action, source, output, *, launcher=None, controls=None, grant=None) def serve(source, output): from shared_audio_native import authorize, message, send # Sharing one system adapter avoids duplicated persistent GSettings caches. - from system_controls import Controls,GLib + from system_controls import Controls,GLib,LockUnconfirmed controls=Controls() while True: try:action=message(source) @@ -63,7 +63,7 @@ def grant(): except Exception as error: # Ordinary adapter exceptions may follow an issued effect. Only a # known authorization abort is definitely stale; never replay others. - code='stale' if not grants and str(error)=='Stale action scope' else 'unsupported' if str(error)=='Unsupported built-in' else 'unavailable' if str(error)=='Application unavailable' else 'uncertain' + code='lock_unconfirmed' if isinstance(error,LockUnconfirmed) else 'stale' if not grants and str(error)=='Stale action scope' else 'unsupported' if str(error)=='Unsupported built-in' else 'unavailable' if str(error)=='Application unavailable' else 'uncertain' reply={'status':'error','code':code} send(output,reply) if reply['status']=='error':return diff --git a/apps/decksmith-studio/system_controls.py b/apps/decksmith-studio/system_controls.py index debd634..e0785e2 100644 --- a/apps/decksmith-studio/system_controls.py +++ b/apps/decksmith-studio/system_controls.py @@ -1,112 +1,498 @@ -"""Allowlisted GNOME system controls. Persistent JSON-lines worker; no shell execution.""" -import json,sys,os +"""Allowlisted GNOME/Hyprland system controls. + +Persistent JSON-lines worker; no shell execution. +""" + +import json, sys, os +import shutil, subprocess +import time from pathlib import Path -from gi.repository import Gio,GLib -COMMANDS={'lock','suspend','reboot','shutdown','dnd','night_light','bluetooth','power_cycle','power_saver','power_balanced','power_performance'} -LABELS={'lock':'Lock Desktop','suspend':'Suspend','reboot':'Reboot','shutdown':'Shutdown','dnd':'Do Not Disturb','night_light':'Night Light','bluetooth':'Bluetooth','power_cycle':'Power Mode','power_saver':'Power Saver','power_balanced':'Balanced','power_performance':'Performance'} +from gi.repository import Gio, GLib + +COMMANDS = { + "lock", + "suspend", + "reboot", + "shutdown", + "dnd", + "night_light", + "bluetooth", + "power_cycle", + "power_saver", + "power_balanced", + "power_performance", +} +LABELS = { + "lock": "Lock Desktop", + "suspend": "Suspend", + "reboot": "Reboot", + "shutdown": "Shutdown", + "dnd": "Do Not Disturb", + "night_light": "Night Light", + "bluetooth": "Bluetooth", + "power_cycle": "Power Mode", + "power_saver": "Power Saver", + "power_balanced": "Balanced", + "power_performance": "Performance", +} +HYPRLAND_COMMANDS = ("dnd", "night_light", "lock") + + +def hyprland(): + return bool(os.environ.get("HYPRLAND_INSTANCE_SIGNATURE")) + + +def run(*argv): + """Run a fixed argv (no shell); return stdout, or None on any failure.""" + if not shutil.which(argv[0]): + return None + try: + result = subprocess.run( + argv, + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + timeout=5, + ) + except (OSError, subprocess.TimeoutExpired): + return None + return result.stdout.strip() if result.returncode == 0 else None + + +def run_action(*argv): + """Run an action without pipes that a background child could inherit.""" + if not shutil.which(argv[0]): + return False + try: + result = subprocess.run( + argv, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, timeout=5, + ) + except (OSError, subprocess.TimeoutExpired): + return False + return result.returncode == 0 + + +def omarchy_lock_ready(): + """Check that the Omarchy shell can lock a real graphical session.""" + status = run("omarchy-shell", "lock", "status") + if status is None: + return False + try: + value = json.loads(status) + return bool(value.get("passwordPam") and value.get("realScreens", 0) > 0) + except (ValueError, TypeError, AttributeError): + return False + + +class LockUnconfirmed(ValueError): + """The compositor did not confirm a requested lock.""" + + +def wait_for_hyprland_lock(timeout=8): + """Only report success after Hyprland confirms the session is locked.""" + deadline = time.monotonic() + timeout + while True: + status = run("hyprctl", "-j", "locked") + try: + if json.loads(status or "null").get("locked") is True: + return True + except (ValueError, TypeError, AttributeError): + pass + if time.monotonic() >= deadline: + return False + time.sleep(0.1) + + class Controls: - def __init__(self,authorize=None):self.buses={};self.settings={};self.authorize=authorize or (lambda:None);self.guarded=authorize is not None - def call(self,system,name,path,iface,method,args=None,timeout=1500): - if system not in self.buses:self.buses[system]=Gio.bus_get_sync(Gio.BusType.SYSTEM if system else Gio.BusType.SESSION,None) - bus=self.buses[system] - if method in ('Set','Lock','Suspend'):self.authorize() - return bus.call_sync(name,path,iface,method,args,None,Gio.DBusCallFlags.NO_AUTO_START,timeout,None).unpack() - def prop(self,system,name,path,iface,key): - return self.call(system,name,path,'org.freedesktop.DBus.Properties','Get',GLib.Variant('(ss)',(iface,key)))[0] - def put(self,system,name,path,iface,key,value): - self.call(system,name,path,'org.freedesktop.DBus.Properties','Set',GLib.Variant('(ssv)',(iface,key,value))) - def setting(self,schema): - if schema not in self.settings:self.settings[schema]=Gio.Settings.new(schema) + def __init__(self, authorize=None): + self.buses = {} + self.settings = {} + self.authorize = authorize or (lambda: None) + self.guarded = authorize is not None + + def call(self, system, name, path, iface, method, args=None, timeout=1500): + if system not in self.buses: + self.buses[system] = Gio.bus_get_sync( + Gio.BusType.SYSTEM if system else Gio.BusType.SESSION, None + ) + bus = self.buses[system] + if method in ("Set", "Lock", "Suspend"): + self.authorize() + return bus.call_sync( + name, + path, + iface, + method, + args, + None, + Gio.DBusCallFlags.NO_AUTO_START, + timeout, + None, + ).unpack() + + def prop(self, system, name, path, iface, key): + return self.call( + system, + name, + path, + "org.freedesktop.DBus.Properties", + "Get", + GLib.Variant("(ss)", (iface, key)), + )[0] + + def put(self, system, name, path, iface, key, value): + self.call( + system, + name, + path, + "org.freedesktop.DBus.Properties", + "Set", + GLib.Variant("(ssv)", (iface, key, value)), + ) + + def setting(self, schema): + if schema not in self.settings: + self.settings[schema] = Gio.Settings.new(schema) return self.settings[schema] - def color(self,key):return self.prop(False,'org.gnome.SettingsDaemon.Color','/org/gnome/SettingsDaemon/Color','org.gnome.SettingsDaemon.Color',key) - def rfkill(self,key):return self.prop(False,'org.gnome.SettingsDaemon.Rfkill','/org/gnome/SettingsDaemon/Rfkill','org.gnome.SettingsDaemon.Rfkill',key) + + def color(self, key): + return self.prop( + False, + "org.gnome.SettingsDaemon.Color", + "/org/gnome/SettingsDaemon/Color", + "org.gnome.SettingsDaemon.Color", + key, + ) + + def rfkill(self, key): + return self.prop( + False, + "org.gnome.SettingsDaemon.Rfkill", + "/org/gnome/SettingsDaemon/Rfkill", + "org.gnome.SettingsDaemon.Rfkill", + key, + ) + def profiles(self): - name='net.hadess.PowerProfiles';path='/net/hadess/PowerProfiles' - return self.prop(True,name,path,name,'Profiles'),self.prop(True,name,path,name,'ActiveProfile') - def state(self,command): - if command not in COMMANDS:raise ValueError('Unsupported system action') - if command=='dnd': - if not self.call(False,'org.freedesktop.DBus','/org/freedesktop/DBus','org.freedesktop.DBus','NameHasOwner',GLib.Variant('(s)',('org.gnome.Shell',)))[0]: - return dict(available=False,active=False,text='Unavailable') - s=self.setting('org.gnome.desktop.notifications') - if not s.is_writable('show-banners'):raise ValueError('Notification setting is managed') - active=not s.get_boolean('show-banners');return dict(available=True,active=active,text='On' if active else 'Off') - if command=='night_light': - active=self.color('NightLightActive') and not self.color('DisabledUntilTomorrow') - return dict(available=True,active=active,text='On' if active else 'Off') - if command=='bluetooth': - available=self.rfkill('BluetoothHasAirplaneMode') and not self.rfkill('BluetoothHardwareAirplaneMode') - active=not self.rfkill('BluetoothAirplaneMode') - return dict(available=available,active=active,text=('On' if active else 'Off') if available else 'Unavailable') - if command.startswith('power_'): - profiles,active=self.profiles();desired=command.removeprefix('power_').replace('_','-') - if desired=='saver':desired='power-saver' - available=command=='power_cycle' or desired in [p['Profile'] for p in profiles] - return dict(available=available,active=active==desired or command=='power_cycle',text={'power-saver':'Saver','balanced':'Balanced','performance':'Performance'}.get(active,active)) - if command in ('suspend','reboot','shutdown'): - cap={'suspend':'CanSuspend','reboot':'CanReboot','shutdown':'CanPowerOff'}[command] - value=self.call(True,'org.freedesktop.login1','/org/freedesktop/login1','org.freedesktop.login1.Manager',cap)[0] - return dict(available=value in ('yes','challenge'),active=False,text='Confirm' if command in ('reboot','shutdown') else 'Ready') - self.call(False,'org.gnome.ScreenSaver','/org/gnome/ScreenSaver','org.gnome.ScreenSaver','GetActive') - return dict(available=True,active=False,text='Ready') - def execute(self,command): - if command not in COMMANDS:raise ValueError('Unsupported system action') - state=self.state(command) - if not state['available']:raise ValueError('Control unavailable') - if command=='lock':self.call(False,'org.gnome.ScreenSaver','/org/gnome/ScreenSaver','org.gnome.ScreenSaver','Lock') - elif command=='suspend':self.call(True,'org.freedesktop.login1','/org/freedesktop/login1','org.freedesktop.login1.Manager','Suspend',GLib.Variant('(b)',(True,))) - elif command in ('reboot','shutdown'): + name = "net.hadess.PowerProfiles" + path = "/net/hadess/PowerProfiles" + return self.prop(True, name, path, name, "Profiles"), self.prop( + True, name, path, name, "ActiveProfile" + ) + + def state(self, command): + if command not in COMMANDS: + raise ValueError("Unsupported system action") + if hyprland() and command in HYPRLAND_COMMANDS: + return self.hypr_state(command) + if command == "dnd": + if not self.call( + False, + "org.freedesktop.DBus", + "/org/freedesktop/DBus", + "org.freedesktop.DBus", + "NameHasOwner", + GLib.Variant("(s)", ("org.gnome.Shell",)), + )[0]: + return dict(available=False, active=False, text="Unavailable") + s = self.setting("org.gnome.desktop.notifications") + if not s.is_writable("show-banners"): + raise ValueError("Notification setting is managed") + active = not s.get_boolean("show-banners") + return dict( + available=True, active=active, text="On" if active else "Off" + ) + if command == "night_light": + active = self.color("NightLightActive") and not self.color( + "DisabledUntilTomorrow" + ) + return dict( + available=True, active=active, text="On" if active else "Off" + ) + if command == "bluetooth": + available = self.rfkill( + "BluetoothHasAirplaneMode" + ) and not self.rfkill("BluetoothHardwareAirplaneMode") + active = not self.rfkill("BluetoothAirplaneMode") + return dict( + available=available, + active=active, + text=( + ("On" if active else "Off") if available else "Unavailable" + ), + ) + if command.startswith("power_"): + profiles, active = self.profiles() + desired = command.removeprefix("power_").replace("_", "-") + if desired == "saver": + desired = "power-saver" + available = command == "power_cycle" or desired in [ + p["Profile"] for p in profiles + ] + return dict( + available=available, + active=active == desired or command == "power_cycle", + text={ + "power-saver": "Saver", + "balanced": "Balanced", + "performance": "Performance", + }.get(active, active), + ) + if command in ("suspend", "reboot", "shutdown"): + cap = { + "suspend": "CanSuspend", + "reboot": "CanReboot", + "shutdown": "CanPowerOff", + }[command] + value = self.call( + True, + "org.freedesktop.login1", + "/org/freedesktop/login1", + "org.freedesktop.login1.Manager", + cap, + )[0] + return dict( + available=value in ("yes", "challenge"), + active=False, + text=( + "Confirm" if command in ("reboot", "shutdown") else "Ready" + ), + ) + self.call( + False, + "org.gnome.ScreenSaver", + "/org/gnome/ScreenSaver", + "org.gnome.ScreenSaver", + "GetActive", + ) + return dict(available=True, active=False, text="Ready") + + def execute(self, command): + if command not in COMMANDS: + raise ValueError("Unsupported system action") + state = self.state(command) + if not state["available"]: + raise ValueError("Control unavailable") + if hyprland() and command in HYPRLAND_COMMANDS: + return self.hypr_execute(command, state["active"]) + if command == "lock": + self.call( + False, + "org.gnome.ScreenSaver", + "/org/gnome/ScreenSaver", + "org.gnome.ScreenSaver", + "Lock", + ) + elif command == "suspend": + self.call( + True, + "org.freedesktop.login1", + "/org/freedesktop/login1", + "org.freedesktop.login1.Manager", + "Suspend", + GLib.Variant("(b)", (True,)), + ) + elif command in ("reboot", "shutdown"): import subprocess + self.authorize() - subprocess.Popen(['/usr/bin/python3','-B',str(Path(__file__).with_name('system_confirm.py')),command],start_new_session=self.guarded,stdin=subprocess.DEVNULL,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL) - elif command=='dnd': - setting=self.setting('org.gnome.desktop.notifications') + subprocess.Popen( + [ + "/usr/bin/python3", + "-B", + str(Path(__file__).with_name("system_confirm.py")), + command, + ], + start_new_session=self.guarded, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + elif command == "dnd": + setting = self.setting("org.gnome.desktop.notifications") self.authorize() - if not setting.set_boolean('show-banners',state['active']):raise ValueError('Setting denied') - elif command=='bluetooth':self.put(False,'org.gnome.SettingsDaemon.Rfkill','/org/gnome/SettingsDaemon/Rfkill','org.gnome.SettingsDaemon.Rfkill','BluetoothAirplaneMode',GLib.Variant('b',state['active'])) - elif command=='night_light':self.night_light(not state['active']) + if not setting.set_boolean("show-banners", state["active"]): + raise ValueError("Setting denied") + elif command == "bluetooth": + self.put( + False, + "org.gnome.SettingsDaemon.Rfkill", + "/org/gnome/SettingsDaemon/Rfkill", + "org.gnome.SettingsDaemon.Rfkill", + "BluetoothAirplaneMode", + GLib.Variant("b", state["active"]), + ) + elif command == "night_light": + self.night_light(not state["active"]) else: - profiles,current=self.profiles();names=[p['Profile'] for p in profiles] - desired=command.removeprefix('power_');desired='power-saver' if desired=='saver' else desired - if desired=='cycle':desired=names[(names.index(current)+1)%len(names)] - if desired not in names:raise ValueError('Profile unavailable') - self.put(True,'net.hadess.PowerProfiles','/net/hadess/PowerProfiles','net.hadess.PowerProfiles','ActiveProfile',GLib.Variant('s',desired)) + profiles, current = self.profiles() + names = [p["Profile"] for p in profiles] + desired = command.removeprefix("power_") + desired = "power-saver" if desired == "saver" else desired + if desired == "cycle": + desired = names[(names.index(current) + 1) % len(names)] + if desired not in names: + raise ValueError("Profile unavailable") + self.put( + True, + "net.hadess.PowerProfiles", + "/net/hadess/PowerProfiles", + "net.hadess.PowerProfiles", + "ActiveProfile", + GLib.Variant("s", desired), + ) Gio.Settings.sync() - def night_light(self,on): - s=self.setting('org.gnome.settings-daemon.plugins.color') - keys=('night-light-schedule-automatic','night-light-schedule-from','night-light-schedule-to') - path=Path(os.environ.get('XDG_CONFIG_HOME',Path.home()/'.config'))/'decksmith/night-light-schedule.json' + + def hypr_state(self, command): + def on(active): + text = "On" if active else "Off" + return dict(available=True, active=active, text=text) + + unavailable = dict(available=False, active=False, text="Unavailable") + if command == "lock": + if omarchy_lock_ready(): + return dict(available=True, active=False, text="Ready") + if run("pgrep", "-u", str(os.getuid()), "-x", "hypridle"): + return dict(available=True, active=False, text="Check") + return unavailable + if command == "dnd": + value = run("omarchy-shell", "notifications", "isDnd") + if value in ("on", "off"): + return on(value == "on") + modes = run("makoctl", "mode") + if modes is not None: + return on("do-not-disturb" in modes.split()) + return unavailable + temp = run("hyprctl", "hyprsunset", "temperature") + if temp and temp.split()[0].isdigit(): + return on(int(temp.split()[0]) < 6000) + # Omarchy starts hyprsunset on demand, so a stopped one means off. + if shutil.which("omarchy-toggle-nightlight"): + return on(False) + return unavailable + + def hypr_execute(self, command, active): + if command == "lock": + if omarchy_lock_ready(): + self.authorize() + if run("omarchy-shell", "lock", "lock") != "ok": + raise ValueError("Control unavailable") + else: + # logind asks hypridle to run its configured lock_cmd. + self.call( + True, + "org.freedesktop.login1", + "/org/freedesktop/login1/session/auto", + "org.freedesktop.login1.Session", + "Lock", + ) + if not wait_for_hyprland_lock(): + raise LockUnconfirmed("Lock not confirmed; desktop may be unlocked") + return + self.authorize() + if command == "dnd" and shutil.which("omarchy-shell"): + value = "off" if active else "on" + ok = run("omarchy-shell", "notifications", "setDnd", value) + run("omarchy-shell", "-q", "omarchy.indicators", "refresh") + elif command == "dnd": + ok = run("makoctl", "mode", "-t", "do-not-disturb") + elif shutil.which("omarchy-toggle-nightlight"): + if not run_action("omarchy-toggle-nightlight"): + raise ValueError("Control unavailable") + return + else: + value = "6500" if active else "4000" + ok = run("hyprctl", "hyprsunset", "temperature", value) + if active and ok is not None: + ok = run("hyprctl", "hyprsunset", "identity") + if ok is None: + raise ValueError("Control unavailable") + + def night_light(self, on): + s = self.setting("org.gnome.settings-daemon.plugins.color") + keys = ( + "night-light-schedule-automatic", + "night-light-schedule-from", + "night-light-schedule-to", + ) + path = ( + Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) + / "decksmith/night-light-schedule.json" + ) if on: if not path.exists(): self.authorize() - path.parent.mkdir(parents=True,exist_ok=True);temp=path.with_suffix('.tmp') - temp.write_text(json.dumps({k:s.get_value(k).unpack() for k in keys}));temp.replace(path) - s.delay();s.set_boolean(keys[0],False);s.set_double(keys[1],0.0);s.set_double(keys[2],24.0);s.set_boolean('night-light-enabled',True);self.authorize();s.apply() - self.put(False,'org.gnome.SettingsDaemon.Color','/org/gnome/SettingsDaemon/Color','org.gnome.SettingsDaemon.Color','DisabledUntilTomorrow',GLib.Variant('b',False)) + path.parent.mkdir(parents=True, exist_ok=True) + temp = path.with_suffix(".tmp") + temp.write_text( + json.dumps({k: s.get_value(k).unpack() for k in keys}) + ) + temp.replace(path) + s.delay() + s.set_boolean(keys[0], False) + s.set_double(keys[1], 0.0) + s.set_double(keys[2], 24.0) + s.set_boolean("night-light-enabled", True) + self.authorize() + s.apply() + self.put( + False, + "org.gnome.SettingsDaemon.Color", + "/org/gnome/SettingsDaemon/Color", + "org.gnome.SettingsDaemon.Color", + "DisabledUntilTomorrow", + GLib.Variant("b", False), + ) else: - s.delay();s.set_boolean('night-light-enabled',False) - if path.exists() and not s.get_boolean(keys[0]) and s.get_double(keys[1])==0.0 and s.get_double(keys[2])==24.0: - # Do not overwrite schedule edits made in GNOME while the override was on. - data=json.loads(path.read_text()) - for key in keys:s.set_value(key,GLib.Variant('b' if key==keys[0] else 'd',data[key])) - self.authorize();s.apply() + s.delay() + s.set_boolean("night-light-enabled", False) + if ( + path.exists() + and not s.get_boolean(keys[0]) + and s.get_double(keys[1]) == 0.0 + and s.get_double(keys[2]) == 24.0 + ): + # Do not overwrite schedule edits made in GNOME while the + # override was on. + data = json.loads(path.read_text()) + for key in keys: + s.set_value( + key, + GLib.Variant("b" if key == keys[0] else "d", data[key]), + ) + self.authorize() + s.apply() if path.exists(): - self.authorize();path.unlink() + self.authorize() + path.unlink() + def main(): - controls=Controls() + controls = Controls() for line in sys.stdin: - context=GLib.MainContext.default() - while context.pending():context.iteration(False) + context = GLib.MainContext.default() + while context.pending(): + context.iteration(False) try: - request=json.loads(line) - if 'execute' in request:controls.execute(request['execute']);reply={'ok':True} + request = json.loads(line) + if "execute" in request: + controls.execute(request["execute"]) + reply = {"ok": True} else: - result={} - for command in request.get('read',[]): - try:result[command]=controls.state(command) - except Exception:result[command]=dict(available=False,active=False,text='Unavailable') - reply={'ok':True,'states':result} - except Exception:reply={'ok':False,'error':'system_action_failed'} - print(json.dumps(reply),flush=True) -if __name__=='__main__':main() + result = {} + for command in request.get("read", []): + try: + result[command] = controls.state(command) + except Exception: + result[command] = dict( + available=False, active=False, text="Unavailable" + ) + reply = {"ok": True, "states": result} + except Exception: + reply = {"ok": False, "error": "system_action_failed"} + print(json.dumps(reply), flush=True) + + +if __name__ == "__main__": + main() diff --git a/apps/decksmith-studio/test_shared_builtin.py b/apps/decksmith-studio/test_shared_builtin.py index fc18d88..23949eb 100644 --- a/apps/decksmith-studio/test_shared_builtin.py +++ b/apps/decksmith-studio/test_shared_builtin.py @@ -11,6 +11,14 @@ def abort():raise ValueError('Stale action scope') class SharedBuiltinTests(unittest.TestCase): + def test_unconfirmed_lock_is_sent_as_distinct_failure(self): + from system_controls import LockUnconfirmed + source=io.StringIO('{"type":"system","command":"lock"}\n') + output=io.StringIO() + with patch.object(shared,'execute',side_effect=LockUnconfirmed('Lock not confirmed')): + shared.serve(source,output) + self.assertEqual(json.loads(output.getvalue()),{'status':'error','code':'lock_unconfirmed'}) + def test_application_discovery_precedes_grant_and_stale_never_launches(self): app=Mock();app.should_show.return_value=True with patch.object(launch.Gio.DesktopAppInfo,'new',return_value=app) as discover: diff --git a/apps/decksmith-studio/test_system_controls.py b/apps/decksmith-studio/test_system_controls.py index 1a99346..be547d7 100644 --- a/apps/decksmith-studio/test_system_controls.py +++ b/apps/decksmith-studio/test_system_controls.py @@ -1,60 +1,262 @@ +import os +import subprocess +import sys +import tempfile +import time import unittest -from unittest.mock import Mock,patch -from system_controls import Controls,COMMANDS,LABELS +from unittest.mock import Mock, patch +from system_controls import Controls, COMMANDS, LABELS from system_confirm import confirmed_power + + class SystemControlTests(unittest.TestCase): - def test_non_gnome_dnd_never_claims_or_changes_gnome_notification_settings(self): - c=Controls();c.call=Mock(return_value=(False,));c.setting=Mock() - self.assertEqual(c.state('dnd'),{'available':False,'active':False,'text':'Unavailable'}) - with self.assertRaises(ValueError):c.execute('dnd') - c.setting.assert_not_called() - self.assertEqual(c.call.call_args.args[4],'NameHasOwner') - def test_gnome_dnd_still_reads_and_toggles_banner_setting(self): - c=Controls();c.call=Mock(return_value=(True,));setting=Mock();setting.get_boolean.return_value=True - c.setting=Mock(return_value=setting) - self.assertEqual(c.state('dnd'),{'available':True,'active':False,'text':'Off'}) - with patch('system_controls.Gio.Settings.sync'):c.execute('dnd') - setting.set_boolean.assert_called_once_with('show-banners',False) - def test_unknown_command_rejected_before_bus_access(self): - c=Controls();c.call=Mock() - with self.assertRaises(ValueError):c.execute('run_shell') - c.call.assert_not_called() - def test_power_actions_only_open_confirmation(self): - for command in ('reboot','shutdown'): - c=Controls();c.state=Mock(return_value={'available':True});c.call=Mock() - with patch('system_controls.Gio.Settings.sync'),patch('subprocess.Popen') as spawn:c.execute(command) - self.assertEqual(spawn.call_args.args[0][-1],command);c.call.assert_not_called() - def test_inhibitor_blocks_confirmed_power(self): - c=Mock();c.call.return_value=(True,) - with self.assertRaises(RuntimeError):confirmed_power('reboot',c) - self.assertEqual(c.call.call_count,1) - def test_confirmed_power_uses_nonforced_authorized_method(self): - c=Mock();c.call.side_effect=[(False,),()];confirmed_power('shutdown',c) - self.assertEqual(c.call.call_args.args[4],'PowerOff');self.assertEqual(c.call.call_args.args[5].unpack(),(True,)) - def test_labels_and_icons_complete(self): - from system_artwork import image - self.assertEqual(set(LABELS),COMMANDS) - for command,label in LABELS.items(): - self.assertLessEqual(len(label),24);self.assertTrue(image(command).startswith(b'\x89PNG')) - - def test_assignment_inherits_theme_and_preserves_custom_styles(self): - from system_artwork import populate - from themes import effective - key={'action':{'type':'system','command':'lock'},'label':'Lock'} - populate(key) - self.assertNotIn('label_position',key) - self.assertNotIn('label_background',key) - layout={'theme':{'preset':'light','appearance':{'label_position':'top','label_background':'dark'}}} - style=effective(layout,key) - self.assertEqual(style['icon_size'],75) - self.assertEqual(style['background'],[236,240,245]) - self.assertEqual(style['label_position'],'top') - self.assertEqual(style['label_background'],'dark') - layout['theme']['appearance']['icon_size']=60 - self.assertEqual(effective(layout,key)['icon_size'],60) - key.update(label_position='bottom',background_color='blue',appearance={'icon_size':50}) - populate(key) - style=effective(layout,key) - self.assertEqual(style['icon_size'],50) - self.assertEqual(style['label_position'],'bottom') - self.assertEqual(style['background'],[90,170,255]) + def test_non_gnome_dnd_never_claims_or_changes_gnome_notification_settings( + self, + ): + c = Controls() + c.call = Mock(return_value=(False,)) + c.setting = Mock() + self.assertEqual( + c.state("dnd"), + {"available": False, "active": False, "text": "Unavailable"}, + ) + with self.assertRaises(ValueError): + c.execute("dnd") + c.setting.assert_not_called() + self.assertEqual(c.call.call_args.args[4], "NameHasOwner") + + def test_gnome_dnd_still_reads_and_toggles_banner_setting(self): + c = Controls() + c.call = Mock(return_value=(True,)) + setting = Mock() + setting.get_boolean.return_value = True + c.setting = Mock(return_value=setting) + self.assertEqual( + c.state("dnd"), {"available": True, "active": False, "text": "Off"} + ) + with patch("system_controls.Gio.Settings.sync"): + c.execute("dnd") + setting.set_boolean.assert_called_once_with("show-banners", False) + + def test_unknown_command_rejected_before_bus_access(self): + c = Controls() + c.call = Mock() + with self.assertRaises(ValueError): + c.execute("run_shell") + c.call.assert_not_called() + + def test_power_actions_only_open_confirmation(self): + for command in ("reboot", "shutdown"): + c = Controls() + c.state = Mock(return_value={"available": True}) + c.call = Mock() + with patch("system_controls.Gio.Settings.sync"), patch( + "subprocess.Popen" + ) as spawn: + c.execute(command) + self.assertEqual(spawn.call_args.args[0][-1], command) + c.call.assert_not_called() + + def test_inhibitor_blocks_confirmed_power(self): + c = Mock() + c.call.return_value = (True,) + with self.assertRaises(RuntimeError): + confirmed_power("reboot", c) + self.assertEqual(c.call.call_count, 1) + + def test_confirmed_power_uses_nonforced_authorized_method(self): + c = Mock() + c.call.side_effect = [(False,), ()] + confirmed_power("shutdown", c) + self.assertEqual(c.call.call_args.args[4], "PowerOff") + self.assertEqual(c.call.call_args.args[5].unpack(), (True,)) + + def test_labels_and_icons_complete(self): + from system_artwork import image + + self.assertEqual(set(LABELS), COMMANDS) + for command, label in LABELS.items(): + self.assertLessEqual(len(label), 24) + self.assertTrue(image(command).startswith(b"\x89PNG")) + + def test_assignment_inherits_theme_and_preserves_custom_styles(self): + from system_artwork import populate + from themes import effective + + key = {"action": {"type": "system", "command": "lock"}, "label": "Lock"} + populate(key) + self.assertNotIn("label_position", key) + self.assertNotIn("label_background", key) + layout = { + "theme": { + "preset": "light", + "appearance": { + "label_position": "top", + "label_background": "dark", + }, + } + } + style = effective(layout, key) + self.assertEqual(style["icon_size"], 75) + self.assertEqual(style["background"], [236, 240, 245]) + self.assertEqual(style["label_position"], "top") + self.assertEqual(style["label_background"], "dark") + layout["theme"]["appearance"]["icon_size"] = 60 + self.assertEqual(effective(layout, key)["icon_size"], 60) + key.update( + label_position="bottom", + background_color="blue", + appearance={"icon_size": 50}, + ) + populate(key) + style = effective(layout, key) + self.assertEqual(style["icon_size"], 50) + self.assertEqual(style["label_position"], "bottom") + self.assertEqual(style["background"], [90, 170, 255]) + + +# Tests may run inside a real Hyprland session; default to GNOME behaviour. +_environment = patch.dict("os.environ") + + +def setUpModule(): + _environment.start() + os.environ.pop("HYPRLAND_INSTANCE_SIGNATURE", None) + + +def tearDownModule(): + _environment.stop() + + +@patch.dict("os.environ", {"HYPRLAND_INSTANCE_SIGNATURE": "test"}) +class HyprlandControlTests(unittest.TestCase): + def test_uses_omarchy_and_logind_instead_of_gnome(self): + calls = [] + replies = {"isDnd": "off", "temperature": "4000"} + + def run(*argv): + calls.append(argv) + return replies.get(argv[2] if len(argv) > 2 else "", "ok") + + c = Controls() + c.call = Mock() + c.setting = Mock() + with ( + patch("system_controls.run", side_effect=run), + patch("system_controls.run_action", side_effect=lambda *argv: calls.append(argv) or True), + patch("system_controls.shutil.which", return_value="/usr/bin/x"), + patch("system_controls.wait_for_hyprland_lock", return_value=True), + ): + off = {"available": True, "active": False, "text": "Off"} + on = {"available": True, "active": True, "text": "On"} + self.assertEqual(c.state("dnd"), off) + self.assertEqual(c.state("night_light"), on) + c.execute("dnd") + c.execute("night_light") + c.execute("lock") + self.assertIn(("omarchy-shell", "notifications", "setDnd", "on"), calls) + self.assertIn(("omarchy-toggle-nightlight",), calls) + lock = ("org.freedesktop.login1.Session", "Lock") + self.assertEqual(c.call.call_args.args[3:5], lock) + c.setting.assert_not_called() + + def test_without_notification_daemon_is_unavailable(self): + with ( + patch("system_controls.run", return_value=None), + patch("system_controls.shutil.which", return_value=None), + ): + self.assertFalse(Controls().state("dnd")["available"]) + self.assertFalse(Controls().state("night_light")["available"]) + + def test_omarchy_night_light_failure_is_reported(self): + with (patch("system_controls.run", return_value=None), + patch("system_controls.run_action", return_value=False), + patch("system_controls.shutil.which", return_value="/usr/bin/x")): + with self.assertRaisesRegex(ValueError, "Control unavailable"): + Controls().execute("night_light") + + def test_omarchy_lock_uses_shell_and_rejects_failure(self): + def helper(*argv): + if argv == ("omarchy-shell", "lock", "status"): + return '{"passwordPam":true,"realScreens":1}' + if argv == ("omarchy-shell", "lock", "lock"): + return "ok" + return None + + with patch("system_controls.run", side_effect=helper) as run, patch( + "system_controls.wait_for_hyprland_lock", return_value=True + ): + controls = Controls() + self.assertTrue(controls.state("lock")["available"]) + controls.execute("lock") + self.assertIn(unittest.mock.call("omarchy-shell", "lock", "lock"), run.call_args_list) + with patch("system_controls.run", side_effect=lambda *args: helper(*args) if args[-1] == "status" else "failed"): + with self.assertRaisesRegex(ValueError, "Control unavailable"): + Controls().execute("lock") + + def test_no_locker_does_not_claim_lock_is_available(self): + with patch("system_controls.run", return_value=None): + self.assertFalse(Controls().state("lock")["available"]) + with self.assertRaisesRegex(ValueError, "Control unavailable"): + Controls().execute("lock") + + def test_hypridle_without_confirmed_lock_reports_failure(self): + def helper(*argv): + return "123" if argv[0] == "pgrep" else None + + controls = Controls() + controls.call = Mock() + with patch("system_controls.run", side_effect=helper), patch( + "system_controls.wait_for_hyprland_lock", return_value=False + ): + self.assertEqual(controls.state("lock")["text"], "Check") + with self.assertRaisesRegex(ValueError, "Lock not confirmed"): + controls.execute("lock") + self.assertEqual(controls.call.call_args.args[4], "Lock") + + def test_lock_confirmation_requires_true_compositor_state(self): + from system_controls import wait_for_hyprland_lock + + with patch("system_controls.run", return_value='{"locked":true}'): + self.assertTrue(wait_for_hyprland_lock(timeout=0)) + for status in ('{"locked":false}', '{}', 'unrecognized', None): + with self.subTest(status=status), patch("system_controls.run", return_value=status): + self.assertFalse(wait_for_hyprland_lock(timeout=0)) + + +class ActionProcessTests(unittest.TestCase): + def test_action_does_not_wait_for_a_background_child_to_close_output(self): + from system_controls import run_action + + # Model Omarchy's launcher: its child inherits stdout and stays alive + # after the helper exits. Capturing a pipe would wait for that child. + with tempfile.TemporaryDirectory() as directory: + pidfile = os.path.join(directory, "child.pid") + code = ( + "import subprocess,sys; " + "p=subprocess.Popen([sys.executable,'-c','import time; time.sleep(8)']); " + "open(sys.argv[1], 'w').write(str(p.pid))" + ) + try: + start = time.monotonic() + self.assertTrue(run_action(sys.executable, "-c", code, pidfile)) + self.assertLess(time.monotonic() - start, 4) + finally: + if os.path.exists(pidfile): + try: + with open(pidfile) as child_pid: + os.kill(int(child_pid.read()), 15) + except ProcessLookupError: + pass + + def test_action_timeout_is_failure(self): + from system_controls import run_action + + with patch("system_controls.subprocess.run", side_effect=subprocess.TimeoutExpired("helper", 5)): + self.assertFalse(run_action(sys.executable, "-c", "pass")) + + def test_query_timeout_is_unavailable(self): + from system_controls import run + + with patch("system_controls.subprocess.run", side_effect=subprocess.TimeoutExpired("helper", 5)): + self.assertIsNone(run(sys.executable, "-c", "pass")) diff --git a/docs/desktop-integration.md b/docs/desktop-integration.md index b833d6e..d6e9922 100644 --- a/docs/desktop-integration.md +++ b/docs/desktop-integration.md @@ -14,6 +14,21 @@ These checks do not qualify an entire distribution or physical device lifecycle. Profiles, pause, diagnostics, and other desktop indicators are future work. +## Hyprland and Omarchy (Arch) + +Hyprland sessions don't need an extension. The daemon picks up the focused app +from Hyprland's event socket and matches its window class to a `.desktop` entry, +so app pages switch the same way they do on GNOME. When Hyprland is running, +these system actions use Hyprland tools in place of GNOME services: + +- **Lock** uses the Omarchy shell when available, or asks logind to run your configured hypridle `lock_cmd`. It reports success only if Hyprland confirms the session is locked. If confirmation fails, check whether the desktop is locked, fix the locker, and restart Background controls before retrying; the lock may still happen late. +- **Do Not Disturb** uses the Omarchy shell, or `makoctl` mode `do-not-disturb`. +- **Night Light** uses `omarchy-toggle-nightlight`, or hyprsunset directly. + +Suspend, reboot, shutdown, Bluetooth and power profiles use the same system +services on every desktop. No signed Arch bundle exists, so build from source. +The runtime doctor lists the Arch package names you need. + ## Install the GNOME menu From a runtime bundle or checkout, run: diff --git a/scripts/bootstrap-install.py b/scripts/bootstrap-install.py index 669bc25..d37071d 100644 --- a/scripts/bootstrap-install.py +++ b/scripts/bootstrap-install.py @@ -1,5 +1,9 @@ #!/usr/bin/env python3 -"""Select, download and authenticate a published Decksmith package before installation.""" +"""Select, download and authenticate a published Decksmith package. + +Authentication happens before installation. +""" + import argparse import hashlib import importlib.util @@ -24,108 +28,212 @@ from runtime_dependencies import runtime_packages RELEASES = { - ('fedora', '44'): { - 'target': 'fedora44', 'label': 'Fedora Workstation 44 GNOME/Wayland', - 'version': 'v1.5.1', 'source': '4891555b388f1806ca0cda909d77af6dbd77a8ee', - 'manager': 'dnf', 'experimental': False, + ("fedora", "44"): { + "target": "fedora44", + "label": "Fedora Workstation 44 GNOME/Wayland", + "version": "v1.5.1", + "source": "4891555b388f1806ca0cda909d77af6dbd77a8ee", + "manager": "dnf", + "experimental": False, }, - ('fedora', '45'): { - 'target': 'fedora44', 'label': 'Fedora Workstation 45 GNOME/Wayland (Fedora 44 bundle)', - 'version': 'v1.5.1', 'source': '4891555b388f1806ca0cda909d77af6dbd77a8ee', - 'manager': 'dnf', 'experimental': True, + ("fedora", "45"): { + "target": "fedora44", + "label": "Fedora Workstation 45 GNOME/Wayland (Fedora 44 bundle)", + "version": "v1.5.1", + "source": "4891555b388f1806ca0cda909d77af6dbd77a8ee", + "manager": "dnf", + "experimental": True, }, - ('debian', '13'): { - 'target': 'debian13', 'label': 'Debian 13 GNOME/Wayland', - 'version': 'v1.0.0-debian-preview.1', 'source': 'dee2f2c285041dc19388ed579b90886e4b58341a', - 'manager': 'apt-get', 'experimental': True, + ("debian", "13"): { + "target": "debian13", + "label": "Debian 13 GNOME/Wayland", + "version": "v1.0.0-debian-preview.1", + "source": "dee2f2c285041dc19388ed579b90886e4b58341a", + "manager": "apt-get", + "experimental": True, }, - ('ubuntu', '26.04'): { - 'target': 'ubuntu2604', 'label': 'Ubuntu 26.04 GNOME/Wayland', - 'version': 'v1.0.0-ubuntu-preview.1', 'source': 'e627ea4f7f660a6ff6ad4772a013989b36b717c4', - 'manager': 'apt-get', 'experimental': True, + ("ubuntu", "26.04"): { + "target": "ubuntu2604", + "label": "Ubuntu 26.04 GNOME/Wayland", + "version": "v1.0.0-ubuntu-preview.1", + "source": "e627ea4f7f660a6ff6ad4772a013989b36b717c4", + "manager": "apt-get", + "experimental": True, }, - ('pop', '24.04'): { - 'target': 'popos2404', 'label': 'Pop!_OS 24.04 COSMIC', - 'version': 'v1.0.0-popos-preview.1', 'source': 'be238905bb3f6aa793dcda2df063fa5a2f206cbc', - 'manager': 'apt-get', 'experimental': True, + ("pop", "24.04"): { + "target": "popos2404", + "label": "Pop!_OS 24.04 COSMIC", + "version": "v1.0.0-popos-preview.1", + "source": "be238905bb3f6aa793dcda2df063fa5a2f206cbc", + "manager": "apt-get", + "experimental": True, }, } +# Arch has no signed bundle: build the main branch on the machine. +SOURCE_REPOSITORY = "https://github.com/infamous-pattern/decksmith.git" +# Command -> package, skipped when the command exists (e.g. rustup's cargo). +BUILD_TOOLS = { + "git": "git", + "cargo": "rust", + "cc": "gcc", + "pkgconf": "pkgconf", + "strip": "binutils", +} +ARCH = { + "target": "arch", + "label": "Arch Linux (source build)", + "version": "main", + "source": SOURCE_REPOSITORY, + "manager": "pacman", + "experimental": False, +} + # Upstream's immutable v2.102.0 release fixes workflow identity verification. -# Digest taken from cli/cli's official gh_2.102.0_checksums.txt, reviewed Oct 11. -GH_VERSION = '2.102.0' -GH_ARCHIVE = 'gh_' + GH_VERSION + '_linux_amd64.tar.gz' -GH_SHA256 = 'bb766f710eef8ede859c18578c72c327597cd4c8a85b06001b1f3843c6019386' +# Digest taken from cli/cli's official gh_2.102.0_checksums.txt, +# reviewed Oct 11. +GH_VERSION = "2.102.0" +GH_ARCHIVE = "gh_" + GH_VERSION + "_linux_amd64.tar.gz" +GH_SHA256 = "bb766f710eef8ede859c18578c72c327597cd4c8a85b06001b1f3843c6019386" + + +PRESERVED = ( + "Saved data and startup preferences are preserved; background controls " + "will not restart." +) def supports_attestation(command): try: - version = subprocess.run([command, '--version'], capture_output=True, text=True, timeout=15) - match = re.search(r'^gh version (\d+)\.(\d+)\.(\d+)\b', version.stdout or '', re.MULTILINE) - if version.returncode or not match or tuple(map(int, match.groups())) < (2, 102, 0): + version = subprocess.run( + [command, "--version"], capture_output=True, text=True, timeout=15 + ) + match = re.search( + r"^gh version (\d+)\.(\d+)\.(\d+)\b", + version.stdout or "", + re.MULTILINE, + ) + if ( + version.returncode + or not match + or tuple(map(int, match.groups())) < (2, 102, 0) + ): return False - return subprocess.run([command, 'attestation', 'verify', '--help'], - capture_output=True, text=True, timeout=15).returncode == 0 + return ( + subprocess.run( + [command, "attestation", "verify", "--help"], + capture_output=True, + text=True, + timeout=15, + ).returncode + == 0 + ) except OSError: return False def verification_cli(directory, *, allow_portable=False): - if supports_attestation('gh'): - return 'gh' - message = ('GitHub CLI 2.102.0+ is required for attestation verification. ' - 'Use its official package source, or allow a pinned temporary upstream copy.') + if supports_attestation("gh"): + return "gh" + message = ( + "GitHub CLI 2.102.0+ is required for attestation verification. " + "Use its official package source, or allow a pinned temporary " + "upstream copy." + ) if not allow_portable: try: # A controlling terminal is not seekable: buffered r+ creates a # BufferedRandom stream and fails before the consent prompt appears. - with open('/dev/tty', 'r', encoding='utf-8') as terminal, \ - open('/dev/tty', 'w', encoding='utf-8') as display: - display.write(message + '\nDownload GitHub CLI ' + GH_VERSION + - ' temporarily for this installation? [y/N] ') + with open("/dev/tty", "r", encoding="utf-8") as terminal, open( + "/dev/tty", "w", encoding="utf-8" + ) as display: + display.write( + message + + "\nDownload GitHub CLI " + + GH_VERSION + + " temporarily for this installation? [y/N] " + ) display.flush() - allow_portable = terminal.readline().strip().lower() in ('y', 'yes') + allow_portable = terminal.readline().strip().lower() in ( + "y", + "yes", + ) except OSError as error: - raise ValueError(message + ' Retry with --allow-portable-gh to opt in.') from error + raise ValueError( + message + " Retry with --allow-portable-gh to opt in." + ) from error if not allow_portable: - raise ValueError(message + ' Installation stopped; no signatures were bypassed.') + raise ValueError( + message + " Installation stopped; no signatures were bypassed." + ) archive = directory / GH_ARCHIVE - print('Downloading pinned GitHub CLI ' + GH_VERSION + ' for temporary verification.', flush=True) - download('https://github.com/cli/cli/releases/download/v' + GH_VERSION + '/' + GH_ARCHIVE, - archive, 64 * 1024 * 1024) - with archive.open('rb') as incoming: - if hashlib.file_digest(incoming, 'sha256').hexdigest() != GH_SHA256: - raise ValueError('GitHub CLI checksum differs from the reviewed pin; refusing execution.') - # Read just the pinned archive's regular executable, never extract its paths. + print( + "Downloading pinned GitHub CLI " + + GH_VERSION + + " for temporary verification.", + flush=True, + ) + download( + "https://github.com/cli/cli/releases/download/v" + + GH_VERSION + + "/" + + GH_ARCHIVE, + archive, + 64 * 1024 * 1024, + ) + with archive.open("rb") as incoming: + if hashlib.file_digest(incoming, "sha256").hexdigest() != GH_SHA256: + raise ValueError( + "GitHub CLI checksum differs from the reviewed pin; " + "refusing execution." + ) + # Read just the pinned archive's regular executable, never extract its + # paths. with tarfile.open(archive) as package: - member = package.getmember('gh_' + GH_VERSION + '_linux_amd64/bin/gh') + member = package.getmember("gh_" + GH_VERSION + "_linux_amd64/bin/gh") if not member.isfile() or member.size > 64 * 1024 * 1024: - raise ValueError('Invalid GitHub CLI executable') + raise ValueError("Invalid GitHub CLI executable") with package.extractfile(member) as incoming: - executable = directory / 'gh' - with executable.open('xb') as outgoing: + executable = directory / "gh" + with executable.open("xb") as outgoing: shutil.copyfileobj(incoming, outgoing) executable.chmod(0o700) if not supports_attestation(str(executable)): - raise ValueError('Pinned GitHub CLI is unavailable; refusing installation.') + raise ValueError( + "Pinned GitHub CLI is unavailable; refusing installation." + ) return str(executable) +def arch_family(release): + return "arch" in { + release.get("ID", ""), + *release.get("ID_LIKE", "").split(), + } + + def select_release(release, *, experimental=False): - key = (release.get('ID'), release.get('VERSION_ID')) + if arch_family(release): + return ARCH + key = (release.get("ID"), release.get("VERSION_ID")) if key not in RELEASES: - raise ValueError('No verified download is configured for this OS/version. ' - 'See the support table for available platforms.') + raise ValueError( + "No verified download is configured for this OS/version. " + "See the support table for available platforms." + ) selected = RELEASES[key] - if selected['experimental'] and not experimental: - raise ValueError(selected['label'] + ' is experimental. Read its preview limits, ' - 'then use --experimental to opt in.') + if selected["experimental"] and not experimental: + raise ValueError( + selected["label"] + " is experimental. Read its preview limits, " + "then use --experimental to opt in." + ) return selected def verifier_module(): spec = importlib.util.spec_from_file_location( - 'reviewed_verifier', Path(__file__).with_name('verify-install.py')) + "reviewed_verifier", Path(__file__).with_name("verify-install.py") + ) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module @@ -134,121 +242,293 @@ def verifier_module(): def missing_packages(packages, manager): missing = [] for package in packages: - command = (['rpm', '-q', package] if manager == 'dnf' else - ['dpkg-query', '-W', '-f=${Status}', package]) - result = subprocess.run(command, capture_output=True, text=True, timeout=15) - if result.returncode or (manager != 'dnf' and result.stdout.strip() != 'install ok installed'): + command = ( + ["rpm", "-q", package] + if manager == "dnf" + else ( + ["pacman", "-Q", package] + if manager == "pacman" + else ["dpkg-query", "-W", "-f=${Status}", package] + ) + ) + result = subprocess.run( + command, capture_output=True, text=True, timeout=15 + ) + if result.returncode or ( + manager == "apt-get" + and result.stdout.strip() != "install ok installed" + ): missing.append(package) return missing def install_dependencies(packages, selected, *, no_deps=False): - missing = missing_packages(packages, selected['manager']) + missing = missing_packages(packages, selected["manager"]) if not missing: return - command = ['sudo', selected['manager'], 'install', *missing] - print('Missing dependencies: ' + ', '.join(missing), flush=True) + command = ( + ["sudo", "pacman", "-S", "--needed", *missing] + if selected["manager"] == "pacman" + else ["sudo", selected["manager"], "install", *missing] + ) + print("Missing dependencies: " + ", ".join(missing), flush=True) if no_deps: - raise ValueError('Install these packages from your OS repositories, then retry: ' - + ' '.join(command)) - if not shutil.which('sudo'): - raise ValueError('sudo is unavailable. Ask an administrator to install: ' + ', '.join(missing)) + raise ValueError( + "Install these packages from your OS repositories, then retry: " + + " ".join(command) + ) + if not shutil.which("sudo"): + raise ValueError( + "sudo is unavailable. Ask an administrator to install: " + + ", ".join(missing) + ) try: - terminal = open('/dev/tty', 'r', encoding='utf-8') + terminal = open("/dev/tty", "r", encoding="utf-8") except OSError as error: - raise ValueError('Dependency installation needs a terminal. Run: ' + ' '.join(command)) from error + raise ValueError( + "Dependency installation needs a terminal. Run: " + + " ".join(command) + ) from error # Trusted OS package sources, normal package-manager confirmation; never -y. with terminal: subprocess.run(command, stdin=terminal, check=True) - if missing_packages(packages, selected['manager']): - raise ValueError('Required packages are still missing; installation stopped.') + if missing_packages(packages, selected["manager"]): + raise ValueError( + "Required packages are still missing; installation stopped." + ) class HTTPSOnlyRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, request, file, code, message, headers, newurl): - if urllib.parse.urlsplit(newurl).scheme != 'https': - raise ValueError('Refusing a download redirect outside HTTPS') - return super().redirect_request(request, file, code, message, headers, newurl) + if urllib.parse.urlsplit(newurl).scheme != "https": + raise ValueError("Refusing a download redirect outside HTTPS") + return super().redirect_request( + request, file, code, message, headers, newurl + ) def download(url, destination, limit): - if urllib.parse.urlsplit(url).scheme != 'https': - raise ValueError('Downloads require HTTPS') + if urllib.parse.urlsplit(url).scheme != "https": + raise ValueError("Downloads require HTTPS") deadline = time.monotonic() + 300 opener = urllib.request.build_opener(HTTPSOnlyRedirect()) - request = urllib.request.Request(url, headers={'User-Agent': 'Decksmith-verified-installer'}) - with opener.open(request, timeout=30) as incoming, destination.open('xb') as outgoing: - if urllib.parse.urlsplit(incoming.geturl()).scheme != 'https': - raise ValueError('Refusing non-HTTPS download') + request = urllib.request.Request( + url, headers={"User-Agent": "Decksmith-verified-installer"} + ) + with opener.open(request, timeout=30) as incoming, destination.open( + "xb" + ) as outgoing: + if urllib.parse.urlsplit(incoming.geturl()).scheme != "https": + raise ValueError("Refusing non-HTTPS download") size = 0 while block := incoming.read(1024 * 1024): if time.monotonic() > deadline: - raise ValueError('Download timed out: ' + destination.name) + raise ValueError("Download timed out: " + destination.name) size += len(block) if size > limit: - raise ValueError('Download exceeds its size limit: ' + destination.name) + raise ValueError( + "Download exceeds its size limit: " + destination.name + ) outgoing.write(block) +def source_packages(release): + tools = [ + package + for command, package in BUILD_TOOLS.items() + if not shutil.which(command) + ] + return tuple(dict.fromkeys((*runtime_packages(release), *tools))) + + +def source_install(packages, *, no_deps=False, stage_root=None): + """Clone, build and install main; the per-user installer never activates.""" + print( + "Decksmith: building " + + SOURCE_REPOSITORY + + " (main) for " + + ARCH["label"], + flush=True, + ) + print(PRESERVED, flush=True) + install_dependencies(packages, ARCH, no_deps=no_deps) + with tempfile.TemporaryDirectory(prefix="decksmith-source-") as temporary: + checkout, output = ( + Path(temporary) / "decksmith", + Path(temporary) / "dist", + ) + subprocess.run( + ["git", "clone", "--depth", "1", SOURCE_REPOSITORY, str(checkout)], + check=True, + ) + subprocess.run( + [ + sys.executable, + "-B", + str(checkout / "scripts/build-bundle.py"), + "--output", + str(output), + ], + check=True, + ) + bundle = next(output.glob("decksmith-*.tar.gz")) + command = [sys.executable, "-B", str(output / "decksmith-install.py")] + if stage_root is not None: + command += ["--stage-root", str(Path(stage_root).absolute())] + subprocess.run( + [*command, "install", str(bundle)], check=True, timeout=120 + ) + print( + "Open Decksmith from your application menu. Close other Stream Deck " + "controllers, then start Background controls when ready. For USB " + "permissions, see the installation guide." + ) + return { + "installed": True, + "staged": stage_root is not None, + "source": SOURCE_REPOSITORY, + } + + def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument('--experimental', action='store_true', help='Opt into an experimental platform') - parser.add_argument('--check', action='store_true', help='Show OS, release and dependency status; no network or changes') - parser.add_argument('--allow-portable-gh', action='store_true', help='Allow the pinned temporary upstream GitHub CLI when the system CLI is absent or too old') - parser.add_argument('--no-deps', action='store_true', help='Stop if dependencies are missing instead of offering to install them') - parser.add_argument('--verify-only', action='store_true', help='Download and verify without installing Decksmith') - parser.add_argument('--stage-root', type=Path, help='Install into an isolated test directory; never manage services') + parser.add_argument( + "--experimental", + action="store_true", + help="Opt into an experimental platform", + ) + parser.add_argument( + "--check", + action="store_true", + help="Show OS, release and dependency status; no network or changes", + ) + parser.add_argument( + "--allow-portable-gh", + action="store_true", + help="Allow the pinned temporary upstream GitHub CLI when the system " + "CLI is absent or too old", + ) + parser.add_argument( + "--no-deps", + action="store_true", + help="Stop if dependencies are missing instead of offering to install " + "them", + ) + parser.add_argument( + "--verify-only", + action="store_true", + help="Download and verify without installing Decksmith", + ) + parser.add_argument( + "--stage-root", + type=Path, + help="Install into an isolated test directory; never manage services", + ) args = parser.parse_args(argv) if os.geteuid() == 0: - raise ValueError('Run as your regular desktop user, not with sudo.') - if platform.system() != 'Linux' or platform.machine() != 'x86_64': - raise ValueError('Published downloads require Linux x86_64.') + raise ValueError("Run as your regular desktop user, not with sudo.") + if platform.system() != "Linux" or platform.machine() != "x86_64": + raise ValueError("Published downloads require Linux x86_64.") if sys.version_info < (3, 11): - raise ValueError('Python 3.11 or newer is required for verification.') + raise ValueError("Python 3.11 or newer is required for verification.") release = platform.freedesktop_os_release() selected = select_release(release, experimental=args.experimental) if args.verify_only and args.stage_root: - raise ValueError('--stage-root cannot be combined with --verify-only') + raise ValueError("--stage-root cannot be combined with --verify-only") + if selected is ARCH: + packages = source_packages(release) + if args.check: + report = { + "platform": ARCH["label"], + "source": SOURCE_REPOSITORY, + "dependencies": list(packages), + "missing_dependencies": missing_packages(packages, "pacman"), + "startup": "unchanged; controls not started", + } + print(json.dumps(report, indent=2)) + return report + if args.verify_only: + raise ValueError( + "Arch builds from source; there is no signed download to " + "verify." + ) + return source_install( + packages, no_deps=args.no_deps, stage_root=args.stage_root + ) verifier = verifier_module() - config = verifier.target_config(selected['target']) + config = verifier.target_config(selected["target"]) packages = runtime_packages(release) - report = {'platform': selected['label'], 'experimental': selected['experimental'], - 'release': selected['version'], 'source_commit': selected['source'], - 'archive': config[2][0], 'dependencies': list(packages), - 'missing_dependencies': missing_packages(packages, selected['manager']), - 'verification_requirement': 'GitHub CLI 2.102.0+ or optional pinned temporary copy', - 'startup': 'unchanged; controls not started'} + report = { + "platform": selected["label"], + "experimental": selected["experimental"], + "release": selected["version"], + "source_commit": selected["source"], + "archive": config[2][0], + "dependencies": list(packages), + "missing_dependencies": missing_packages(packages, selected["manager"]), + "verification_requirement": ( + "GitHub CLI 2.102.0+ or optional pinned temporary copy" + ), + "startup": "unchanged; controls not started", + } if args.check: print(json.dumps(report, indent=2)) return report - print('Decksmith: ' + selected['version'] + ' for ' + selected['label'], flush=True) - if selected['experimental']: - print('Experimental preview: physical hardware/audio and lifecycle support are not fully qualified.', flush=True) - print('Saved data and startup preferences are preserved; background controls will not restart.', flush=True) + print( + "Decksmith: " + selected["version"] + " for " + selected["label"], + flush=True, + ) + if selected["experimental"]: + print( + "Experimental preview: physical hardware/audio and lifecycle " + "support are not fully qualified.", + flush=True, + ) + print(PRESERVED, flush=True) install_dependencies(packages, selected, no_deps=args.no_deps) - with tempfile.TemporaryDirectory(prefix='decksmith-downloads-') as temporary: + with tempfile.TemporaryDirectory( + prefix="decksmith-downloads-" + ) as temporary: directory = Path(temporary) os.chmod(directory, 0o700) - gh_command = verification_cli(directory, allow_portable=args.allow_portable_gh) - base = 'https://github.com/infamous-pattern/decksmith/releases/download/' + selected['version'] + '/' - for name in (*config[3], 'attestation.json'): - print('Downloading ' + name, flush=True) - download(base + name, directory / name, config[4].get(name, 8 * 1024 * 1024)) - result = verifier.verify_install(directory, selected['source'], - bundle=directory / 'attestation.json', - target=selected['target'], install=not args.verify_only, - stage_root=args.stage_root, gh_command=gh_command) + gh_command = verification_cli( + directory, allow_portable=args.allow_portable_gh + ) + base = ( + "https://github.com/infamous-pattern/decksmith/releases/download/" + + selected["version"] + + "/" + ) + for name in (*config[3], "attestation.json"): + print("Downloading " + name, flush=True) + download( + base + name, + directory / name, + config[4].get(name, 8 * 1024 * 1024), + ) + result = verifier.verify_install( + directory, + selected["source"], + bundle=directory / "attestation.json", + target=selected["target"], + install=not args.verify_only, + stage_root=args.stage_root, + gh_command=gh_command, + ) print(json.dumps(result, indent=2)) - if result['installed']: - print('Open Decksmith from your application menu. Close other Stream Deck controllers, ' - 'then start Background controls when ready. For USB permissions or the optional ' - 'panel icon, see the installation guide.') + if result["installed"]: + print( + "Open Decksmith from your application menu. Close other Stream " + "Deck controllers, then start Background controls when ready. For " + "USB permissions or the optional panel icon, see the installation " + "guide." + ) return result -if __name__ == '__main__': +if __name__ == "__main__": try: main() except (OSError, ValueError, subprocess.SubprocessError) as error: - print('Installation stopped: ' + str(error), file=sys.stderr) + print("Installation stopped: " + str(error), file=sys.stderr) sys.exit(1) diff --git a/scripts/build-bootstrap.py b/scripts/build-bootstrap.py index 837976c..1e0a3d8 100644 --- a/scripts/build-bootstrap.py +++ b/scripts/build-bootstrap.py @@ -1,12 +1,14 @@ #!/usr/bin/env python3 """Generate the single-download shell installer from reviewed Python helpers.""" + from pathlib import Path import sys ROOT = Path(__file__).resolve().parent -FILES = ('runtime_dependencies.py', 'verify-install.py', 'bootstrap-install.py') -HEADER = '''#!/bin/sh -# Generated by scripts/build-bootstrap.py. Initial trust: reviewed GitHub HTTPS source. +FILES = ("runtime_dependencies.py", "verify-install.py", "bootstrap-install.py") +HEADER = """#!/bin/sh +# Generated by scripts/build-bootstrap.py. +# Initial trust: reviewed GitHub HTTPS source. # Signed release downloads are authenticated before their code can execute. set -eu if [ "$(id -u)" -eq 0 ]; then @@ -15,8 +17,10 @@ if ! command -v python3 >/dev/null 2>&1; then echo 'Python 3.11+ is needed to verify downloads.' >&2 case "$(. /etc/os-release; echo "${ID:-}")" in - fedora) echo 'Install it with sudo dnf install python3, then retry.' >&2;; - debian|ubuntu|pop) echo 'Install it with sudo apt-get install python3, then retry.' >&2;; + fedora) echo 'Run sudo dnf install python3, then retry.' >&2;; + debian|ubuntu|pop) + echo 'Run sudo apt-get install python3, then retry.' >&2;; + arch|omarchy) echo 'Run sudo pacman -S python, then retry.' >&2;; esac exit 1 fi @@ -25,25 +29,37 @@ trap 'rm -rf -- "$temporary"' EXIT trap 'exit 130' INT trap 'exit 143' TERM HUP -''' +""" def render(): content = HEADER for number, name in enumerate(FILES): - marker = 'DECKSMITH_SOURCE_' + str(number) + marker = "DECKSMITH_SOURCE_" + str(number) source = (ROOT / name).read_text() if marker in source.splitlines(): - raise ValueError('Heredoc marker occurs in source') - content += 'cat > "$temporary/' + name + '" <<\'' + marker + "'\n" + source + '\n' + marker + '\n' + raise ValueError("Heredoc marker occurs in source") + content += ( + 'cat > "$temporary/' + + name + + "\" <<'" + + marker + + "'\n" + + source + + "\n" + + marker + + "\n" + ) return content + 'python3 -I -B "$temporary/bootstrap-install.py" "$@"\n' -if __name__ == '__main__': +if __name__ == "__main__": expected = render() - destination = ROOT / 'install.sh' - if '--check' in sys.argv: + destination = ROOT / "install.sh" + if "--check" in sys.argv: if destination.read_text() != expected: - raise SystemExit('Installer is stale. Run python3 scripts/build-bootstrap.py.') + raise SystemExit( + "Installer is stale. Run python3 scripts/build-bootstrap.py." + ) else: destination.write_text(expected) diff --git a/scripts/build_platform.py b/scripts/build_platform.py index 3d3b230..c67aaaf 100644 --- a/scripts/build_platform.py +++ b/scripts/build_platform.py @@ -1,26 +1,36 @@ """Record the native bundle environment without claiming release acceptance.""" + import platform def bundle_platform(): - if platform.system() != 'Linux': - raise ValueError('Decksmith bundles must be built on Linux.') + if platform.system() != "Linux": + raise ValueError("Decksmith bundles must be built on Linux.") release = platform.freedesktop_os_release() - distribution = release.get('ID', '') - version = release.get('VERSION_ID', '') + distribution = release.get("ID", "") + # Rolling releases such as Arch have no VERSION_ID, only BUILD_ID=rolling. + version = release.get("VERSION_ID") or release.get("BUILD_ID", "") if not distribution or not version: - raise ValueError('The build environment must identify its distribution and version.') - names = {'fedora': 'Fedora', 'debian': 'Debian', 'ubuntu': 'Ubuntu'} - label = names.get(distribution, release.get('NAME', distribution)) + ' ' + version + raise ValueError( + "The build environment must identify its distribution and version." + ) + names = {"fedora": "Fedora", "debian": "Debian", "ubuntu": "Ubuntu"} + label = ( + names.get(distribution, release.get("NAME", distribution)) + + " " + + version + ) libc, libc_version = platform.libc_ver() return { - 'platform': 'Linux', - 'architecture': platform.machine(), + "platform": "Linux", + "architecture": platform.machine(), # Existing candidate admission checks this field. It identifies the # native target; qualification evidence remains a separate requirement. - 'tested_distribution': label, - 'build_environment': { - 'distribution': distribution, 'version': version, - 'libc': libc, 'libc_version': libc_version, + "tested_distribution": label, + "build_environment": { + "distribution": distribution, + "version": version, + "libc": libc, + "libc_version": libc_version, }, } diff --git a/scripts/install.sh b/scripts/install.sh index 6f33c9e..06ef60e 100755 --- a/scripts/install.sh +++ b/scripts/install.sh @@ -1,5 +1,6 @@ #!/bin/sh -# Generated by scripts/build-bootstrap.py. Initial trust: reviewed GitHub HTTPS source. +# Generated by scripts/build-bootstrap.py. +# Initial trust: reviewed GitHub HTTPS source. # Signed release downloads are authenticated before their code can execute. set -eu if [ "$(id -u)" -eq 0 ]; then @@ -8,8 +9,10 @@ fi if ! command -v python3 >/dev/null 2>&1; then echo 'Python 3.11+ is needed to verify downloads.' >&2 case "$(. /etc/os-release; echo "${ID:-}")" in - fedora) echo 'Install it with sudo dnf install python3, then retry.' >&2;; - debian|ubuntu|pop) echo 'Install it with sudo apt-get install python3, then retry.' >&2;; + fedora) echo 'Run sudo dnf install python3, then retry.' >&2;; + debian|ubuntu|pop) + echo 'Run sudo apt-get install python3, then retry.' >&2;; + arch|omarchy) echo 'Run sudo pacman -S python, then retry.' >&2;; esac exit 1 fi @@ -22,40 +25,95 @@ cat > "$temporary/runtime_dependencies.py" <<'DECKSMITH_SOURCE_0' """Runtime package names shared by installation and read-only diagnostics.""" FEDORA = { - 'python3': ('python3',), 'systemctl': ('systemd',), 'gdbus': ('glib2',), - 'pactl': ('pulseaudio-utils',), 'wpctl': ('wireplumber',), - 'artwork': ('gtk4', 'libadwaita', 'python3-gobject', 'python3-pillow', - 'python3-cairo', 'librsvg2'), - 'libpulse.so.0': ('pulseaudio-libs',), 'libudev.so.1': ('systemd-libs',), + "python3": ("python3",), + "systemctl": ("systemd",), + "gdbus": ("glib2",), + "pactl": ("pulseaudio-utils",), + "wpctl": ("wireplumber",), + "artwork": ( + "gtk4", + "libadwaita", + "python3-gobject", + "python3-pillow", + "python3-cairo", + "librsvg2", + ), + "libpulse.so.0": ("pulseaudio-libs",), + "libudev.so.1": ("systemd-libs",), } DEBIAN = { - 'python3': ('python3',), 'systemctl': ('systemd',), 'gdbus': ('libglib2.0-bin',), - 'pactl': ('pulseaudio-utils',), 'wpctl': ('wireplumber',), - 'artwork': ('gir1.2-gtk-4.0', 'gir1.2-adw-1', 'gir1.2-rsvg-2.0', - 'python3-gi', 'python3-gi-cairo', 'python3-pil', 'python3-cairo'), - 'libpulse.so.0': ('libpulse0',), 'libudev.so.1': ('libudev1',), + "python3": ("python3",), + "systemctl": ("systemd",), + "gdbus": ("libglib2.0-bin",), + "pactl": ("pulseaudio-utils",), + "wpctl": ("wireplumber",), + "artwork": ( + "gir1.2-gtk-4.0", + "gir1.2-adw-1", + "gir1.2-rsvg-2.0", + "python3-gi", + "python3-gi-cairo", + "python3-pil", + "python3-cairo", + ), + "libpulse.so.0": ("libpulse0",), + "libudev.so.1": ("libudev1",), +} +ARCH = { + "python3": ("python",), + "systemctl": ("systemd",), + "gdbus": ("glib2",), + "pactl": ("libpulse",), + "wpctl": ("wireplumber",), + "artwork": ( + "gtk4", + "libadwaita", + "python-gobject", + "python-pillow", + "python-cairo", + "librsvg", + ), + "libpulse.so.0": ("libpulse",), + "libudev.so.1": ("systemd-libs",), } def package_groups(release): - ids = {release.get('ID', ''), *release.get('ID_LIKE', '').split()} - if ids & {'debian', 'ubuntu'}: + ids = {release.get("ID", ""), *release.get("ID_LIKE", "").split()} + if ids & {"debian", "ubuntu"}: return DEBIAN - if 'fedora' in ids: + if "fedora" in ids: return FEDORA + if "arch" in ids: + return ARCH return {} def runtime_packages(release): - return tuple(dict.fromkeys(package for group in package_groups(release).values() - for package in group)) + return tuple( + dict.fromkeys( + package + for group in package_groups(release).values() + for package in group + ) + ) def package_hints(release): groups = package_groups(release) - return {name: ', '.join(groups.get(name, (name + ' (see distribution guide)',))) - for name in ('python3', 'systemctl', 'gdbus', 'pactl', 'wpctl', - 'artwork', 'libpulse.so.0', 'libudev.so.1')} + return { + name: ", ".join(groups.get(name, (name + " (see distribution guide)",))) + for name in ( + "python3", + "systemctl", + "gdbus", + "pactl", + "wpctl", + "artwork", + "libpulse.so.0", + "libudev.so.1", + ) + } DECKSMITH_SOURCE_0 cat > "$temporary/verify-install.py" <<'DECKSMITH_SOURCE_1' @@ -242,7 +300,11 @@ if __name__ == '__main__': DECKSMITH_SOURCE_1 cat > "$temporary/bootstrap-install.py" <<'DECKSMITH_SOURCE_2' #!/usr/bin/env python3 -"""Select, download and authenticate a published Decksmith package before installation.""" +"""Select, download and authenticate a published Decksmith package. + +Authentication happens before installation. +""" + import argparse import hashlib import importlib.util @@ -267,108 +329,212 @@ sys.path.insert(0, str(Path(__file__).resolve().parent)) from runtime_dependencies import runtime_packages RELEASES = { - ('fedora', '44'): { - 'target': 'fedora44', 'label': 'Fedora Workstation 44 GNOME/Wayland', - 'version': 'v1.5.1', 'source': '4891555b388f1806ca0cda909d77af6dbd77a8ee', - 'manager': 'dnf', 'experimental': False, + ("fedora", "44"): { + "target": "fedora44", + "label": "Fedora Workstation 44 GNOME/Wayland", + "version": "v1.5.1", + "source": "4891555b388f1806ca0cda909d77af6dbd77a8ee", + "manager": "dnf", + "experimental": False, }, - ('fedora', '45'): { - 'target': 'fedora44', 'label': 'Fedora Workstation 45 GNOME/Wayland (Fedora 44 bundle)', - 'version': 'v1.5.1', 'source': '4891555b388f1806ca0cda909d77af6dbd77a8ee', - 'manager': 'dnf', 'experimental': True, + ("fedora", "45"): { + "target": "fedora44", + "label": "Fedora Workstation 45 GNOME/Wayland (Fedora 44 bundle)", + "version": "v1.5.1", + "source": "4891555b388f1806ca0cda909d77af6dbd77a8ee", + "manager": "dnf", + "experimental": True, }, - ('debian', '13'): { - 'target': 'debian13', 'label': 'Debian 13 GNOME/Wayland', - 'version': 'v1.0.0-debian-preview.1', 'source': 'dee2f2c285041dc19388ed579b90886e4b58341a', - 'manager': 'apt-get', 'experimental': True, + ("debian", "13"): { + "target": "debian13", + "label": "Debian 13 GNOME/Wayland", + "version": "v1.0.0-debian-preview.1", + "source": "dee2f2c285041dc19388ed579b90886e4b58341a", + "manager": "apt-get", + "experimental": True, }, - ('ubuntu', '26.04'): { - 'target': 'ubuntu2604', 'label': 'Ubuntu 26.04 GNOME/Wayland', - 'version': 'v1.0.0-ubuntu-preview.1', 'source': 'e627ea4f7f660a6ff6ad4772a013989b36b717c4', - 'manager': 'apt-get', 'experimental': True, + ("ubuntu", "26.04"): { + "target": "ubuntu2604", + "label": "Ubuntu 26.04 GNOME/Wayland", + "version": "v1.0.0-ubuntu-preview.1", + "source": "e627ea4f7f660a6ff6ad4772a013989b36b717c4", + "manager": "apt-get", + "experimental": True, }, - ('pop', '24.04'): { - 'target': 'popos2404', 'label': 'Pop!_OS 24.04 COSMIC', - 'version': 'v1.0.0-popos-preview.1', 'source': 'be238905bb3f6aa793dcda2df063fa5a2f206cbc', - 'manager': 'apt-get', 'experimental': True, + ("pop", "24.04"): { + "target": "popos2404", + "label": "Pop!_OS 24.04 COSMIC", + "version": "v1.0.0-popos-preview.1", + "source": "be238905bb3f6aa793dcda2df063fa5a2f206cbc", + "manager": "apt-get", + "experimental": True, }, } +# Arch has no signed bundle: build the main branch on the machine. +SOURCE_REPOSITORY = "https://github.com/infamous-pattern/decksmith.git" +# Command -> package, skipped when the command exists (e.g. rustup's cargo). +BUILD_TOOLS = { + "git": "git", + "cargo": "rust", + "cc": "gcc", + "pkgconf": "pkgconf", + "strip": "binutils", +} +ARCH = { + "target": "arch", + "label": "Arch Linux (source build)", + "version": "main", + "source": SOURCE_REPOSITORY, + "manager": "pacman", + "experimental": False, +} + # Upstream's immutable v2.102.0 release fixes workflow identity verification. -# Digest taken from cli/cli's official gh_2.102.0_checksums.txt, reviewed Oct 11. -GH_VERSION = '2.102.0' -GH_ARCHIVE = 'gh_' + GH_VERSION + '_linux_amd64.tar.gz' -GH_SHA256 = 'bb766f710eef8ede859c18578c72c327597cd4c8a85b06001b1f3843c6019386' +# Digest taken from cli/cli's official gh_2.102.0_checksums.txt, +# reviewed Oct 11. +GH_VERSION = "2.102.0" +GH_ARCHIVE = "gh_" + GH_VERSION + "_linux_amd64.tar.gz" +GH_SHA256 = "bb766f710eef8ede859c18578c72c327597cd4c8a85b06001b1f3843c6019386" + + +PRESERVED = ( + "Saved data and startup preferences are preserved; background controls " + "will not restart." +) def supports_attestation(command): try: - version = subprocess.run([command, '--version'], capture_output=True, text=True, timeout=15) - match = re.search(r'^gh version (\d+)\.(\d+)\.(\d+)\b', version.stdout or '', re.MULTILINE) - if version.returncode or not match or tuple(map(int, match.groups())) < (2, 102, 0): + version = subprocess.run( + [command, "--version"], capture_output=True, text=True, timeout=15 + ) + match = re.search( + r"^gh version (\d+)\.(\d+)\.(\d+)\b", + version.stdout or "", + re.MULTILINE, + ) + if ( + version.returncode + or not match + or tuple(map(int, match.groups())) < (2, 102, 0) + ): return False - return subprocess.run([command, 'attestation', 'verify', '--help'], - capture_output=True, text=True, timeout=15).returncode == 0 + return ( + subprocess.run( + [command, "attestation", "verify", "--help"], + capture_output=True, + text=True, + timeout=15, + ).returncode + == 0 + ) except OSError: return False def verification_cli(directory, *, allow_portable=False): - if supports_attestation('gh'): - return 'gh' - message = ('GitHub CLI 2.102.0+ is required for attestation verification. ' - 'Use its official package source, or allow a pinned temporary upstream copy.') + if supports_attestation("gh"): + return "gh" + message = ( + "GitHub CLI 2.102.0+ is required for attestation verification. " + "Use its official package source, or allow a pinned temporary " + "upstream copy." + ) if not allow_portable: try: # A controlling terminal is not seekable: buffered r+ creates a # BufferedRandom stream and fails before the consent prompt appears. - with open('/dev/tty', 'r', encoding='utf-8') as terminal, \ - open('/dev/tty', 'w', encoding='utf-8') as display: - display.write(message + '\nDownload GitHub CLI ' + GH_VERSION + - ' temporarily for this installation? [y/N] ') + with open("/dev/tty", "r", encoding="utf-8") as terminal, open( + "/dev/tty", "w", encoding="utf-8" + ) as display: + display.write( + message + + "\nDownload GitHub CLI " + + GH_VERSION + + " temporarily for this installation? [y/N] " + ) display.flush() - allow_portable = terminal.readline().strip().lower() in ('y', 'yes') + allow_portable = terminal.readline().strip().lower() in ( + "y", + "yes", + ) except OSError as error: - raise ValueError(message + ' Retry with --allow-portable-gh to opt in.') from error + raise ValueError( + message + " Retry with --allow-portable-gh to opt in." + ) from error if not allow_portable: - raise ValueError(message + ' Installation stopped; no signatures were bypassed.') + raise ValueError( + message + " Installation stopped; no signatures were bypassed." + ) archive = directory / GH_ARCHIVE - print('Downloading pinned GitHub CLI ' + GH_VERSION + ' for temporary verification.', flush=True) - download('https://github.com/cli/cli/releases/download/v' + GH_VERSION + '/' + GH_ARCHIVE, - archive, 64 * 1024 * 1024) - with archive.open('rb') as incoming: - if hashlib.file_digest(incoming, 'sha256').hexdigest() != GH_SHA256: - raise ValueError('GitHub CLI checksum differs from the reviewed pin; refusing execution.') - # Read just the pinned archive's regular executable, never extract its paths. + print( + "Downloading pinned GitHub CLI " + + GH_VERSION + + " for temporary verification.", + flush=True, + ) + download( + "https://github.com/cli/cli/releases/download/v" + + GH_VERSION + + "/" + + GH_ARCHIVE, + archive, + 64 * 1024 * 1024, + ) + with archive.open("rb") as incoming: + if hashlib.file_digest(incoming, "sha256").hexdigest() != GH_SHA256: + raise ValueError( + "GitHub CLI checksum differs from the reviewed pin; " + "refusing execution." + ) + # Read just the pinned archive's regular executable, never extract its + # paths. with tarfile.open(archive) as package: - member = package.getmember('gh_' + GH_VERSION + '_linux_amd64/bin/gh') + member = package.getmember("gh_" + GH_VERSION + "_linux_amd64/bin/gh") if not member.isfile() or member.size > 64 * 1024 * 1024: - raise ValueError('Invalid GitHub CLI executable') + raise ValueError("Invalid GitHub CLI executable") with package.extractfile(member) as incoming: - executable = directory / 'gh' - with executable.open('xb') as outgoing: + executable = directory / "gh" + with executable.open("xb") as outgoing: shutil.copyfileobj(incoming, outgoing) executable.chmod(0o700) if not supports_attestation(str(executable)): - raise ValueError('Pinned GitHub CLI is unavailable; refusing installation.') + raise ValueError( + "Pinned GitHub CLI is unavailable; refusing installation." + ) return str(executable) +def arch_family(release): + return "arch" in { + release.get("ID", ""), + *release.get("ID_LIKE", "").split(), + } + + def select_release(release, *, experimental=False): - key = (release.get('ID'), release.get('VERSION_ID')) + if arch_family(release): + return ARCH + key = (release.get("ID"), release.get("VERSION_ID")) if key not in RELEASES: - raise ValueError('No verified download is configured for this OS/version. ' - 'See the support table for available platforms.') + raise ValueError( + "No verified download is configured for this OS/version. " + "See the support table for available platforms." + ) selected = RELEASES[key] - if selected['experimental'] and not experimental: - raise ValueError(selected['label'] + ' is experimental. Read its preview limits, ' - 'then use --experimental to opt in.') + if selected["experimental"] and not experimental: + raise ValueError( + selected["label"] + " is experimental. Read its preview limits, " + "then use --experimental to opt in." + ) return selected def verifier_module(): spec = importlib.util.spec_from_file_location( - 'reviewed_verifier', Path(__file__).with_name('verify-install.py')) + "reviewed_verifier", Path(__file__).with_name("verify-install.py") + ) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module @@ -377,123 +543,295 @@ def verifier_module(): def missing_packages(packages, manager): missing = [] for package in packages: - command = (['rpm', '-q', package] if manager == 'dnf' else - ['dpkg-query', '-W', '-f=${Status}', package]) - result = subprocess.run(command, capture_output=True, text=True, timeout=15) - if result.returncode or (manager != 'dnf' and result.stdout.strip() != 'install ok installed'): + command = ( + ["rpm", "-q", package] + if manager == "dnf" + else ( + ["pacman", "-Q", package] + if manager == "pacman" + else ["dpkg-query", "-W", "-f=${Status}", package] + ) + ) + result = subprocess.run( + command, capture_output=True, text=True, timeout=15 + ) + if result.returncode or ( + manager == "apt-get" + and result.stdout.strip() != "install ok installed" + ): missing.append(package) return missing def install_dependencies(packages, selected, *, no_deps=False): - missing = missing_packages(packages, selected['manager']) + missing = missing_packages(packages, selected["manager"]) if not missing: return - command = ['sudo', selected['manager'], 'install', *missing] - print('Missing dependencies: ' + ', '.join(missing), flush=True) + command = ( + ["sudo", "pacman", "-S", "--needed", *missing] + if selected["manager"] == "pacman" + else ["sudo", selected["manager"], "install", *missing] + ) + print("Missing dependencies: " + ", ".join(missing), flush=True) if no_deps: - raise ValueError('Install these packages from your OS repositories, then retry: ' - + ' '.join(command)) - if not shutil.which('sudo'): - raise ValueError('sudo is unavailable. Ask an administrator to install: ' + ', '.join(missing)) + raise ValueError( + "Install these packages from your OS repositories, then retry: " + + " ".join(command) + ) + if not shutil.which("sudo"): + raise ValueError( + "sudo is unavailable. Ask an administrator to install: " + + ", ".join(missing) + ) try: - terminal = open('/dev/tty', 'r', encoding='utf-8') + terminal = open("/dev/tty", "r", encoding="utf-8") except OSError as error: - raise ValueError('Dependency installation needs a terminal. Run: ' + ' '.join(command)) from error + raise ValueError( + "Dependency installation needs a terminal. Run: " + + " ".join(command) + ) from error # Trusted OS package sources, normal package-manager confirmation; never -y. with terminal: subprocess.run(command, stdin=terminal, check=True) - if missing_packages(packages, selected['manager']): - raise ValueError('Required packages are still missing; installation stopped.') + if missing_packages(packages, selected["manager"]): + raise ValueError( + "Required packages are still missing; installation stopped." + ) class HTTPSOnlyRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, request, file, code, message, headers, newurl): - if urllib.parse.urlsplit(newurl).scheme != 'https': - raise ValueError('Refusing a download redirect outside HTTPS') - return super().redirect_request(request, file, code, message, headers, newurl) + if urllib.parse.urlsplit(newurl).scheme != "https": + raise ValueError("Refusing a download redirect outside HTTPS") + return super().redirect_request( + request, file, code, message, headers, newurl + ) def download(url, destination, limit): - if urllib.parse.urlsplit(url).scheme != 'https': - raise ValueError('Downloads require HTTPS') + if urllib.parse.urlsplit(url).scheme != "https": + raise ValueError("Downloads require HTTPS") deadline = time.monotonic() + 300 opener = urllib.request.build_opener(HTTPSOnlyRedirect()) - request = urllib.request.Request(url, headers={'User-Agent': 'Decksmith-verified-installer'}) - with opener.open(request, timeout=30) as incoming, destination.open('xb') as outgoing: - if urllib.parse.urlsplit(incoming.geturl()).scheme != 'https': - raise ValueError('Refusing non-HTTPS download') + request = urllib.request.Request( + url, headers={"User-Agent": "Decksmith-verified-installer"} + ) + with opener.open(request, timeout=30) as incoming, destination.open( + "xb" + ) as outgoing: + if urllib.parse.urlsplit(incoming.geturl()).scheme != "https": + raise ValueError("Refusing non-HTTPS download") size = 0 while block := incoming.read(1024 * 1024): if time.monotonic() > deadline: - raise ValueError('Download timed out: ' + destination.name) + raise ValueError("Download timed out: " + destination.name) size += len(block) if size > limit: - raise ValueError('Download exceeds its size limit: ' + destination.name) + raise ValueError( + "Download exceeds its size limit: " + destination.name + ) outgoing.write(block) +def source_packages(release): + tools = [ + package + for command, package in BUILD_TOOLS.items() + if not shutil.which(command) + ] + return tuple(dict.fromkeys((*runtime_packages(release), *tools))) + + +def source_install(packages, *, no_deps=False, stage_root=None): + """Clone, build and install main; the per-user installer never activates.""" + print( + "Decksmith: building " + + SOURCE_REPOSITORY + + " (main) for " + + ARCH["label"], + flush=True, + ) + print(PRESERVED, flush=True) + install_dependencies(packages, ARCH, no_deps=no_deps) + with tempfile.TemporaryDirectory(prefix="decksmith-source-") as temporary: + checkout, output = ( + Path(temporary) / "decksmith", + Path(temporary) / "dist", + ) + subprocess.run( + ["git", "clone", "--depth", "1", SOURCE_REPOSITORY, str(checkout)], + check=True, + ) + subprocess.run( + [ + sys.executable, + "-B", + str(checkout / "scripts/build-bundle.py"), + "--output", + str(output), + ], + check=True, + ) + bundle = next(output.glob("decksmith-*.tar.gz")) + command = [sys.executable, "-B", str(output / "decksmith-install.py")] + if stage_root is not None: + command += ["--stage-root", str(Path(stage_root).absolute())] + subprocess.run( + [*command, "install", str(bundle)], check=True, timeout=120 + ) + print( + "Open Decksmith from your application menu. Close other Stream Deck " + "controllers, then start Background controls when ready. For USB " + "permissions, see the installation guide." + ) + return { + "installed": True, + "staged": stage_root is not None, + "source": SOURCE_REPOSITORY, + } + + def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument('--experimental', action='store_true', help='Opt into an experimental platform') - parser.add_argument('--check', action='store_true', help='Show OS, release and dependency status; no network or changes') - parser.add_argument('--allow-portable-gh', action='store_true', help='Allow the pinned temporary upstream GitHub CLI when the system CLI is absent or too old') - parser.add_argument('--no-deps', action='store_true', help='Stop if dependencies are missing instead of offering to install them') - parser.add_argument('--verify-only', action='store_true', help='Download and verify without installing Decksmith') - parser.add_argument('--stage-root', type=Path, help='Install into an isolated test directory; never manage services') + parser.add_argument( + "--experimental", + action="store_true", + help="Opt into an experimental platform", + ) + parser.add_argument( + "--check", + action="store_true", + help="Show OS, release and dependency status; no network or changes", + ) + parser.add_argument( + "--allow-portable-gh", + action="store_true", + help="Allow the pinned temporary upstream GitHub CLI when the system " + "CLI is absent or too old", + ) + parser.add_argument( + "--no-deps", + action="store_true", + help="Stop if dependencies are missing instead of offering to install " + "them", + ) + parser.add_argument( + "--verify-only", + action="store_true", + help="Download and verify without installing Decksmith", + ) + parser.add_argument( + "--stage-root", + type=Path, + help="Install into an isolated test directory; never manage services", + ) args = parser.parse_args(argv) if os.geteuid() == 0: - raise ValueError('Run as your regular desktop user, not with sudo.') - if platform.system() != 'Linux' or platform.machine() != 'x86_64': - raise ValueError('Published downloads require Linux x86_64.') + raise ValueError("Run as your regular desktop user, not with sudo.") + if platform.system() != "Linux" or platform.machine() != "x86_64": + raise ValueError("Published downloads require Linux x86_64.") if sys.version_info < (3, 11): - raise ValueError('Python 3.11 or newer is required for verification.') + raise ValueError("Python 3.11 or newer is required for verification.") release = platform.freedesktop_os_release() selected = select_release(release, experimental=args.experimental) if args.verify_only and args.stage_root: - raise ValueError('--stage-root cannot be combined with --verify-only') + raise ValueError("--stage-root cannot be combined with --verify-only") + if selected is ARCH: + packages = source_packages(release) + if args.check: + report = { + "platform": ARCH["label"], + "source": SOURCE_REPOSITORY, + "dependencies": list(packages), + "missing_dependencies": missing_packages(packages, "pacman"), + "startup": "unchanged; controls not started", + } + print(json.dumps(report, indent=2)) + return report + if args.verify_only: + raise ValueError( + "Arch builds from source; there is no signed download to " + "verify." + ) + return source_install( + packages, no_deps=args.no_deps, stage_root=args.stage_root + ) verifier = verifier_module() - config = verifier.target_config(selected['target']) + config = verifier.target_config(selected["target"]) packages = runtime_packages(release) - report = {'platform': selected['label'], 'experimental': selected['experimental'], - 'release': selected['version'], 'source_commit': selected['source'], - 'archive': config[2][0], 'dependencies': list(packages), - 'missing_dependencies': missing_packages(packages, selected['manager']), - 'verification_requirement': 'GitHub CLI 2.102.0+ or optional pinned temporary copy', - 'startup': 'unchanged; controls not started'} + report = { + "platform": selected["label"], + "experimental": selected["experimental"], + "release": selected["version"], + "source_commit": selected["source"], + "archive": config[2][0], + "dependencies": list(packages), + "missing_dependencies": missing_packages(packages, selected["manager"]), + "verification_requirement": ( + "GitHub CLI 2.102.0+ or optional pinned temporary copy" + ), + "startup": "unchanged; controls not started", + } if args.check: print(json.dumps(report, indent=2)) return report - print('Decksmith: ' + selected['version'] + ' for ' + selected['label'], flush=True) - if selected['experimental']: - print('Experimental preview: physical hardware/audio and lifecycle support are not fully qualified.', flush=True) - print('Saved data and startup preferences are preserved; background controls will not restart.', flush=True) + print( + "Decksmith: " + selected["version"] + " for " + selected["label"], + flush=True, + ) + if selected["experimental"]: + print( + "Experimental preview: physical hardware/audio and lifecycle " + "support are not fully qualified.", + flush=True, + ) + print(PRESERVED, flush=True) install_dependencies(packages, selected, no_deps=args.no_deps) - with tempfile.TemporaryDirectory(prefix='decksmith-downloads-') as temporary: + with tempfile.TemporaryDirectory( + prefix="decksmith-downloads-" + ) as temporary: directory = Path(temporary) os.chmod(directory, 0o700) - gh_command = verification_cli(directory, allow_portable=args.allow_portable_gh) - base = 'https://github.com/infamous-pattern/decksmith/releases/download/' + selected['version'] + '/' - for name in (*config[3], 'attestation.json'): - print('Downloading ' + name, flush=True) - download(base + name, directory / name, config[4].get(name, 8 * 1024 * 1024)) - result = verifier.verify_install(directory, selected['source'], - bundle=directory / 'attestation.json', - target=selected['target'], install=not args.verify_only, - stage_root=args.stage_root, gh_command=gh_command) + gh_command = verification_cli( + directory, allow_portable=args.allow_portable_gh + ) + base = ( + "https://github.com/infamous-pattern/decksmith/releases/download/" + + selected["version"] + + "/" + ) + for name in (*config[3], "attestation.json"): + print("Downloading " + name, flush=True) + download( + base + name, + directory / name, + config[4].get(name, 8 * 1024 * 1024), + ) + result = verifier.verify_install( + directory, + selected["source"], + bundle=directory / "attestation.json", + target=selected["target"], + install=not args.verify_only, + stage_root=args.stage_root, + gh_command=gh_command, + ) print(json.dumps(result, indent=2)) - if result['installed']: - print('Open Decksmith from your application menu. Close other Stream Deck controllers, ' - 'then start Background controls when ready. For USB permissions or the optional ' - 'panel icon, see the installation guide.') + if result["installed"]: + print( + "Open Decksmith from your application menu. Close other Stream " + "Deck controllers, then start Background controls when ready. For " + "USB permissions or the optional panel icon, see the installation " + "guide." + ) return result -if __name__ == '__main__': +if __name__ == "__main__": try: main() except (OSError, ValueError, subprocess.SubprocessError) as error: - print('Installation stopped: ' + str(error), file=sys.stderr) + print("Installation stopped: " + str(error), file=sys.stderr) sys.exit(1) DECKSMITH_SOURCE_2 diff --git a/scripts/runtime_dependencies.py b/scripts/runtime_dependencies.py index d27f880..4216041 100644 --- a/scripts/runtime_dependencies.py +++ b/scripts/runtime_dependencies.py @@ -1,37 +1,92 @@ """Runtime package names shared by installation and read-only diagnostics.""" FEDORA = { - 'python3': ('python3',), 'systemctl': ('systemd',), 'gdbus': ('glib2',), - 'pactl': ('pulseaudio-utils',), 'wpctl': ('wireplumber',), - 'artwork': ('gtk4', 'libadwaita', 'python3-gobject', 'python3-pillow', - 'python3-cairo', 'librsvg2'), - 'libpulse.so.0': ('pulseaudio-libs',), 'libudev.so.1': ('systemd-libs',), + "python3": ("python3",), + "systemctl": ("systemd",), + "gdbus": ("glib2",), + "pactl": ("pulseaudio-utils",), + "wpctl": ("wireplumber",), + "artwork": ( + "gtk4", + "libadwaita", + "python3-gobject", + "python3-pillow", + "python3-cairo", + "librsvg2", + ), + "libpulse.so.0": ("pulseaudio-libs",), + "libudev.so.1": ("systemd-libs",), } DEBIAN = { - 'python3': ('python3',), 'systemctl': ('systemd',), 'gdbus': ('libglib2.0-bin',), - 'pactl': ('pulseaudio-utils',), 'wpctl': ('wireplumber',), - 'artwork': ('gir1.2-gtk-4.0', 'gir1.2-adw-1', 'gir1.2-rsvg-2.0', - 'python3-gi', 'python3-gi-cairo', 'python3-pil', 'python3-cairo'), - 'libpulse.so.0': ('libpulse0',), 'libudev.so.1': ('libudev1',), + "python3": ("python3",), + "systemctl": ("systemd",), + "gdbus": ("libglib2.0-bin",), + "pactl": ("pulseaudio-utils",), + "wpctl": ("wireplumber",), + "artwork": ( + "gir1.2-gtk-4.0", + "gir1.2-adw-1", + "gir1.2-rsvg-2.0", + "python3-gi", + "python3-gi-cairo", + "python3-pil", + "python3-cairo", + ), + "libpulse.so.0": ("libpulse0",), + "libudev.so.1": ("libudev1",), +} +ARCH = { + "python3": ("python",), + "systemctl": ("systemd",), + "gdbus": ("glib2",), + "pactl": ("libpulse",), + "wpctl": ("wireplumber",), + "artwork": ( + "gtk4", + "libadwaita", + "python-gobject", + "python-pillow", + "python-cairo", + "librsvg", + ), + "libpulse.so.0": ("libpulse",), + "libudev.so.1": ("systemd-libs",), } def package_groups(release): - ids = {release.get('ID', ''), *release.get('ID_LIKE', '').split()} - if ids & {'debian', 'ubuntu'}: + ids = {release.get("ID", ""), *release.get("ID_LIKE", "").split()} + if ids & {"debian", "ubuntu"}: return DEBIAN - if 'fedora' in ids: + if "fedora" in ids: return FEDORA + if "arch" in ids: + return ARCH return {} def runtime_packages(release): - return tuple(dict.fromkeys(package for group in package_groups(release).values() - for package in group)) + return tuple( + dict.fromkeys( + package + for group in package_groups(release).values() + for package in group + ) + ) def package_hints(release): groups = package_groups(release) - return {name: ', '.join(groups.get(name, (name + ' (see distribution guide)',))) - for name in ('python3', 'systemctl', 'gdbus', 'pactl', 'wpctl', - 'artwork', 'libpulse.so.0', 'libudev.so.1')} + return { + name: ", ".join(groups.get(name, (name + " (see distribution guide)",))) + for name in ( + "python3", + "systemctl", + "gdbus", + "pactl", + "wpctl", + "artwork", + "libpulse.so.0", + "libudev.so.1", + ) + } diff --git a/scripts/test_bootstrap_install.py b/scripts/test_bootstrap_install.py index 6ebe16e..6975067 100644 --- a/scripts/test_bootstrap_install.py +++ b/scripts/test_bootstrap_install.py @@ -1,4 +1,5 @@ -"""Guard distro selection, dependency handling and signed bootstrap execution.""" +"""Guard distro selection, dependency handling and signed bootstrap runs.""" + import importlib.util import io import os @@ -16,231 +17,407 @@ def load(name, filename): - spec = importlib.util.spec_from_file_location(name, Path(__file__).with_name(filename)) + spec = importlib.util.spec_from_file_location( + name, Path(__file__).with_name(filename) + ) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module -BOOT = load('bootstrap', 'bootstrap-install.py') -BUILD = load('bootstrap_build', 'build-bootstrap.py') +BOOT = load("bootstrap", "bootstrap-install.py") +BUILD = load("bootstrap_build", "build-bootstrap.py") class BootstrapTests(unittest.TestCase): def terminal_child(self, program, marker, reply): - """Use a real controlling terminal: regular-file mocks miss r+ failures.""" + """Use a real controlling terminal. + + Regular-file mocks miss r+ failures. + """ import sys + pid, descriptor = pty.fork() if pid == 0: - os.execv(sys.executable, [sys.executable, '-B', '-c', program]) - output = b''; responded = False; ended = False + os.execv(sys.executable, [sys.executable, "-B", "-c", program]) + output = b"" + responded = False + ended = False try: deadline = time.monotonic() + 10 while time.monotonic() < deadline: - if select.select([descriptor], [], [], .1)[0]: + if select.select([descriptor], [], [], 0.1)[0]: try: output += os.read(descriptor, 65536) except OSError: pass if not responded and marker.encode() in output: - os.write(descriptor, reply.encode()); responded = True + os.write(descriptor, reply.encode()) + responded = True finished, status = os.waitpid(pid, os.WNOHANG) if finished: ended = True - self.assertEqual(os.waitstatus_to_exitcode(status), 0, output.decode(errors='replace')) - self.assertTrue(responded, output.decode(errors='replace')) - return output.decode(errors='replace') - self.fail('Controlling-terminal test timed out: ' + output.decode(errors='replace')) + self.assertEqual( + os.waitstatus_to_exitcode(status), + 0, + output.decode(errors="replace"), + ) + self.assertTrue(responded, output.decode(errors="replace")) + return output.decode(errors="replace") + self.fail( + "Controlling-terminal test timed out: " + + output.decode(errors="replace") + ) finally: if not ended: - os.killpg(pid, signal.SIGKILL); os.waitpid(pid, 0) + os.killpg(pid, signal.SIGKILL) + os.waitpid(pid, 0) os.close(descriptor) def terminal_loader(self): - return ('import sys; sys.path.insert(0, ' + repr(str(Path(__file__).parent.resolve())) + ')\n' - 'import importlib.util\n' - 's=importlib.util.spec_from_file_location("boot", ' + repr(str(Path(__file__).with_name('bootstrap-install.py').resolve())) + ')\n' - 'b=importlib.util.module_from_spec(s);s.loader.exec_module(b)\n') + return ( + "import sys; sys.path.insert(0, " + + repr(str(Path(__file__).parent.resolve())) + + ")\n" + "import importlib.util\n" + 's=importlib.util.spec_from_file_location("boot", ' + + repr( + str(Path(__file__).with_name("bootstrap-install.py").resolve()) + ) + + ")\n" + "b=importlib.util.module_from_spec(s);s.loader.exec_module(b)\n" + ) def test_real_tty_portable_consent_reaches_download(self): - program = self.terminal_loader() + ''' + program = self.terminal_loader() + """ from pathlib import Path b.supports_attestation=lambda command: False def download(*args):raise RuntimeError('CONSENT_ACCEPTED') b.download=download try:b.verification_cli(Path('/unused')) -except RuntimeError as e:assert str(e)=='CONSENT_ACCEPTED';print('PASS accepted',flush=True) +except RuntimeError as e: + assert str(e)=='CONSENT_ACCEPTED' + print('PASS accepted',flush=True) else:raise AssertionError('Expected intercepted download') -''' - self.assertIn('PASS accepted', self.terminal_child(program, '[y/N] ', 'y\n')) +""" + self.assertIn( + "PASS accepted", self.terminal_child(program, "[y/N] ", "y\n") + ) def test_real_tty_decline_never_downloads(self): - program = self.terminal_loader() + ''' + program = self.terminal_loader() + """ from pathlib import Path b.supports_attestation=lambda command: False def download(*args):raise AssertionError('Declined download executed') b.download=download try:b.verification_cli(Path('/unused')) -except ValueError as e:assert 'no signatures were bypassed' in str(e);print('PASS declined',flush=True) +except ValueError as e: + assert 'no signatures were bypassed' in str(e) + print('PASS declined',flush=True) else:raise AssertionError('Expected safe refusal') -''' - self.assertIn('PASS declined', self.terminal_child(program, '[y/N] ', 'n\n')) +""" + self.assertIn( + "PASS declined", self.terminal_child(program, "[y/N] ", "n\n") + ) def test_real_tty_dependency_confirmation_input(self): - program = self.terminal_loader() + ''' + program = self.terminal_loader() + """ from unittest.mock import patch checks=iter([['fixture-package'],[]]) def package_manager(command,stdin,check): assert command==['sudo','dnf','install','fixture-package'] and check print('DEPENDENCY_CONFIRMATION',flush=True) assert stdin.readline().strip()=='y' -with patch.object(b,'missing_packages',side_effect=lambda *args:next(checks)),patch.object(b.shutil,'which',return_value='/usr/bin/sudo'),patch.object(b.subprocess,'run',side_effect=package_manager): +with patch.object(b,'missing_packages',side_effect=lambda *a:next(checks)),\ + patch.object(b.shutil,'which',return_value='/usr/bin/sudo'),\ + patch.object(b.subprocess,'run',side_effect=package_manager): b.install_dependencies(('fixture-package',), b.RELEASES['fedora','44']) print('PASS terminal input',flush=True) -''' - self.assertIn('PASS terminal input', self.terminal_child(program, 'DEPENDENCY_CONFIRMATION', 'y\n')) +""" + self.assertIn( + "PASS terminal input", + self.terminal_child(program, "DEPENDENCY_CONFIRMATION", "y\n"), + ) def test_shell_contains_exact_reviewed_sources(self): - self.assertEqual(Path(__file__).with_name('install.sh').read_text(), BUILD.render()) + self.assertEqual( + Path(__file__).with_name("install.sh").read_text(), BUILD.render() + ) for number, name in enumerate(BUILD.FILES): shell = BUILD.render() - block = shell.split("<<'DECKSMITH_SOURCE_" + str(number) + "'\n", 1)[1].split( - '\nDECKSMITH_SOURCE_' + str(number), 1)[0] + block = shell.split( + "<<'DECKSMITH_SOURCE_" + str(number) + "'\n", 1 + )[1].split("\nDECKSMITH_SOURCE_" + str(number), 1)[0] self.assertEqual(block, Path(__file__).with_name(name).read_text()) def test_explicit_preview_opt_in_and_exact_os_version(self): for (distribution, version), selected in BOOT.RELEASES.items(): - release = {'ID': distribution, 'VERSION_ID': version} - self.assertEqual(BOOT.select_release(release, experimental=True), selected) - if selected['experimental']: - with self.assertRaisesRegex(ValueError, 'experimental'): + release = {"ID": distribution, "VERSION_ID": version} + self.assertEqual( + BOOT.select_release(release, experimental=True), selected + ) + if selected["experimental"]: + with self.assertRaisesRegex(ValueError, "experimental"): BOOT.select_release(release) - for release in ({'ID': 'fedora', 'VERSION_ID': '46'}, - {'ID': 'ubuntu', 'VERSION_ID': '24.04'}, - {'ID': 'opensuse-tumbleweed'}, - {'ID': 'unknown', 'ID_LIKE': 'ubuntu', 'VERSION_ID': '26.04'}): + for release in ( + {"ID": "fedora", "VERSION_ID": "46"}, + {"ID": "ubuntu", "VERSION_ID": "24.04"}, + {"ID": "opensuse-tumbleweed"}, + {"ID": "unknown", "ID_LIKE": "ubuntu", "VERSION_ID": "26.04"}, + ): with self.assertRaises(ValueError): BOOT.select_release(release, experimental=True) def test_fedora45_uses_verified_fedora44_bundle_only_with_opt_in(self): - release = {'ID': 'fedora', 'VERSION_ID': '45'} - with self.assertRaisesRegex(ValueError, 'experimental'): + release = {"ID": "fedora", "VERSION_ID": "45"} + with self.assertRaisesRegex(ValueError, "experimental"): BOOT.select_release(release) selected = BOOT.select_release(release, experimental=True) - self.assertEqual(selected['target'], 'fedora44') - self.assertEqual(selected['source'], BOOT.RELEASES['fedora', '44']['source']) - self.assertEqual(selected['version'], BOOT.RELEASES['fedora', '44']['version']) + self.assertEqual(selected["target"], "fedora44") + self.assertEqual(selected["source"], BOOT.RELEASES["fedora", "44"]["source"]) + self.assertEqual(selected["version"], BOOT.RELEASES["fedora", "44"]["version"]) - def test_approved_release_archive_workflow_and_source_pins_match_guides(self): - guides = {'fedora44': 'v1.5-verified-installation.md', 'debian13': 'debian-preview-installation.md', - 'ubuntu2604': 'ubuntu-preview-installation.md', 'popos2404': 'popos-preview-installation.md'} + def test_approved_release_archive_workflow_and_source_pins_match_guides( + self, + ): + guides = { + "fedora44": "v1.5-verified-installation.md", + "debian13": "debian-preview-installation.md", + "ubuntu2604": "ubuntu-preview-installation.md", + "popos2404": "popos-preview-installation.md", + } verifier = BOOT.verifier_module() for selected in BOOT.RELEASES.values(): - guide = (Path(__file__).parents[1] / 'docs' / guides[selected['target']]).read_text() - self.assertIn(selected['source'], guide) - self.assertIn(selected['version'], guide) - self.assertIn(verifier.target_config(selected['target'])[2][0], guide) + guide = ( + Path(__file__).parents[1] / "docs" / guides[selected["target"]] + ).read_text() + self.assertIn(selected["source"], guide) + self.assertIn(selected["version"], guide) + self.assertIn( + verifier.target_config(selected["target"])[2][0], guide + ) def test_apt_requires_installed_status_not_just_package_record(self): - results = [subprocess.CompletedProcess([], 0, 'install ok installed'), - subprocess.CompletedProcess([], 0, 'deinstall ok config-files'), - subprocess.CompletedProcess([], 1, '')] - with patch.object(BOOT.subprocess, 'run', side_effect=results): - self.assertEqual(BOOT.missing_packages(('one', 'two', 'three'), 'apt-get'), ['two', 'three']) + results = [ + subprocess.CompletedProcess([], 0, "install ok installed"), + subprocess.CompletedProcess([], 0, "deinstall ok config-files"), + subprocess.CompletedProcess([], 1, ""), + ] + with patch.object(BOOT.subprocess, "run", side_effect=results): + self.assertEqual( + BOOT.missing_packages(("one", "two", "three"), "apt-get"), + ["two", "three"], + ) def test_no_deps_never_escalates(self): - with patch.object(BOOT, 'missing_packages', return_value=['gh']), patch.object(BOOT.subprocess, 'run') as run: - with self.assertRaisesRegex(ValueError, 'sudo dnf install gh'): - BOOT.install_dependencies(('gh',), BOOT.RELEASES['fedora', '44'], no_deps=True) + with patch.object( + BOOT, "missing_packages", return_value=["gh"] + ), patch.object(BOOT.subprocess, "run") as run: + with self.assertRaisesRegex(ValueError, "sudo dnf install gh"): + BOOT.install_dependencies( + ("gh",), BOOT.RELEASES["fedora", "44"], no_deps=True + ) run.assert_not_called() def test_package_installation_uses_tty_and_normal_confirmation(self): - with tempfile.TemporaryFile(mode='w+') as terminal: - with patch.object(BOOT, 'missing_packages', side_effect=[['gh'], []]), \ - patch.object(BOOT.shutil, 'which', return_value='/usr/bin/sudo'), \ - patch('builtins.open', return_value=terminal), \ - patch.object(BOOT.subprocess, 'run') as run: - BOOT.install_dependencies(('gh',), BOOT.RELEASES['debian', '13']) - run.assert_called_once_with(['sudo', 'apt-get', 'install', 'gh'], stdin=terminal, check=True) + with tempfile.TemporaryFile(mode="w+") as terminal: + with patch.object( + BOOT, "missing_packages", side_effect=[["gh"], []] + ), patch.object( + BOOT.shutil, "which", return_value="/usr/bin/sudo" + ), patch( + "builtins.open", return_value=terminal + ), patch.object( + BOOT.subprocess, "run" + ) as run: + BOOT.install_dependencies( + ("gh",), BOOT.RELEASES["debian", "13"] + ) + run.assert_called_once_with( + ["sudo", "apt-get", "install", "gh"], + stdin=terminal, + check=True, + ) def test_http_redirect_is_rejected(self): - with self.assertRaisesRegex(ValueError, 'HTTPS'): - BOOT.HTTPSOnlyRedirect().redirect_request(None, None, 302, '', {}, 'http://example.com/file') + with self.assertRaisesRegex(ValueError, "HTTPS"): + BOOT.HTTPSOnlyRedirect().redirect_request( + None, None, 302, "", {}, "http://example.com/file" + ) def test_download_size_cap_and_non_https_url(self): with tempfile.TemporaryDirectory() as temporary: - destination = Path(temporary) / 'file' + destination = Path(temporary) / "file" with self.assertRaises(ValueError): - BOOT.download('http://github.com/file', destination, 10) - response = io.BytesIO(b'payload-too-long') - response.geturl = lambda: 'https://github.com/file' - with patch.object(BOOT.urllib.request, 'build_opener') as opener: + BOOT.download("http://github.com/file", destination, 10) + response = io.BytesIO(b"payload-too-long") + response.geturl = lambda: "https://github.com/file" + with patch.object(BOOT.urllib.request, "build_opener") as opener: opener.return_value.open.return_value = response - with self.assertRaisesRegex(ValueError, 'size limit'): - BOOT.download('https://github.com/file', destination, 10) + with self.assertRaisesRegex(ValueError, "size limit"): + BOOT.download("https://github.com/file", destination, 10) def test_portable_cli_rejects_wrong_digest_before_import_or_execution(self): with tempfile.TemporaryDirectory() as temporary: + def fake_download(url, destination, limit): - destination.write_bytes(b'not the reviewed CLI archive') - with patch.object(BOOT, 'supports_attestation', return_value=False), \ - patch.object(BOOT, 'download', side_effect=fake_download), \ - patch.object(BOOT.tarfile, 'open') as read_archive: - with self.assertRaisesRegex(ValueError, 'reviewed pin'): + destination.write_bytes(b"not the reviewed CLI archive") + + with patch.object( + BOOT, "supports_attestation", return_value=False + ), patch.object( + BOOT, "download", side_effect=fake_download + ), patch.object( + BOOT.tarfile, "open" + ) as read_archive: + with self.assertRaisesRegex(ValueError, "reviewed pin"): BOOT.verification_cli(Path(temporary), allow_portable=True) read_archive.assert_not_called() def test_old_cli_and_capability_failure_are_not_accepted(self): - for text in ('gh version 2.46.0', 'gh version 2.97.0', 'unparseable'): - with patch.object(BOOT.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, text)): - self.assertFalse(BOOT.supports_attestation('gh')) + for text in ("gh version 2.46.0", "gh version 2.97.0", "unparseable"): + with patch.object( + BOOT.subprocess, + "run", + return_value=subprocess.CompletedProcess([], 0, text), + ): + self.assertFalse(BOOT.supports_attestation("gh")) def environment(self): - stack = __import__('contextlib').ExitStack() - stack.enter_context(patch.object(BOOT.os, 'geteuid', return_value=1000)) - stack.enter_context(patch.object(BOOT.platform, 'system', return_value='Linux')) - stack.enter_context(patch.object(BOOT.platform, 'machine', return_value='x86_64')) - stack.enter_context(patch.object(BOOT.platform, 'freedesktop_os_release', return_value={'ID': 'fedora', 'VERSION_ID': '44'})) + stack = __import__("contextlib").ExitStack() + stack.enter_context(patch.object(BOOT.os, "geteuid", return_value=1000)) + stack.enter_context( + patch.object(BOOT.platform, "system", return_value="Linux") + ) + stack.enter_context( + patch.object(BOOT.platform, "machine", return_value="x86_64") + ) + stack.enter_context( + patch.object( + BOOT.platform, + "freedesktop_os_release", + return_value={"ID": "fedora", "VERSION_ID": "44"}, + ) + ) return stack def test_check_mode_does_not_download_install_or_mutate(self): - with self.environment(), patch.object(BOOT, 'missing_packages', return_value=['gh']), \ - patch.object(BOOT, 'download') as download, patch.object(BOOT, 'install_dependencies') as install, \ - patch.object(BOOT.subprocess, 'run') as run, patch('sys.stdout', new_callable=io.StringIO): - report = BOOT.main(['--check']) - self.assertEqual(report['release'], 'v1.5.1') - self.assertIn('2.102.0', report['verification_requirement']) - download.assert_not_called(); install.assert_not_called(); run.assert_not_called() + with self.environment(), patch.object( + BOOT, "missing_packages", return_value=["gh"] + ), patch.object(BOOT, "download") as download, patch.object( + BOOT, "install_dependencies" + ) as install, patch.object( + BOOT.subprocess, "run" + ) as run, patch( + "sys.stdout", new_callable=io.StringIO + ): + report = BOOT.main(["--check"]) + self.assertEqual(report["release"], "v1.5.1") + self.assertIn("2.102.0", report["verification_requirement"]) + download.assert_not_called() + install.assert_not_called() + run.assert_not_called() def test_old_gh_stops_before_download(self): - with self.environment(), patch.object(BOOT, 'missing_packages', return_value=[]), \ - patch.object(BOOT, 'download') as download, \ - patch.object(BOOT.subprocess, 'run', return_value=subprocess.CompletedProcess([], 1)), \ - patch('builtins.open', side_effect=OSError('no tty')), \ - patch('sys.stdout', new_callable=io.StringIO): - with self.assertRaisesRegex(ValueError, 'attestation verification'): + with self.environment(), patch.object( + BOOT, "missing_packages", return_value=[] + ), patch.object(BOOT, "download") as download, patch.object( + BOOT.subprocess, + "run", + return_value=subprocess.CompletedProcess([], 1), + ), patch( + "builtins.open", side_effect=OSError("no tty") + ), patch( + "sys.stdout", new_callable=io.StringIO + ): + with self.assertRaisesRegex(ValueError, "attestation verification"): BOOT.main([]) download.assert_not_called() def test_orchestrator_passes_exact_pin_target_and_never_activates(self): verifier = BOOT.verifier_module() - with self.environment(), patch.object(BOOT, 'missing_packages', return_value=[]), \ - patch.object(BOOT, 'download') as download, patch.object(BOOT, 'verifier_module', return_value=verifier), \ - patch.object(verifier, 'verify_install', return_value={'installed': True}) as verify, \ - patch.object(BOOT.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, 'gh version 2.102.0')), \ - patch('sys.stdout', new_callable=io.StringIO): - BOOT.main(['--stage-root', '/tmp/test-stage']) + with self.environment(), patch.object( + BOOT, "missing_packages", return_value=[] + ), patch.object(BOOT, "download") as download, patch.object( + BOOT, "verifier_module", return_value=verifier + ), patch.object( + verifier, "verify_install", return_value={"installed": True} + ) as verify, patch.object( + BOOT.subprocess, + "run", + return_value=subprocess.CompletedProcess( + [], 0, "gh version 2.102.0" + ), + ), patch( + "sys.stdout", new_callable=io.StringIO + ): + BOOT.main(["--stage-root", "/tmp/test-stage"]) self.assertEqual(download.call_count, 7) args, kwargs = verify.call_args - self.assertEqual(args[1], BOOT.RELEASES['fedora', '44']['source']) - self.assertEqual(kwargs['target'], 'fedora44') - self.assertTrue(kwargs['install']) - self.assertNotIn('activate', kwargs) - self.assertFalse(args[0].exists(), 'Temporary downloads were not cleaned up') + self.assertEqual(args[1], BOOT.RELEASES["fedora", "44"]["source"]) + self.assertEqual(kwargs["target"], "fedora44") + self.assertTrue(kwargs["install"]) + self.assertNotIn("activate", kwargs) + self.assertFalse( + args[0].exists(), "Temporary downloads were not cleaned up" + ) + + def test_arch_family_builds_fork_from_source_with_pacman(self): + release = {"ID": "omarchy", "ID_LIKE": "arch", "VERSION_ID": "4.0.0"} + self.assertIs(BOOT.select_release(release), BOOT.ARCH) + self.assertIs( + BOOT.select_release({"ID": "arch", "BUILD_ID": "rolling"}), + BOOT.ARCH, + ) + with patch.object( + BOOT.subprocess, + "run", + side_effect=[ + subprocess.CompletedProcess([], 0), + subprocess.CompletedProcess([], 1), + ], + ): + self.assertEqual( + BOOT.missing_packages(("gtk4", "librsvg"), "pacman"), + ["librsvg"], + ) + with patch.object(BOOT, "missing_packages", return_value=["gtk4"]): + with self.assertRaisesRegex( + ValueError, "sudo pacman -S --needed gtk4" + ): + BOOT.install_dependencies(("gtk4",), BOOT.ARCH, no_deps=True) + with patch.object(BOOT.shutil, "which", return_value=None): + self.assertIn("rust", BOOT.source_packages(release)) + with patch.object(BOOT.shutil, "which", return_value="/usr/bin/cargo"): + self.assertNotIn("rust", BOOT.source_packages(release)) + + def test_arch_install_clones_builds_and_installs_without_activation(self): + def run(command, **kwargs): + if "build-bundle.py" in " ".join(command): + output = Path(command[command.index("--output") + 1]) + output.mkdir() + (output / "decksmith-1.5.1-abc.tar.gz").touch() + return subprocess.CompletedProcess(command, 0) + + with patch.object(BOOT, "install_dependencies"), patch.object( + BOOT.subprocess, "run", side_effect=run + ) as calls, patch("sys.stdout", new_callable=io.StringIO): + BOOT.source_install(("gtk4",), stage_root="/tmp/test-stage") + commands = [c.args[0] for c in calls.call_args_list] + self.assertEqual(commands[0][:2], ["git", "clone"]) + self.assertIn(BOOT.SOURCE_REPOSITORY, commands[0]) + self.assertEqual(commands[2][-2], "install") + self.assertIn("--stage-root", commands[2]) + self.assertNotIn("--activate", commands[2]) def test_dependency_hints_are_shared_and_unknown_distro_not_fedora(self): - self.assertIn('python3-gi', runtime_packages({'ID': 'debian'})) - self.assertIn('python3-gobject', runtime_packages({'ID': 'fedora'})) - self.assertEqual(runtime_packages({'ID': 'unknown'}), ()) - self.assertIn('distribution guide', package_hints({'ID': 'unknown'})['gdbus']) - self.assertIn('distribution guide', package_hints({'ID': 'opensuse-tumbleweed'})['gdbus']) + self.assertIn("python3-gi", runtime_packages({"ID": "debian"})) + self.assertIn("python3-gobject", runtime_packages({"ID": "fedora"})) + self.assertEqual(runtime_packages({"ID": "unknown"}), ()) + self.assertIn( + "distribution guide", package_hints({"ID": "unknown"})["gdbus"] + ) + self.assertIn( + "distribution guide", + package_hints({"ID": "opensuse-tumbleweed"})["gdbus"], + )