22 files changed
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5084,6 +5084,7 @@
"remowt-ui-prompt",
"tokio",
"tracing",
+ "tracing-journald",
"tracing-subscriber",
]
@@ -6576,6 +6577,17 @@
]
[[package]]
+name = "tracing-journald"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2d3a81ed245bfb62592b1e2bc153e77656d94ee6a0497683a65a12ccaf2438d0"
+dependencies = [
+ "libc",
+ "tracing-core",
+ "tracing-subscriber",
+]
+
+[[package]]
name = "tracing-log"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -75,6 +75,7 @@
tokio = { version = "1.45.1", features = ["fs", "macros", "rt", "rt-multi-thread", "sync", "time"] }
tracing = "0.1"
tracing-indicatif = "0.3.13"
+tracing-journald = "0.3.2"
tracing-opentelemetry = "0.33.0"
uuid = { version = "1", features = ["v4"] }
# For fixed coloring needs to be updated to https://github.com/tokio-rs/tracing/pull/3484
--- a/cmds/fleet/src/cmds/build_systems.rs
+++ b/cmds/fleet/src/cmds/build_systems.rs
@@ -14,6 +14,8 @@
use tokio::task::spawn_blocking;
use tracing::{Instrument, error, field, info, info_span, warn};
+const TARGET: &str = "build-systems";
+
#[derive(Parser)]
pub struct Deploy {
/// Disable automatic rollback
@@ -32,7 +34,7 @@
}
async fn build_task(config: Config, hostname: String, build_attr: &str) -> Result<Utf8PathBuf> {
- info!("building");
+ info!(target: TARGET, "building");
let host = config.host(&hostname)?;
// let action = Action::from(self.subcommand.clone());
let nixos = host.nixos_config()?;
@@ -43,7 +45,7 @@
// We already have system profiles for backups.
if !host.local {
- info!("adding gc root");
+ info!(target: TARGET, "adding gc root");
let local = config.local_host();
let plugin_id = local.ensure_nix_plugin().await?;
let nix = local
@@ -70,7 +72,7 @@
let build_attr = self.build_attr.clone();
for host in hosts {
let config = config.clone();
- let span = info_span!("build", host = field::display(&host.name));
+ let span = info_span!(target: TARGET, "build", host = field::display(&host.name));
let hostname = host.name;
let build_attr = build_attr.clone();
tasks.push(
@@ -78,7 +80,7 @@
let built = match build_task(config, hostname.clone(), &build_attr).await {
Ok(path) => path,
Err(e) => {
- error!("failed to deploy host: {:?}", e);
+ error!(target: TARGET, "failed to deploy host: {:?}", e);
return;
}
};
@@ -86,9 +88,9 @@
let mut out = current_dir().expect("cwd exists");
out.push(format!("built-{hostname}"));
- info!("linking iso image to {:?}", out);
+ info!(target: TARGET, "linking iso image to {:?}", out);
if let Err(e) = symlink(built, out) {
- error!("failed to symlink: {e}")
+ error!(target: TARGET, "failed to symlink: {e}")
}
})
.instrument(span),
@@ -105,7 +107,7 @@
let tasks = FuturesUnordered::new();
for host in hosts.into_iter() {
let config = config.clone();
- let span = info_span!("deploy", host = field::display(&host.name));
+ let span = info_span!(target: TARGET, "deploy", host = field::display(&host.name));
let hostname = host.name.clone();
let opts = opts.clone();
if let Some(deploy_kind) = opts.action_attr::<DeployKind>(&host, "deploy_kind")? {
@@ -125,7 +127,7 @@
{
Ok(path) => path,
Err(e) => {
- error!("failed to build host system closure: {:?}", e);
+ error!(target: TARGET, "failed to build host system closure: {:?}", e);
return;
}
};
@@ -133,7 +135,7 @@
let deploy_kind = match host.deploy_kind().await {
Ok(v) => v,
Err(e) => {
- error!("failed to query target deploy kind: {e}");
+ error!(target: TARGET, "failed to query target deploy kind: {e}");
return;
}
};
@@ -141,7 +143,7 @@
// TODO: Make disable_rollback a host attribute instead
let mut disable_rollback = self.disable_rollback;
if !disable_rollback && deploy_kind != DeployKind::Fleet {
- warn!("disabling rollback, as not supported by non-fleet deployment kinds");
+ warn!(target: TARGET, "disabling rollback, as not supported by non-fleet deployment kinds");
disable_rollback = true;
}
@@ -149,7 +151,7 @@
match upload_task(&host, GenerationStorage::Deployer, built).await {
Ok(v) => v,
Err(e) => {
- error!("upload failed: {e}");
+ error!(target: TARGET, "upload failed: {e}");
return;
}
};
@@ -169,7 +171,7 @@
)
.await
{
- error!("activation failed: {e}");
+ error!(target: TARGET, "activation failed: {e}");
}
})
.instrument(span),
--- /dev/null
+++ b/cmds/fleet/src/log_tree.rs
@@ -0,0 +1,243 @@
+//! Streaming, indented tree formatter.
+//!
+//! Unlike `tracing-tree` it keeps no per-thread depth state, so interleaved
+//! parallel output stays correct; unlike `tracing-forest` it doesn't buffer
+//! whole trees, so logs stream live. Visual style mirrors `@meteor-it/logger`'s
+//! node receiver: a fixed-width, right-aligned, level-colored target column,
+//! then a magenta `>`/`<` gutter for span enter/exit (with duration on exit),
+//! and 2-space indentation per depth for events.
+
+use std::{
+ fmt::Write as _,
+ io::{IsTerminal as _, Write as _},
+ sync::{
+ Mutex,
+ atomic::{AtomicBool, Ordering},
+ },
+ time::{Duration, Instant},
+};
+
+use tracing::{Event, Level, Subscriber, field::Field, field::Visit, span::Attributes, span::Id};
+use tracing_subscriber::{
+ fmt::MakeWriter,
+ layer::{Context, Layer},
+ registry::{LookupSpan, Scope},
+};
+
+/// Width of the right-aligned target column.
+const NAME_LIMIT: usize = 18;
+
+#[derive(Default)]
+struct Fields {
+ message: String,
+ rest: String,
+}
+
+impl Visit for Fields {
+ fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
+ if field.name() == "message" {
+ let _ = write!(self.message, "{value:?}");
+ } else {
+ let _ = write!(self.rest, " {}={:?}", field.name(), value);
+ }
+ }
+}
+
+impl Fields {
+ fn render(&self) -> String {
+ format!("{}{}", self.message, self.rest).trim().to_owned()
+ }
+}
+
+/// Per-span state stashed in the registry's extensions.
+struct SpanData {
+ fields: String,
+ started: Instant,
+ // Whether the span's enter line was written, so we only emit an exit line
+ // for spans that actually appeared.
+ opened: AtomicBool,
+}
+
+/// SGR code → escape for the level-colored target column.
+fn level_code(level: &Level) -> &'static str {
+ match *level {
+ Level::ERROR => "31m", // red
+ Level::WARN => "33m", // yellow
+ Level::INFO => "34m", // blue
+ Level::DEBUG => "39m", // default fg
+ Level::TRACE => "90m", // gray
+ }
+}
+
+/// 2-space indent per depth, then a gutter symbol.
+fn gutter(depth: usize, sym: char) -> String {
+ let mut s = " ".repeat(depth);
+ s.push(sym);
+ s
+}
+
+fn fmt_dur(d: Duration) -> String {
+ if d.as_secs_f64() >= 1.0 {
+ format!("{:.2}s", d.as_secs_f64())
+ } else if d.as_millis() >= 1 {
+ format!("{}ms", d.as_millis())
+ } else {
+ format!("{}µs", d.as_micros())
+ }
+}
+
+pub struct TreeLayer<W> {
+ writer: W,
+ color: bool,
+ // Serializes the whole compute-and-write so enter/exit decisions, output
+ // ordering, and the writes stay consistent under parallel emission.
+ last: Mutex<Vec<Id>>,
+}
+
+impl<W> TreeLayer<W> {
+ pub fn new(writer: W) -> Self {
+ Self {
+ writer,
+ color: std::io::stderr().is_terminal() && std::env::var_os("NO_COLOR").is_none(),
+ last: Mutex::new(vec![]),
+ }
+ }
+
+ /// Raw SGR escape, or empty when color is disabled.
+ fn sgr(&self, code: &str) -> String {
+ if self.color {
+ format!("\x1b[{code}")
+ } else {
+ String::new()
+ }
+ }
+
+ fn name_col(&self, mut target: &str, code: &str) -> String {
+ let w = NAME_LIMIT;
+ if target.len() > w {
+ target = &target[..w];
+ }
+ if self.color {
+ format!("\x1b[{code}\x1b[1m{target:>w$}\x1b[0m")
+ } else {
+ format!("{target:>w$}")
+ }
+ }
+
+ fn print_ancestors<S>(&self, scope: Option<Scope<'_, S>>, out: &mut String) -> usize
+ where
+ S: Subscriber + for<'a> LookupSpan<'a>,
+ {
+ let scope: Vec<_> = scope.map(|s| s.from_root().collect()).unwrap_or_default();
+
+ let mut _last = self.last.lock().expect("not poisoned");
+
+ let skip = scope
+ .iter()
+ .enumerate()
+ .take_while(|(i, s)| _last.get(*i) == Some(&s.id()))
+ .count();
+
+ for (depth, span) in scope.iter().enumerate().skip(skip) {
+ let ext = span.extensions();
+ let Some(data) = ext.get::<SpanData>() else {
+ continue;
+ };
+ let newly_open = !data.opened.swap(true, Ordering::Relaxed);
+ let _ = writeln!(
+ out,
+ "{} {}{}{} {}{}{}",
+ self.name_col(span.metadata().target(), "44m"),
+ self.sgr("35m"),
+ gutter(depth, if newly_open { '>' } else { '.' }),
+ self.sgr("1m"),
+ span.name(),
+ data.fields,
+ self.sgr("0m"),
+ );
+ }
+
+ *_last = scope.iter().map(|v| v.id()).collect();
+
+ scope.len()
+ }
+}
+
+impl<S, W> Layer<S> for TreeLayer<W>
+where
+ S: Subscriber + for<'a> LookupSpan<'a>,
+ W: for<'a> MakeWriter<'a> + 'static,
+{
+ fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) {
+ let span = ctx.span(id).expect("new span exists");
+ let mut fields = Fields::default();
+ attrs.record(&mut fields);
+ span.extensions_mut().insert(SpanData {
+ fields: fields.rest,
+ started: Instant::now(),
+ opened: AtomicBool::new(false),
+ });
+ }
+
+ fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) {
+ let mut out = String::new();
+
+ let scope = ctx.event_scope(event);
+ let pad = self.print_ancestors(scope, &mut out);
+
+ let mut fields = Fields::default();
+ event.record(&mut fields);
+ let message = fields.render();
+ let meta = event.metadata();
+ for (i, message) in message.split('\n').enumerate() {
+ // Event line: black bg + colored target column, then indent + message.
+ let _ = writeln!(
+ out,
+ "{}{}{}{}{}",
+ self.sgr("40m"),
+ self.name_col(
+ if i == 0 { meta.target() } else { "|" },
+ level_code(meta.level())
+ ),
+ self.sgr("0m"),
+ gutter(pad, ' '),
+ message,
+ );
+ }
+
+ let _ = self.writer.make_writer().write_all(out.as_bytes());
+ }
+
+ fn on_close(&self, id: Id, ctx: Context<'_, S>) {
+ let Some(span) = ctx.span(&id) else { return };
+ let (opened, dur) = span
+ .extensions()
+ .get::<SpanData>()
+ .map_or((false, Duration::ZERO), |d| {
+ (d.opened.load(Ordering::Relaxed), d.started.elapsed())
+ });
+ if !opened {
+ return;
+ }
+
+ let mut out = String::new();
+ let _pad = self.print_ancestors(ctx.span_scope(&id), &mut out);
+
+ let depth = span.scope().skip(1).count(); // ancestors == depth
+ let _ = writeln!(
+ out,
+ "{} {}{}{} {}{} (Done in {}){}",
+ self.name_col(span.metadata().target(), "44m"),
+ self.sgr("35m"),
+ gutter(depth, '<'),
+ self.sgr("1m"),
+ span.name(),
+ self.sgr("22m"),
+ fmt_dur(dur),
+ self.sgr("0m"),
+ );
+
+ let _guard = self.last.lock().expect("not poisoned");
+ let _ = self.writer.make_writer().write_all(out.as_bytes());
+ }
+}
--- a/cmds/fleet/src/main.rs
+++ b/cmds/fleet/src/main.rs
@@ -1,6 +1,7 @@
#![recursion_limit = "512"]
pub(crate) mod cmds;
+mod log_tree;
use std::{process::ExitCode, sync::Arc};
@@ -21,6 +22,7 @@
use human_repr::HumanCount;
#[cfg(feature = "indicatif")]
use indicatif::{ProgressState, ProgressStyle};
+use log_tree::TreeLayer;
use nix_eval::{
eval_store, gc_register_my_thread, gc_unregister_my_thread, init_libraries, init_tokio_for_nix,
};
@@ -128,15 +130,21 @@
fn setup_logging(opts: &RootOpts) -> Result<()> {
#[cfg(feature = "indicatif")]
let indicatif_layer = {
+ use std::fmt;
use std::time::Duration;
IndicatifLayer::new().with_max_progress_bars(10, Some(ProgressStyle::default_spinner()))
+ .with_span_child_prefix_indent(" ")
+ .with_span_child_prefix_symbol("")
.with_progress_style(
ProgressStyle::with_template(
- "{color_start}{span_child_prefix} {span_name}{{{span_fields}}}{color_end} {wide_msg} {color_start}{download_progress} {elapsed}{color_end}",
+ "{span_child_prefix:.magenta}{chevron:.magenta}{span_name:<18.blue.bold} {wide_msg} {span_fields:.dim} {color_start}{download_progress} {elapsed}{color_end}",
)
.unwrap()
- .with_key("download_progress", |state: &ProgressState, writer: &mut dyn std::fmt::Write| {
+ .with_key("chevron", |_: &ProgressState, writer: &mut dyn fmt::Write| {
+ let _ = write!(writer, "> ");
+ })
+ .with_key("download_progress", |state: &ProgressState, writer: &mut dyn fmt::Write| {
let Some(len) = state.len() else {
return;
};
@@ -149,7 +157,7 @@
})
.with_key(
"color_start",
- |state: &ProgressState, writer: &mut dyn std::fmt::Write| {
+ |state: &ProgressState, writer: &mut dyn fmt::Write| {
let elapsed = state.elapsed();
if elapsed > Duration::from_secs(60) {
@@ -163,7 +171,7 @@
)
.with_key(
"color_end",
- |state: &ProgressState, writer: &mut dyn std::fmt::Write| {
+ |state: &ProgressState, writer: &mut dyn fmt::Write| {
if state.elapsed() > Duration::from_secs(30) {
let _ = write!(writer, "\x1b[0m");
}
@@ -174,12 +182,15 @@
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
- let reg = tracing_subscriber::registry().with({
- let sub = tracing_subscriber::fmt::layer().without_time();
+ let tree = {
#[cfg(feature = "indicatif")]
- let sub = sub.with_writer(indicatif_layer.get_stderr_writer());
- sub.with_filter(filter) // .without,
- });
+ let writer = indicatif_layer.get_stderr_writer();
+ #[cfg(not(feature = "indicatif"))]
+ let writer = || std::io::stderr();
+ TreeLayer::new(writer)
+ };
+
+ let reg = tracing_subscriber::registry().with(tree.with_filter(filter));
#[cfg(feature = "indicatif")]
let reg = reg.with(indicatif_layer);
@@ -200,14 +211,12 @@
let logger = OpenTelemetryTracingBridge::new(&log_provider);
let tracer = span_provider.tracer("fleet");
- let reg = reg
- .with(tracing_opentelemetry::layer().with_tracer(tracer))
- .with(logger);
-
- reg.init();
+ reg.with(tracing_opentelemetry::layer().with_tracer(tracer))
+ .with(logger)
+ .init();
} else {
reg.init();
- };
+ }
Ok(())
}
@@ -252,7 +261,6 @@
.await
.expect("primary task panicked")
})
- // async_main(opts)
}
async fn main_real(opts: RootOpts) -> Result<()> {
--- a/crates/fleet-base/src/deploy.rs
+++ b/crates/fleet-base/src/deploy.rs
@@ -290,8 +290,10 @@
if !host.local {
info!("uploading system closure");
let mut tries = 0;
+ // TODO: Use gc prefix?..
loop {
- match host.remote_derivation(&generation).await {
+ let name = format!("{}-system", host.gc_root_prefix());
+ match host.remote_derivation(&generation, Some(&name)).await {
Ok(remote) => {
assert!(remote == generation, "CA derivations aren't implemented");
return Ok(remote);
--- a/crates/fleet-base/src/host.rs
+++ b/crates/fleet-base/src/host.rs
@@ -13,16 +13,16 @@
use camino::{Utf8Path, Utf8PathBuf};
use chrono::{DateTime, Utc};
use fleet_shared::SecretData;
-use nix_eval::{Store, Value, drv::DrvGraph, eval_store, nix_go, nix_go_json, util::assert_warn};
+use nix_eval::{Store, Value, eval_store, nix_go, nix_go_json, util::assert_warn};
use remowt_client::{AgentBundle, Remowt};
use remowt_endpoints::fs::FsClient;
use remowt_fleet::NixClient;
-use remowt_link_shared::BifConfig;
+use remowt_link_shared::{Address, BifConfig};
use remowt_ui_prompt::auto::AutoPrompter;
use remowt_ui_prompt::bifrost::PromptEndpoints;
use remowt_ui_prompt::{PrependSourcePrompter, Source};
use tempfile::NamedTempFile;
-use tokio::{sync::OnceCell, task::spawn_blocking};
+use tokio::{process::Command, sync::OnceCell, task::spawn_blocking};
use tracing::{info, instrument, warn};
use crate::fleetdata::{
@@ -104,11 +104,14 @@
session_destination: OnceLock<String>,
legacy_ssh_store: OnceLock<bool>,
+ /// fleetConfiguration.hosts.host, None for local (as local is only used for local tool overrides)
pub host_config: Option<Value>,
pub nixos_config: OnceLock<Value>,
pub nixos_unchecked_config: OnceLock<Value>,
pub pkgs_override: Option<Value>,
+ binary_cache: OnceCell<Option<BinaryCache>>,
+
// TODO: Move command helpers away with connectivity refactor
pub local: bool,
pub remowt: OnceCell<Remowt>,
@@ -131,13 +134,35 @@
AgentBundle::from_dir(agents_dir()?)
}
-enum RemoteDerivationMode {
- /// Closure is (pre)signed, remote host is expected to trust our signatures.
- CopySigned,
- /// Closure is not signed, using escalated nix daemon to bypass signature verification.
- CopyPrivileged,
- /// Closure is uploaded to attic, remote host is expected to trust its signature.
- Attic { name: String },
+#[derive(Clone)]
+pub enum BinaryCache {
+ Attic { name: String, pin: bool },
+}
+impl BinaryCache {
+ async fn upload(&self, local: &Remowt, path: &Utf8Path, pin_name: Option<&str>) -> Result<()> {
+ match self {
+ BinaryCache::Attic { name, pin } => {
+ let mut cmd = local.cmd("attic");
+ cmd.arg("push").arg(name).arg(path.as_str());
+ cmd.run()
+ .await
+ .context("failed to push path to attic (is attic-cli on PATH?)")?;
+
+ if *pin && let Some(pin_name) = pin_name {
+ let mut cmd = local.cmd("attic");
+ cmd.arg("pin")
+ .arg("create")
+ .arg(name)
+ .arg(pin_name)
+ .arg(path.as_str());
+ cmd.run()
+ .await
+ .context("failed to pin path in attic (is your attic-cli compatible?)")?;
+ }
+ Ok(())
+ }
+ }
+ }
}
impl ConfigHost {
@@ -156,6 +181,32 @@
.set(legacy)
.expect("legacy ssh store is already set")
}
+ pub fn binary_cache(&self) -> Result<Option<BinaryCache>> {
+ if let Some(v) = self.binary_cache.get() {
+ return Ok(v.clone());
+ }
+ let cache = if let Some(host_config) = &self.host_config {
+ let cache = nix_go!(host_config.cache);
+ let enable: bool = nix_go_json!(cache.enable);
+ if enable {
+ let kind: String = nix_go_json!(cache.kind);
+ let name: String = nix_go_json!(cache.name);
+ match kind.as_str() {
+ "attic" => {
+ let pin: bool = nix_go_json!(cache.pin);
+ Some(BinaryCache::Attic { name, pin })
+ }
+ v => bail!("unknown binary cache kind: {v}"),
+ }
+ } else {
+ None
+ }
+ } else {
+ None
+ };
+ let _ = self.binary_cache.set(cache.clone());
+ Ok(cache)
+ }
pub async fn deploy_kind(&self) -> Result<DeployKind> {
self.deploy_kind.get_or_try_init(|| async {
let remowt = self.remowt().await?;
@@ -238,8 +289,10 @@
let built = plugin
.build("out")
.context("failed to build the fleet nix plugin")?;
+ let pin = format!("{}-remowt-plugin-fleet", self.gc_root_prefix());
let copied = self
- .remote_derivation(&built)
+ // TODO: Pin? But this pin will be host-specific, and heavy because of nix dependency...
+ .remote_derivation(&built, Some(&pin))
.await
.context("failed to copy the fleet nix plugin to the host store")?;
let bin = copied.join("bin/remowt-plugin-fleet");
@@ -248,6 +301,13 @@
.run0_load_plugin_path(NIX_PLUGIN_ID, bin.as_str())
.await
.context("failed to load the fleet nix plugin")?;
+ // Should not fail, as plugin loading waits for connection on remote, and we only wait for forward... On the other hand, we should monitor connection to the remote itself somewhere too.
+ self.remowt()
+ .await?
+ .rpc()
+ .wait_for_connection_to(Address::Plugin(NIX_PLUGIN_ID))
+ .await
+ .map_err(|_| anyhow!("failed to wait"))?;
anyhow::Ok(())
})
.await?;
@@ -339,17 +399,31 @@
}
/// Returns path for futureproofing, as path might change i.e on conversion to CA
#[instrument(skip(self))]
- pub async fn remote_derivation(&self, path: &Utf8Path) -> Result<Utf8PathBuf> {
+ pub async fn remote_derivation(
+ &self,
+ path: &Utf8Path,
+ pin_name: Option<&str>,
+ ) -> Result<Utf8PathBuf> {
let path = path.to_owned();
if self.local {
// Path is located locally, thus already trusted.
return Ok(path);
}
+ let cache = self.binary_cache()?;
+ let mut cache = if let Some(cache) = cache {
+ let local = self.config.local_host().remowt().await?;
+ let path = path.to_owned();
+ let pin_name = pin_name.map(str::to_owned);
+ Some(tokio::spawn(async move {
+ if let Err(e) = cache.upload(&local, &path, pin_name.as_deref()).await {
+ warn!("failed to upload closure {path} to cache: {e:?}");
+ }
+ }))
+ } else {
+ None
+ };
let sign: Pin<Box<dyn Future<Output = Result<()>> + Send>> = {
let path = path.clone();
- let graph = DrvGraph::resolve(&eval_store(), &path)
- .context("failed to resolve graph to be uploaded")?;
- info!("signing {} paths", graph.nodes.len());
Box::pin(async move {
let local = self.config.local_host();
@@ -377,13 +451,62 @@
.expect("copy_to panicked")
.context("copying closure to remote store")?;
}
+
+ if let Some(cache) = cache.take() {
+ let _ = cache.await;
+ }
+
Ok(path)
}
+
+ /// Like [`Self::remote_derivation`], but instead of copying the closure over
+ /// the nix daemon tunnel, uploads it to an attic binary cache and has the
+ /// remote host substitute it from there. The remote is expected to already
+ /// trust the cache's signing key and have it configured as a substituter.
+ #[instrument(skip(self))]
+ pub async fn remote_derivation_attic(
+ &self,
+ path: &Utf8Path,
+ cache_name: &str,
+ ) -> Result<Utf8PathBuf> {
+ let path = path.to_owned();
+ if self.local {
+ // Path is located locally, thus already trusted.
+ return Ok(path);
+ }
+
+ info!("uploading {path} closure to attic cache {cache_name}");
+ let status = Command::new("attic")
+ .arg("push")
+ .arg(cache_name)
+ .arg(path.as_str())
+ .status()
+ .await
+ .context("failed to spawn attic-cli, is it on PATH?")?;
+ ensure!(status.success(), "attic push failed: {status}");
+
+ info!("substituting {path} on the remote host");
+ let nix = self.nix_client().await?;
+ let substituted = nix
+ .substitute(vec![path.clone()])
+ .await
+ .map_err(|e| anyhow!("{e:?}"))?
+ .map_err(|e| anyhow!("{e}"))?;
+ ensure!(
+ substituted.contains(&path),
+ "remote host failed to substitute {path} from attic cache {cache_name}"
+ );
+ Ok(path)
+ }
}
struct HostSecretDefinition(Value);
impl ConfigHost {
+ pub fn gc_root_prefix(&self) -> String {
+ format!("{}-{}", self.config.gc_root_prefix(), self.name)
+ }
+
// TOCTOU is possible here in case if config is changed, but this case is not handled anywhere anyway,
// assuming getting tags always returns the same value.
pub fn tags(&self) -> Result<Vec<String>> {
@@ -504,6 +627,7 @@
cell
},
pkgs_override: Some(self.default_pkgs.clone()),
+ binary_cache: OnceCell::new(),
local: true,
remowt: OnceCell::new(),
@@ -546,6 +670,7 @@
nixos_unchecked_config: OnceLock::new(),
groups: OnceLock::new(),
pkgs_override: None,
+ binary_cache: OnceCell::new(),
// TODO: Remove with connectivit refactor
local: self.localhost == name,
--- a/crates/fleet-base/src/keys.rs
+++ b/crates/fleet-base/src/keys.rs
@@ -9,6 +9,9 @@
use crate::{fleetdata::SecretOwner, host::Config};
impl Config {
+ pub fn gc_root_prefix(&self) -> String {
+ self.data.gc_root_prefix.clone()
+ }
fn cached_host_key(&self, host: &str) -> Option<String> {
let hosts = self.data.hosts.read().expect("no poisoning");
let key = hosts.get(host).map(|h| &h.encryption_key);
--- a/crates/fleet-base/src/primops.rs
+++ b/crates/fleet-base/src/primops.rs
@@ -119,7 +119,7 @@
.context("failed to build generator for target host")?;
let generator = host_on
- .remote_derivation(&generator)
+ .remote_derivation(&generator, None)
.await
.context("failed to copy generator to target host")?;
--- a/crates/nix-eval/src/lib.rs
+++ b/crates/nix-eval/src/lib.rs
@@ -39,7 +39,7 @@
make_attrs, make_bindings_builder, make_list, make_list_builder, realised_string,
realised_string_free, realised_string_get_buffer_size, realised_string_get_buffer_start,
realised_string_get_store_path, realised_string_get_store_path_count, register_primop,
- set_err_msg, setting_set, state_free, store_copy_closure, store_free, store_open,
+ set_err_msg, setting_get, setting_set, state_free, store_copy_closure, store_free, store_open,
store_parse_path, store_path_free, store_path_name, string_realise, value, value_call,
value_decref, value_force, value_incref,
};
@@ -390,6 +390,14 @@
with_default_context(|c, _| unsafe { setting_set(c, s.as_ptr(), v.as_ptr()) }).map(|_| ())
}
+pub fn get_setting(s: &CStr) -> Result<String> {
+ let mut out = String::new();
+ with_default_context(|c, _| unsafe {
+ setting_get(c, s.as_ptr(), Some(copy_nix_str), (&raw mut out).cast())
+ })?;
+ Ok(out)
+}
+
#[derive(Debug)]
pub struct AddedFile {
pub store_path: Utf8PathBuf,
--- a/crates/nix-eval/src/logging.rs
+++ b/crates/nix-eval/src/logging.rs
@@ -109,6 +109,12 @@
(ActivityType::CopyPaths, []) => {
debug_span!(target: "nix::copy-paths", "copying paths")
}
+ // Umbrella progress activity. Nix emits it at lvlError ("always show"),
+ // which would otherwise become a noisy ERROR `action` span — give it an
+ // explicit debug level like the other container activities.
+ (ActivityType::Builds, []) => {
+ debug_span!(target: "nix::builds", "building")
+ }
(ActivityType::Unknown, [])
if s.starts_with("copying \"") && s.ends_with("\" to the store") =>
{
@@ -365,6 +371,16 @@
BuildGraphGuard { paths }
}
+/// Drop the lazily-created `building` span for a derivation. Used when a build
+/// fails: its dependents will never build, so their speculatively-created spans
+/// (made when a dependency started building) would otherwise linger until the
+/// whole build graph is torn down.
+pub fn remove_drv_span(drv_path: &Utf8Path) {
+ if let Some(entry) = DRV_GRAPH.lock().expect("not poisoned").get_mut(drv_path) {
+ entry.span = None;
+ }
+}
+
fn ensure_drv_span(drv_path: &Utf8Path) -> Option<Span> {
let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");
@@ -524,7 +540,7 @@
// ResultType::FileLinked => todo!(),
(ResultType::BuildLogLine, [Str(s)]) => {
let s = ansi_filter(s);
- info!("{s}");
+ info!(target: "nix::build", "{s}");
}
// ResultType::UntrustedPath => todo!(),
// ResultType::CorruptedPath => todo!(),
--- a/crates/nix-eval/src/scheduler.rs
+++ b/crates/nix-eval/src/scheduler.rs
@@ -45,9 +45,36 @@
},
}
+// Derivations whose names contain any of these are "heavy": memory-hungry builds that already
+// saturate all cores internally (e.g. composable-kernels), so running anything alongside them
+// just thrashes RAM. A heavy build grabs every semaphore permit and runs exclusively.
+const DEFAULT_HEAVY_DRVS: &[&str] = &[
+ "composable-kernel",
+ "composable_kernel",
+ "ghc",
+ "gfortran",
+ "llvm",
+ "clang",
+];
+
+fn heavy_patterns_from_env() -> Vec<String> {
+ let mut pats: Vec<String> = DEFAULT_HEAVY_DRVS.iter().map(|s| (*s).to_owned()).collect();
+ if let Ok(extra) = std::env::var("FLEET_HEAVY_DRVS") {
+ pats.extend(
+ extra
+ .split(',')
+ .map(str::trim)
+ .filter(|s| !s.is_empty())
+ .map(str::to_owned),
+ );
+ }
+ pats
+}
+
pub struct Scheduler {
store: Arc<Store>,
parallelism: usize,
+ heavy_patterns: Vec<String>,
events: broadcast::Sender<BuildEvent>,
}
@@ -58,10 +85,20 @@
Self {
store: eval_store(),
parallelism,
+ heavy_patterns: heavy_patterns_from_env(),
events,
}
}
+ // Permits a drv must hold to run. Heavy drvs take all of them, pausing everything else.
+ fn permits_for(&self, name: &str) -> u32 {
+ if self.heavy_patterns.iter().any(|p| name.contains(p)) {
+ self.parallelism as u32
+ } else {
+ 1
+ }
+ }
+
pub fn subscribe(&self) -> broadcast::Receiver<BuildEvent> {
self.events.subscribe()
}
@@ -139,6 +176,7 @@
.get(&path)
.map(|n| n.name.clone())
.unwrap_or_default();
+ crate::logging::remove_drv_span(&path);
let _ = self.events.send(BuildEvent::DrvCancelled {
drv_path: path.clone(),
name,
@@ -153,8 +191,21 @@
let graph = graph.clone();
let wanted_here = wanted.get(&path).cloned().unwrap_or_default();
let store = self.store.clone();
+ let permits = graph
+ .nodes
+ .get(&path)
+ .map(|n| self.permits_for(&n.name))
+ .unwrap_or(1);
in_flight.push(tokio::spawn(async move {
- let _permit = sem.acquire_owned().await.expect("semaphore not closed");
+ if permits > 1 {
+ debug!(permits, "heavy drv: waiting for exclusive build slot");
+ }
+ // Heavy drvs acquire every permit, so they only start once all in-flight
+ // builds drain and nothing new can slip in (the semaphore is FIFO-fair).
+ let _permit = sem
+ .acquire_many_owned(permits)
+ .await
+ .expect("semaphore not closed");
let node = graph
.nodes
.get(&path)
@@ -224,6 +275,7 @@
propagate_done(&dependents, &mut indeg, &mut ready, &finished);
}
Err(e) => {
+ crate::logging::remove_drv_span(&finished);
failed.insert(finished.clone(), format!("{e:#}"));
mark_tainted(&dependents, &finished, &mut tainted);
propagate_done(&dependents, &mut indeg, &mut ready, &finished);
@@ -368,11 +420,23 @@
// TODO: Parallelism as a metric works poorly with multiple machines, but I haven't thought about bringing
// hercy here yet. In case of remote machines - they will handle parallelism on their own, and this one
// will work as a hard cap.
+/// Number of derivations to build concurrently, taken from Nix's `max-jobs`
+/// setting. `auto` resolves to the CPU count, matching Nix itself.
+fn max_jobs() -> usize {
+ let cores = || {
+ std::thread::available_parallelism()
+ .map(|p| p.get())
+ .unwrap_or(4)
+ };
+ match crate::get_setting(c"max-jobs") {
+ Ok(v) if v.trim() == "auto" => cores(),
+ Ok(v) => v.trim().parse().unwrap_or_else(|_| cores()),
+ Err(_) => cores(),
+ }
+}
+
pub fn build_graph_sync(graph: Arc<DrvGraph>, root_outputs: Vec<String>) -> Result<()> {
- let parallelism = std::thread::available_parallelism()
- .map(|p| p.get())
- .unwrap_or(4);
- let scheduler = Scheduler::new(parallelism);
+ let scheduler = Scheduler::new(max_jobs());
crate::await_in_nix(async move { scheduler.run(graph, root_outputs).await })
.context("scheduler run")
}
--- a/crates/remowt-fleet/src/lib.rs
+++ b/crates/remowt-fleet/src/lib.rs
@@ -28,6 +28,8 @@
Sign(String),
#[error("listing generations failed: {0}")]
ListGenerations(String),
+ #[error("substitution failed: {0}")]
+ Substitute(String),
}
#[endpoints(ns = 91)]
@@ -60,6 +62,20 @@
.map_err(|e| NixError::Sign(e.to_string()))
}
+ /// Download the given paths (and their closures) into the local store using
+ /// the configured substituters. Used to fetch closures that were uploaded to
+ /// a binary cache (e.g. attic) instead of copied over the nix daemon tunnel.
+ #[endpoints(id = 6)]
+ async fn substitute(&self, paths: Vec<Utf8PathBuf>) -> Result<Vec<Utf8PathBuf>, NixError> {
+ spawn_blocking(move || {
+ let store = eval_store();
+ store.substitute_paths(&paths)
+ })
+ .await
+ .expect("substitution panicked")
+ .map_err(|e| NixError::Substitute(e.to_string()))
+ }
+
#[endpoints(id = 5)]
async fn list_generations(
&self,
--- /dev/null
+++ b/modules/cache.nix
@@ -0,0 +1,61 @@
+{
+ lib,
+ fleetLib,
+ config,
+ ...
+}:
+let
+ inherit (lib.options) mkOption mkEnableOption literalExpression;
+ inherit (lib.types)
+ enum
+ str
+ bool
+ submodule
+ ;
+ inherit (fleetLib.options) mkHostsOption;
+
+ _file = ./cache.nix;
+
+ cacheOptions = {
+ enable = mkEnableOption "fleet binary cache";
+ kind = mkOption {
+ description = ''
+ Kind of the binary cache to use.
+ '';
+ type = enum [ "attic" ];
+ default = "attic";
+ };
+ name = mkOption {
+ description = ''
+ Name of the binary cache.
+ '';
+ type = str;
+ };
+ pin = mkOption {
+ description = ''
+ Whether to pin pushed paths in the attic cache, preventing them from
+ being garbage collected.
+
+ Requires compatible fork of attic server+client: https://github.com/zhaofengli/attic/pull/227#issuecomment-4759154796
+ '';
+ type = bool;
+ default = false;
+ };
+ };
+in
+{
+ options = {
+ cache = cacheOptions;
+ hosts = mkHostsOption {
+ inherit _file;
+ options.cache = mkOption {
+ description = ''
+ Binary cache configuration for this host.
+ '';
+ type = submodule { options = cacheOptions; };
+ default = config.cache;
+ defaultText = literalExpression "fleetConfiguration.cache";
+ };
+ };
+ };
+}
--- a/modules/module-list.nix
+++ b/modules/module-list.nix
@@ -1,5 +1,6 @@
[
./assertions.nix
+ ./cache.nix
./fleetLib.nix
./hosts.nix
./nixos.nix
--- a/pkgs/default.nix
+++ b/pkgs/default.nix
@@ -12,8 +12,6 @@
fleet-generator-helper = callPackage ./fleet-generator-helper.nix { inherit craneLib; };
inherit remowt-agents-bundle;
- remowt-plugin-fleet = callPackage ./remowt-plugin-fleet.nix {
- inherit craneLib inputs remowt-agents-bundle;
- };
+ remowt-plugin-fleet = callPackage ./remowt-plugin-fleet.nix { inherit craneLib inputs; };
remowt-ssh = callPackage ./remowt-ssh.nix { inherit craneLib remowt-agents-bundle; };
}
1use std::borrow::Cow;2use std::collections::{BTreeMap, HashMap};3use std::fs::Permissions;4use std::future::pending;5use std::io;6use std::os::unix::fs::PermissionsExt as _;7use std::path::PathBuf;8use std::sync::{Arc, Mutex, OnceLock};910use bifrostlink::Rpc;11use bifrostlink::declarative::RemoteEndpoints;12use bifrostlink_ports::stdio::from_stdio;13use bifrostlink_ports::unix_socket::from_socket;14use clap::Parser;15use remowt_endpoints::{16 forward::Forward, fs::Fs, iroh_tunnel::IrohTunnel, nix_daemon::NixDaemon, pty::Pty,17 subprocess::Subprocess, systemd::Systemd,18};19use remowt_link_shared::iroh_tunnel::TunnelDialer;20use remowt_link_shared::{Address, BifConfig, gateway};21use remowt_polkit_shared::{Identity, PidDisplay, emphasize};22use remowt_ui_prompt::bifrost::{PromptEndpointsClient, serve_prompts};23use remowt_ui_prompt::rofi::RofiPrompter;24use remowt_ui_prompt::{PrependSourcePrompter, Prompter, Source};25use tokio::fs;26use tokio::net::UnixStream;27use tokio::runtime::Builder;28use tokio::task::AbortHandle;29use tracing::{debug, trace};30use zbus::fdo;31use zbus::zvariant::{OwnedValue, Str};32use zbus::{Connection, interface};33use zbus_polkit::policykit1::Subject;3435use self::helper::{Helper, SocketHelper, SuidHelper};3637pub mod askpass;38pub mod editor;39pub mod helper;4041struct CancelTaskOnDrop {42 tasks: Arc<Mutex<HashMap<String, AbortHandle>>>,43 handle: String,44}45impl Drop for CancelTaskOnDrop {46 fn drop(&mut self) {47 debug!("cancel on drop");48 if let Some(task) = self49 .tasks50 .lock()51 .expect("not poisoned")52 .remove(&self.handle)53 {54 task.abort();55 }56 }57}5859struct Agent<H, P> {60 tasks: Arc<Mutex<HashMap<String, AbortHandle>>>,61 helper: H,62 prompter: P,63}64impl<H, P> Agent<H, P> {65 fn new(helper: H, prompter: P) -> Self {66 Agent {67 tasks: Arc::new(Mutex::new(HashMap::new())),68 helper,69 prompter,70 }71 }72}7374#[interface(name = "org.freedesktop.PolicyKit1.AuthenticationAgent")]75impl<H, P> Agent<H, P>76where77 H: Helper + Clone + Send + Sync + 'static,78 P: Prompter + Clone + Send + Sync + 'static,79{80 81 #[allow(clippy::too_many_arguments)]82 async fn begin_authentication(83 &self,84 action_id: String,85 message: String,86 _icon_name: String,87 mut details: BTreeMap<String, String>,88 cookie: String,89 identities: Vec<Identity>,90 ) -> zbus::fdo::Result<()> {91 use std::fmt::Write;92 debug!("begin auth");93 let _cancel_guard = Arc::new(OnceLock::new());94 let task = {95 let helper = self.helper.clone();96 let prompter = self.prompter.clone();97 let cookie = cookie.clone();98 let _cancel_guard = _cancel_guard.clone();99 tokio::task::spawn(async move {100 let _cancel_guard = _cancel_guard.clone();101 trace!("conversation task");102 let mut description = format!("{message}\n\n<b>Action id:</b> {action_id}",);103 if let Some(subject) = details.remove("polkit.caller-pid") {104 let _ = write!(description, "\n<b>Caller:</b> ");105 if let Ok(pid) = subject.parse::<u32>() {106 let _ = write!(description, "{}", PidDisplay(pid));107 } else {108 let _ = write!(description, "{}", emphasize("invalid pid"));109 }110 }111 if let Some(subject) = details.remove("polkit.subject-pid") {112 let _ = write!(description, "\n<b>Subject:</b> ");113 if let Ok(pid) = subject.parse::<u32>() {114 let _ = write!(description, "{}", PidDisplay(pid));115 } else {116 let _ = write!(description, "{}", emphasize("invalid pid"));117 }118 }119 let mut prompter = PrependSourcePrompter {120 source: vec![Source(Cow::Borrowed("polkit agent"))],121 description: description.clone(),122 prompter,123 };124125 let identity_displays: Vec<String> =126 identities.iter().map(|v| v.to_string()).collect();127 let identity_displays: Vec<&str> =128 identity_displays.iter().map(|v| v.as_str()).collect();129 debug!("choose identity");130 let choosen_identity = match identity_displays.len() {131 0 => {132 return Err(fdo::Error::AuthFailed(133 "no identity to authenticate as".to_owned(),134 ));135 }136 1 => 0,137 _ => prompter138 .prompt_enum(139 "Identity",140 "Select identity to use for polkit authorization",141 &identity_displays,142 &[],143 )144 .await145 .map_err(prompt_err)?,146 };147 debug!("identity chosen");148149 let _ = write!(150 description,151 "\n<b>Identity:</b> {}",152 identities[choosen_identity as usize]153 );154 prompter.description = description;155156 prompter.source.push(Source(Cow::Borrowed("polkit daemon")));157158 helper159 .help_me(160 &cookie,161 prompter,162 identities[choosen_identity as usize].clone(),163 )164 .await165 .map_err(|e| fdo::Error::Failed(e.to_string()))?;166 167 168169 Ok(())170 })171 };172 self.tasks173 .lock()174 .unwrap()175 .insert(cookie.clone(), task.abort_handle());176 debug!("abort handle stored");177 let _ = _cancel_guard.set(CancelTaskOnDrop {178 tasks: self.tasks.clone(),179 handle: cookie.clone(),180 });181182 let _ = task.await;183184 Ok(())185 }186187 188 async fn cancel_authentication(&self, cookie: &str) -> zbus::fdo::Result<()> {189 debug!("auth cancelled");190 if let Some(abort) = self.tasks.lock().unwrap().remove(cookie) {191 debug!("abort handle found");192 abort.abort();193 }194 195 Ok(())196 }197}198199fn prompt_err(value: remowt_ui_prompt::Error) -> fdo::Error {200 use remowt_ui_prompt::Error;201 match value {202 Error::Cancel => fdo::Error::NoReply("input was cancelled".to_owned()),203 Error::Remote(e) => fdo::Error::NoReply(format!("remote error occured: {e}")),204 Error::InputError(e) => fdo::Error::Failed(e),205 }206}207208const OBJ_PATH: &str = "/org/freedesktop/PolicyKit1/AuthenticationAgent";209210#[derive(Parser)]211enum Opts {212 AskPass {213 prompt: String,214 description: String,215 },216 Editor {217 218 path: String,219 },220 221 Forward {222 223 proto: ForwardProto,224 225 addr: String,226 },227 RealAgent {228 #[arg(long)]229 path: Option<PathBuf>,230 231 #[arg(long)]232 privileged: bool,233 #[arg(long)]234 local: bool,235 },236 LocalAgent,237}238239#[derive(Clone, Copy, clap::ValueEnum)]240enum ForwardProto {241 Tcp,242 Udp,243}244245fn main() -> anyhow::Result<()> {246 tracing_subscriber::fmt()247 .with_writer(io::stderr)248 .without_time()249 .init();250 let opts = Opts::parse();251252 let runtime = Builder::new_current_thread().enable_all().build()?;253254 match opts {255 Opts::AskPass {256 prompt,257 description,258 } => runtime.block_on(askpass::ask(&prompt, description)),259 Opts::LocalAgent => runtime.block_on(main_real()),260 Opts::Editor { path } => runtime.block_on(editor::edit(path)),261 Opts::Forward { proto, addr } => {262 runtime.block_on(editor::forward(matches!(proto, ForwardProto::Udp), addr))263 }264 Opts::RealAgent {265 path,266 privileged,267 local,268 } => runtime.block_on(main_real_agent(path, privileged, local)),269 }270}271async fn main_real() -> anyhow::Result<()> {272 let system_conn = Connection::system().await?;273 let helper = SocketHelper {274 fallback: SuidHelper,275 };276 register_auth_agent(&system_conn, Agent::new(helper, RofiPrompter)).await?;277278 let mut rpc = Rpc::<BifConfig>::new(Address::User);279 serve_prompts(&mut rpc, RofiPrompter);280281 gateway::serve(rpc.clone(), &gateway::local_socket()?).await?;282283 let _keep_alive = (system_conn, rpc);284 pending().await285}286async fn main_real_agent(287 path: Option<PathBuf>,288 privileged: bool,289 local: bool,290) -> anyhow::Result<()> {291 let address = if privileged {292 Address::AgentPrivileged293 } else {294 Address::Agent295 };296 let mut rpc = Rpc::<BifConfig>::new(address);297298 let dialer = Arc::new(TunnelDialer::new());299 Fs::new().register_endpoints(&mut rpc);300 Systemd.register_endpoints(&mut rpc);301 Pty::new(dialer.clone()).register_endpoints(&mut rpc);302 Subprocess::new(dialer.clone()).register_endpoints(&mut rpc);303 NixDaemon::new(dialer.clone()).register_endpoints(&mut rpc);304 IrohTunnel::new(dialer.clone()).register_endpoints(&mut rpc);305 Forward::new(dialer.clone()).register_endpoints(&mut rpc);306307 remowt_plugin::host::serve(&mut rpc);308309 let user_prompter = PromptEndpointsClient::wrap(rpc.remote(Address::User));310311 let helpers = tempfile::Builder::new().prefix("remowt-path.").tempdir()?;312 let gateway_socket = helpers.path().join(gateway::SOCKET_NAME);313 let exe = std::env::current_exe()?;314 let askpass_helper = helpers.path().join("remowt-askpass");315 let editor_helper = helpers.path().join("remowt-editor");316 let forward_helper = helpers.path().join("remowt-forward");317 {318 let script = format!(319 "#!/bin/sh\nexec {} ask-pass \"password\" \"$1\"\n",320 sh_quote(&exe.to_string_lossy())321 );322 fs::write(&askpass_helper, script).await?;323 fs::set_permissions(&askpass_helper, Permissions::from_mode(0o755)).await?;324 }325 {326 let script = format!(327 "#!/bin/sh\nexec {} editor \"$1\"\n",328 sh_quote(&exe.to_string_lossy())329 );330 fs::write(&editor_helper, script).await?;331 fs::set_permissions(&editor_helper, Permissions::from_mode(0o755)).await?;332 }333 {334 let script = format!(335 "#!/bin/sh\nexec {} forward \"$@\"\n",336 sh_quote(&exe.to_string_lossy())337 );338 fs::write(&forward_helper, script).await?;339 fs::set_permissions(&forward_helper, Permissions::from_mode(0o755)).await?;340 }341342 343 unsafe {344 prepend_path(helpers.path());345 std::env::set_var("SUDO_ASKPASS", &askpass_helper);346 std::env::set_var("SSH_ASKPASS", &askpass_helper);347 std::env::set_var("SSH_ASKPASS_REQUIRE", "force");348 std::env::set_var("EDITOR", &editor_helper);349 std::env::set_var("VISUAL", &editor_helper);350 std::env::set_var(gateway::SOCKET_ENV, &gateway_socket);351 }352353 let port = match path {354 Some(path) => from_socket(UnixStream::connect(path).await?),355 None => from_stdio(),356 };357 rpc.add_direct(Address::User, port, bifrostlink::Rtt(0));358359 gateway::serve(rpc.clone(), &gateway_socket).await?;360361 let polkit_conn = if !privileged && !local {362 let conn = Connection::system().await?;363 let helper = SocketHelper {364 fallback: SuidHelper,365 };366 register_auth_agent(&conn, Agent::new(helper, user_prompter)).await?;367 Some(conn)368 } else {369 None370 };371372 let _keep_alive = (helpers, polkit_conn);373 pending().await374}375376async fn register_auth_agent<H, P>(conn: &Connection, agent: Agent<H, P>) -> anyhow::Result<()>377where378 H: Helper + Clone + Send + Sync + 'static,379 P: Prompter + Clone + Send + Sync + 'static,380{381 let proxy = zbus_polkit::policykit1::AuthorityProxy::new(conn).await?;382 conn.object_server().at(OBJ_PATH, agent).await?;383384 let subject = auth_agent_subject()?;385 proxy386 .register_authentication_agent(&subject, "C", OBJ_PATH)387 .await?;388 debug!(kind = subject.subject_kind, "registered polkit agent");389 Ok(())390}391392fn auth_agent_subject() -> anyhow::Result<Subject> {393 let mut details = HashMap::new();394 if let Ok(session_id) = std::env::var("XDG_SESSION_ID") {395 let val: OwnedValue = Str::from(session_id).into();396 details.insert("session-id".to_string(), val);397 return Ok(Subject {398 subject_kind: "unix-session".to_string(),399 subject_details: details,400 });401 }402403 details.insert("pid".to_string(), OwnedValue::from(std::process::id()));404 Ok(Subject {405 subject_kind: "unix-process".to_string(),406 subject_details: details,407 })408}409410fn sh_quote(s: &str) -> String {411 format!("'{}'", s.replace('\'', "'\\''"))412}413414415416417418419unsafe fn prepend_path(dir: &std::path::Path) {420 let value = match std::env::var_os("PATH") {421 Some(existing) => {422 let mut v = dir.as_os_str().to_owned();423 v.push(":");424 v.push(existing);425 v426 }427 None => dir.as_os_str().to_owned(),428 };429 unsafe {430 std::env::set_var("PATH", value);431 }432}
1use std::borrow::Cow;2use std::collections::{BTreeMap, HashMap};3use std::fs::Permissions;4use std::future::pending;5use std::io;6use std::os::unix::fs::PermissionsExt as _;7use std::path::PathBuf;8use std::sync::{Arc, Mutex, OnceLock};910use bifrostlink::Rpc;11use bifrostlink::declarative::RemoteEndpoints;12use bifrostlink_ports::stdio::from_stdio;13use bifrostlink_ports::unix_socket::from_socket;14use clap::Parser;15use remowt_endpoints::{16 forward::Forward, fs::Fs, iroh_tunnel::IrohTunnel, nix_daemon::NixDaemon, pty::Pty,17 subprocess::Subprocess, systemd::Systemd,18};19use remowt_link_shared::iroh_tunnel::TunnelDialer;20use remowt_link_shared::{Address, BifConfig, gateway};21use remowt_polkit_shared::{Identity, PidDisplay, emphasize};22use remowt_ui_prompt::bifrost::{PromptEndpointsClient, serve_prompts};23use remowt_ui_prompt::rofi::RofiPrompter;24use remowt_ui_prompt::{PrependSourcePrompter, Prompter, Source};25use tokio::fs;26use tokio::net::UnixStream;27use tokio::runtime::Builder;28use tokio::task::AbortHandle;29use tracing::{debug, trace};30use zbus::fdo;31use zbus::zvariant::{OwnedValue, Str};32use zbus::{Connection, interface};33use zbus_polkit::policykit1::Subject;3435use self::helper::{Helper, SocketHelper, SuidHelper};3637pub mod askpass;38pub mod editor;39pub mod helper;4041struct CancelTaskOnDrop {42 tasks: Arc<Mutex<HashMap<String, AbortHandle>>>,43 handle: String,44}45impl Drop for CancelTaskOnDrop {46 fn drop(&mut self) {47 debug!("cancel on drop");48 if let Some(task) = self49 .tasks50 .lock()51 .expect("not poisoned")52 .remove(&self.handle)53 {54 task.abort();55 }56 }57}5859struct Agent<H, P> {60 tasks: Arc<Mutex<HashMap<String, AbortHandle>>>,61 helper: H,62 prompter: P,63}64impl<H, P> Agent<H, P> {65 fn new(helper: H, prompter: P) -> Self {66 Agent {67 tasks: Arc::new(Mutex::new(HashMap::new())),68 helper,69 prompter,70 }71 }72}7374#[interface(name = "org.freedesktop.PolicyKit1.AuthenticationAgent")]75impl<H, P> Agent<H, P>76where77 H: Helper + Clone + Send + Sync + 'static,78 P: Prompter + Clone + Send + Sync + 'static,79{80 81 #[allow(clippy::too_many_arguments)]82 async fn begin_authentication(83 &self,84 action_id: String,85 message: String,86 _icon_name: String,87 mut details: BTreeMap<String, String>,88 cookie: String,89 identities: Vec<Identity>,90 ) -> zbus::fdo::Result<()> {91 use std::fmt::Write;92 debug!("begin auth");93 let _cancel_guard = Arc::new(OnceLock::new());94 let task = {95 let helper = self.helper.clone();96 let prompter = self.prompter.clone();97 let cookie = cookie.clone();98 let _cancel_guard = _cancel_guard.clone();99 tokio::task::spawn(async move {100 let _cancel_guard = _cancel_guard.clone();101 trace!("conversation task");102 let mut description = format!("{message}\n\n<b>Action id:</b> {action_id}",);103 if let Some(subject) = details.remove("polkit.caller-pid") {104 let _ = write!(description, "\n<b>Caller:</b> ");105 if let Ok(pid) = subject.parse::<u32>() {106 let _ = write!(description, "{}", PidDisplay(pid));107 } else {108 let _ = write!(description, "{}", emphasize("invalid pid"));109 }110 }111 if let Some(subject) = details.remove("polkit.subject-pid") {112 let _ = write!(description, "\n<b>Subject:</b> ");113 if let Ok(pid) = subject.parse::<u32>() {114 let _ = write!(description, "{}", PidDisplay(pid));115 } else {116 let _ = write!(description, "{}", emphasize("invalid pid"));117 }118 }119 let mut prompter = PrependSourcePrompter {120 source: vec![Source(Cow::Borrowed("polkit agent"))],121 description: description.clone(),122 prompter,123 };124125 let identity_displays: Vec<String> =126 identities.iter().map(|v| v.to_string()).collect();127 let identity_displays: Vec<&str> =128 identity_displays.iter().map(|v| v.as_str()).collect();129 debug!("choose identity");130 let choosen_identity = match identity_displays.len() {131 0 => {132 return Err(fdo::Error::AuthFailed(133 "no identity to authenticate as".to_owned(),134 ));135 }136 1 => 0,137 _ => prompter138 .prompt_enum(139 "Identity",140 "Select identity to use for polkit authorization",141 &identity_displays,142 &[],143 )144 .await145 .map_err(prompt_err)?,146 };147 debug!("identity chosen");148149 let _ = write!(150 description,151 "\n<b>Identity:</b> {}",152 identities[choosen_identity as usize]153 );154 prompter.description = description;155156 prompter.source.push(Source(Cow::Borrowed("polkit daemon")));157158 helper159 .help_me(160 &cookie,161 prompter,162 identities[choosen_identity as usize].clone(),163 )164 .await165 .map_err(|e| fdo::Error::Failed(e.to_string()))?;166 167 168169 Ok(())170 })171 };172 self.tasks173 .lock()174 .unwrap()175 .insert(cookie.clone(), task.abort_handle());176 debug!("abort handle stored");177 let _ = _cancel_guard.set(CancelTaskOnDrop {178 tasks: self.tasks.clone(),179 handle: cookie.clone(),180 });181182 let _ = task.await;183184 Ok(())185 }186187 188 async fn cancel_authentication(&self, cookie: &str) -> zbus::fdo::Result<()> {189 debug!("auth cancelled");190 if let Some(abort) = self.tasks.lock().unwrap().remove(cookie) {191 debug!("abort handle found");192 abort.abort();193 }194 195 Ok(())196 }197}198199fn prompt_err(value: remowt_ui_prompt::Error) -> fdo::Error {200 use remowt_ui_prompt::Error;201 match value {202 Error::Cancel => fdo::Error::NoReply("input was cancelled".to_owned()),203 Error::Remote(e) => fdo::Error::NoReply(format!("remote error occured: {e}")),204 Error::InputError(e) => fdo::Error::Failed(e),205 }206}207208const OBJ_PATH: &str = "/org/freedesktop/PolicyKit1/AuthenticationAgent";209210#[derive(Parser)]211enum Opts {212 AskPass {213 prompt: String,214 description: String,215 },216 Editor {217 218 path: String,219 },220 221 Forward {222 223 proto: ForwardProto,224 225 addr: String,226 },227 RealAgent {228 #[arg(long)]229 path: Option<PathBuf>,230 231 #[arg(long)]232 privileged: bool,233 #[arg(long)]234 local: bool,235 },236 LocalAgent,237}238239#[derive(Clone, Copy, clap::ValueEnum)]240enum ForwardProto {241 Tcp,242 Udp,243}244245fn main() -> anyhow::Result<()> {246 tracing_subscriber::fmt()247 .with_writer(io::stderr)248 .without_time()249 .init();250 let opts = Opts::parse();251252 let runtime = Builder::new_current_thread().enable_all().build()?;253254 match opts {255 Opts::AskPass {256 prompt,257 description,258 } => runtime.block_on(askpass::ask(&prompt, description)),259 Opts::LocalAgent => runtime.block_on(main_real()),260 Opts::Editor { path } => runtime.block_on(editor::edit(path)),261 Opts::Forward { proto, addr } => {262 runtime.block_on(editor::forward(matches!(proto, ForwardProto::Udp), addr))263 }264 Opts::RealAgent {265 path,266 privileged,267 local,268 } => runtime.block_on(main_real_agent(path, privileged, local)),269 }270}271async fn main_real() -> anyhow::Result<()> {272 let system_conn = Connection::system().await?;273 let helper = SocketHelper {274 fallback: SuidHelper,275 };276 let rofi = RofiPrompter::default();277 register_auth_agent(&system_conn, Agent::new(helper, rofi.clone())).await?;278279 let mut rpc = Rpc::<BifConfig>::new(Address::User);280 serve_prompts(&mut rpc, rofi);281282 gateway::serve(rpc.clone(), &gateway::local_socket()?).await?;283284 let _keep_alive = (system_conn, rpc);285 pending().await286}287async fn main_real_agent(288 path: Option<PathBuf>,289 privileged: bool,290 local: bool,291) -> anyhow::Result<()> {292 let address = if privileged {293 Address::AgentPrivileged294 } else {295 Address::Agent296 };297 let mut rpc = Rpc::<BifConfig>::new(address);298299 let dialer = Arc::new(TunnelDialer::new());300 Fs::new().register_endpoints(&mut rpc);301 Systemd.register_endpoints(&mut rpc);302 Pty::new(dialer.clone()).register_endpoints(&mut rpc);303 Subprocess::new(dialer.clone()).register_endpoints(&mut rpc);304 NixDaemon::new(dialer.clone()).register_endpoints(&mut rpc);305 IrohTunnel::new(dialer.clone()).register_endpoints(&mut rpc);306 Forward::new(dialer.clone()).register_endpoints(&mut rpc);307308 remowt_plugin::host::serve(&mut rpc);309310 let user_prompter = PromptEndpointsClient::wrap(rpc.remote(Address::User));311312 let helpers = tempfile::Builder::new().prefix("remowt-path.").tempdir()?;313 let gateway_socket = helpers.path().join(gateway::SOCKET_NAME);314 let exe = std::env::current_exe()?;315 let askpass_helper = helpers.path().join("remowt-askpass");316 let editor_helper = helpers.path().join("remowt-editor");317 let forward_helper = helpers.path().join("remowt-forward");318 {319 let script = format!(320 "#!/bin/sh\nexec {} ask-pass \"password\" \"$1\"\n",321 sh_quote(&exe.to_string_lossy())322 );323 fs::write(&askpass_helper, script).await?;324 fs::set_permissions(&askpass_helper, Permissions::from_mode(0o755)).await?;325 }326 {327 let script = format!(328 "#!/bin/sh\nexec {} editor \"$1\"\n",329 sh_quote(&exe.to_string_lossy())330 );331 fs::write(&editor_helper, script).await?;332 fs::set_permissions(&editor_helper, Permissions::from_mode(0o755)).await?;333 }334 {335 let script = format!(336 "#!/bin/sh\nexec {} forward \"$@\"\n",337 sh_quote(&exe.to_string_lossy())338 );339 fs::write(&forward_helper, script).await?;340 fs::set_permissions(&forward_helper, Permissions::from_mode(0o755)).await?;341 }342343 344 unsafe {345 prepend_path(helpers.path());346 std::env::set_var("SUDO_ASKPASS", &askpass_helper);347 std::env::set_var("SSH_ASKPASS", &askpass_helper);348 std::env::set_var("SSH_ASKPASS_REQUIRE", "force");349 std::env::set_var("EDITOR", &editor_helper);350 std::env::set_var("VISUAL", &editor_helper);351 std::env::set_var(gateway::SOCKET_ENV, &gateway_socket);352 }353354 let port = match path {355 Some(path) => from_socket(UnixStream::connect(path).await?),356 None => from_stdio(),357 };358 rpc.add_direct(Address::User, port, bifrostlink::Rtt(0));359360 gateway::serve(rpc.clone(), &gateway_socket).await?;361362 let polkit_conn = if !privileged && !local {363 let conn = Connection::system().await?;364 let helper = SocketHelper {365 fallback: SuidHelper,366 };367 register_auth_agent(&conn, Agent::new(helper, user_prompter)).await?;368 Some(conn)369 } else {370 None371 };372373 let _keep_alive = (helpers, polkit_conn);374 pending().await375}376377async fn register_auth_agent<H, P>(conn: &Connection, agent: Agent<H, P>) -> anyhow::Result<()>378where379 H: Helper + Clone + Send + Sync + 'static,380 P: Prompter + Clone + Send + Sync + 'static,381{382 let proxy = zbus_polkit::policykit1::AuthorityProxy::new(conn).await?;383 conn.object_server().at(OBJ_PATH, agent).await?;384385 let subject = auth_agent_subject()?;386 proxy387 .register_authentication_agent(&subject, "C", OBJ_PATH)388 .await?;389 debug!(kind = subject.subject_kind, "registered polkit agent");390 Ok(())391}392393fn auth_agent_subject() -> anyhow::Result<Subject> {394 let mut details = HashMap::new();395 if let Ok(session_id) = std::env::var("XDG_SESSION_ID") {396 let val: OwnedValue = Str::from(session_id).into();397 details.insert("session-id".to_string(), val);398 return Ok(Subject {399 subject_kind: "unix-session".to_string(),400 subject_details: details,401 });402 }403404 details.insert("pid".to_string(), OwnedValue::from(std::process::id()));405 Ok(Subject {406 subject_kind: "unix-process".to_string(),407 subject_details: details,408 })409}410411fn sh_quote(s: &str) -> String {412 format!("'{}'", s.replace('\'', "'\\''"))413}414415416417418419420unsafe fn prepend_path(dir: &std::path::Path) {421 let value = match std::env::var_os("PATH") {422 Some(existing) => {423 let mut v = dir.as_os_str().to_owned();424 v.push(":");425 v.push(existing);426 v427 }428 None => dir.as_os_str().to_owned(),429 };430 unsafe {431 std::env::set_var("PATH", value);432 }433}
--- a/remowt/cmds/remowt-ssh/Cargo.toml
+++ b/remowt/cmds/remowt-ssh/Cargo.toml
@@ -8,6 +8,7 @@
[dependencies]
clap = { workspace = true, features = ["derive"] }
tracing-subscriber.workspace = true
+tracing-journald.workspace = true
remowt-link-shared.workspace = true
remowt-client.workspace = true
tokio = { workspace = true, features = [
--- a/remowt/cmds/remowt-ssh/src/main.rs
+++ b/remowt/cmds/remowt-ssh/src/main.rs
@@ -20,6 +20,8 @@
use tokio::io::{AsyncRead, ReadBuf};
use tokio::signal::unix::{signal, SignalKind};
use tracing::debug;
+use tracing_subscriber::prelude::*;
+use tracing_subscriber::EnvFilter;
#[derive(Parser)]
enum Opts {
@@ -45,10 +47,16 @@
#[tokio::main(flavor = "current_thread")]
async fn main() -> anyhow::Result<()> {
- tracing_subscriber::fmt()
- .with_writer(std::io::stderr)
- .without_time()
- .init();
+ // Log to the journal, not stderr: this is an interactive client that puts
+ // the terminal in raw mode, so anything on stderr corrupts the session
+ // output. If the journal socket is unavailable, drop logs rather than fall
+ // back to stderr.
+ if let Ok(journald) = tracing_journald::layer() {
+ tracing_subscriber::registry()
+ .with(EnvFilter::from_default_env())
+ .with(journald.with_syslog_identifier("remowt-ssh".to_owned()))
+ .init();
+ }
let opts = Opts::parse();
let bundle = AgentBundle::from_dir(agents_dir()?)?;
--- a/remowt/crates/remowt-ui-prompt/Cargo.toml
+++ b/remowt/crates/remowt-ui-prompt/Cargo.toml
@@ -12,5 +12,5 @@
remowt-link-shared.workspace = true
serde.workspace = true
thiserror.workspace = true
-tokio = { workspace = true, features = ["io-util", "macros", "process", "rt"] }
+tokio = { workspace = true, features = ["io-util", "macros", "process", "rt", "sync"] }
tracing.workspace = true
--- a/remowt/crates/remowt-ui-prompt/src/auto.rs
+++ b/remowt/crates/remowt-ui-prompt/src/auto.rs
@@ -24,7 +24,7 @@
};
Self {
remote,
- fallback: RofiPrompter,
+ fallback: RofiPrompter::default(),
}
}
--- a/remowt/crates/remowt-ui-prompt/src/rofi.rs
+++ b/remowt/crates/remowt-ui-prompt/src/rofi.rs
@@ -1,13 +1,18 @@
use std::process::Stdio;
+use std::sync::Arc;
use tokio::io::AsyncWriteExt;
use tokio::process::Command;
+use tokio::sync::Mutex;
use tracing::trace;
use crate::{Error, Prompter, Result, Source};
-#[derive(Clone)]
-pub struct RofiPrompter;
+#[derive(Clone, Default)]
+pub struct RofiPrompter {
+ // Rofi can't run concurrently; serialize invocations.
+ lock: Arc<Mutex<()>>,
+}
fn fixup_prompt(prompt: &str) -> &str {
// Rofi always appends such suffix
@@ -27,6 +32,7 @@
source: &[Source],
) -> Result<u32> {
trace!("rofi radio");
+ let _guard = self.lock.lock().await;
let mut cmd = rofi_command();
let mesg = if source.is_empty() {
description.to_owned()
@@ -113,6 +119,7 @@
source: &[Source],
) -> Result<String> {
trace!("rofi text");
+ let _guard = self.lock.lock().await;
let mut cmd = rofi_command();
let mesg = if source.is_empty() {
description.to_owned()
@@ -162,6 +169,7 @@
async fn display_text(&self, error: bool, description: &str, source: &[Source]) -> Result<()> {
trace!("rofi display");
+ let _guard = self.lock.lock().await;
let mut cmd = rofi_command();
let mut mesg = if source.is_empty() {
description.to_owned()
@@ -206,7 +214,7 @@
#[ignore = "interactive"]
async fn test() {
let prompter = PrependSourcePrompter {
- prompter: RofiPrompter,
+ prompter: RofiPrompter::default(),
description: "test".to_owned(),
source: vec![Source(Cow::Borrowed("ssh"))],
};