diff --git a/Cargo.lock b/Cargo.lock index a1077fb..f6a2a65 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1729,8 +1729,10 @@ dependencies = [ "lib", "n0-error", "rand 0.9.4", + "rustls", "sentry", "serde", + "serde_json", "serde_yml", "tokio", "tokio-util", @@ -5238,6 +5240,7 @@ dependencies = [ "iroh-tickets", "k8s-openapi", "kube", + "libc", "log", "n0-error", "n0-future", diff --git a/cli/Cargo.toml b/cli/Cargo.toml index 51d3ccd..554f124 100644 --- a/cli/Cargo.toml +++ b/cli/Cargo.toml @@ -21,4 +21,6 @@ hickory-proto = "0.25.2" iroh-base.workspace = true z32 = "1.0.3" rand.workspace = true -sentry.workspace = true \ No newline at end of file +sentry.workspace = true +serde_json.workspace = true +rustls.workspace = true diff --git a/cli/src/cloud.rs b/cli/src/cloud.rs new file mode 100644 index 0000000..5de1772 --- /dev/null +++ b/cli/src/cloud.rs @@ -0,0 +1,396 @@ +//! Login, context, agent, and tunnel CLI commands. + +use std::time::Duration; + +use lib::agent::{self, AgentClient, AgentStatus, CreateTunnelRequest, running_agent_info}; +use lib::datum_cloud::{ApiEnv, DatumCloudClient, LoginState, resolve_selected_context}; +use lib::{Repo, SelectedContext}; +use n0_error::{Result, StdResultExt}; +use serde::Serialize; +use tokio_util::sync::CancellationToken; + +use crate::{AgentCommands, ContextCommands, TunnelCommands}; + +pub async fn login(repo: Repo, force: bool, json: bool) -> Result<()> { + let datum = DatumCloudClient::with_repo(ApiEnv::default(), repo).await?; + if force { + datum.auth().logout().await?; + } + datum.auth().login().await?; + let auth = datum.auth_state(); + let auth = auth.get()?; + let payload = LoginOutput { + email: auth.profile.email.clone(), + user_id: auth.profile.user_id.clone(), + }; + if json { + print_json(&payload)?; + } else { + println!("Logged in as {}", payload.email); + } + Ok(()) +} + +pub async fn logout(repo: Repo, json: bool) -> Result<()> { + let datum = DatumCloudClient::with_repo(ApiEnv::default(), repo).await?; + datum.auth().logout().await?; + if json { + print_json(&serde_json::json!({ "logged_in": false }))?; + } else { + println!("Logged out."); + } + Ok(()) +} + +pub async fn status(repo: Repo, json: bool) -> Result<()> { + let payload = collect_status(&repo).await?; + if json { + print_json(&payload)?; + return Ok(()); + } + + match &payload.email { + Some(email) => println!("Logged in as {email} ({})", payload.login_state), + None => println!("Not logged in. Run `datum-connect login`."), + } + match &payload.context { + Some(ctx) => println!("Project: {} / {}", ctx.org_name, ctx.project_name), + None => println!("No project selected. Run `datum-connect context set --project `."), + } + match &payload.agent { + agent if agent.running => { + println!( + "Agent: running (pid {}, endpoint {})", + agent.pid.unwrap_or_default(), + agent.endpoint_id.as_deref().unwrap_or("-") + ); + } + _ => println!("Agent: not running. Run `datum-connect agent start`."), + } + Ok(()) +} + +async fn collect_status(repo: &Repo) -> Result { + let datum = DatumCloudClient::with_repo(ApiEnv::default(), repo.clone()).await?; + let auth = datum.auth_state(); + let profile = auth.get().ok(); + let agent = match AgentClient::connect(repo) { + Ok(client) => match client.status().await { + Ok(status) => AgentStatusOutput::from_running(status), + Err(_) => AgentStatusOutput::from_info(running_agent_info(repo)?), + }, + Err(_) => AgentStatusOutput::from_info(running_agent_info(repo)?), + }; + Ok(StatusOutput { + logged_in: profile.is_some(), + login_state: login_state_name(datum.login_state()).to_string(), + email: profile.map(|p| p.profile.email.clone()), + user_id: profile.map(|p| p.profile.user_id.clone()), + context: datum.selected_context(), + agent, + }) +} + +pub async fn context(repo: Repo, command: ContextCommands) -> Result<()> { + match command { + ContextCommands::Show { json } => context_show(repo, json).await, + ContextCommands::List { json } => context_list(repo, json).await, + ContextCommands::Set { project, org, json } => context_set(repo, project, org, json).await, + } +} + +async fn context_show(repo: Repo, json: bool) -> Result<()> { + let datum = require_logged_in(repo).await?; + let selected = datum.selected_context(); + if json { + print_json(&ContextShowOutput { + context: selected.clone(), + })?; + return Ok(()); + } + match selected { + Some(ctx) => { + println!("{} / {}", ctx.org_name, ctx.project_name); + println!(" org: {}", ctx.org_id); + println!(" project: {}", ctx.project_id); + } + None => println!("No project selected. Run `datum-connect context set --project `."), + } + Ok(()) +} + +async fn context_list(repo: Repo, json: bool) -> Result<()> { + let datum = require_logged_in(repo).await?; + let orgs = datum.orgs_and_projects().await?; + if json { + print_json(&orgs)?; + return Ok(()); + } + if orgs.is_empty() { + println!("No organizations found."); + return Ok(()); + } + let selected = datum.selected_context(); + for org in &orgs { + println!("{} ({})", org.org.display_name, org.org.resource_id); + if org.projects.is_empty() { + println!(" (no projects)"); + continue; + } + for project in &org.projects { + let marker = if selected + .as_ref() + .is_some_and(|ctx| ctx.project_id == project.resource_id) + { + "*" + } else { + " " + }; + println!( + " {marker} {} ({})", + project.display_name, project.resource_id + ); + } + } + Ok(()) +} + +async fn context_set(repo: Repo, project: String, org: Option, json: bool) -> Result<()> { + let datum = require_logged_in(repo).await?; + let orgs = datum.orgs_and_projects().await?; + let ctx = resolve_selected_context(&orgs, &project, org.as_deref())?; + datum.set_selected_context(Some(ctx.clone())).await?; + if json { + print_json(&ctx)?; + } else { + println!("Selected {} / {}", ctx.org_name, ctx.project_name); + } + Ok(()) +} + +pub async fn agent(repo: Repo, command: AgentCommands) -> Result<()> { + match command { + AgentCommands::Start => agent_start(repo).await, + AgentCommands::Stop => { + agent::stop_agent(&repo).await?; + println!("Agent stopped."); + Ok(()) + } + AgentCommands::Status { json } => { + let info = running_agent_info(&repo)?; + if json { + print_json(&AgentStatusOutput::from_info(info))?; + return Ok(()); + } + match info { + Some(info) => println!( + "Agent running (pid {}, endpoint {}, control http://127.0.0.1:{})", + info.pid, info.endpoint_id, info.port + ), + None => println!("Agent is not running."), + } + Ok(()) + } + } +} + +async fn agent_start(repo: Repo) -> Result<()> { + if let Some(info) = running_agent_info(&repo)? { + n0_error::bail_any!( + "Agent is already running (pid {}). Stop it with `datum-connect agent stop`.", + info.pid + ); + } + + let shutdown = CancellationToken::new(); + let shutdown_for_signal = shutdown.clone(); + tokio::spawn(async move { + shutdown_signal().await; + shutdown_for_signal.cancel(); + }); + + println!("Starting agent…"); + agent::run_agent(repo, shutdown).await?; + println!("Agent stopped."); + Ok(()) +} + +pub async fn tunnel(repo: Repo, command: TunnelCommands) -> Result<()> { + match command { + TunnelCommands::Create { + label, + endpoint, + wait_hostname, + timeout, + json, + } => { + let client = connect_or_start(&repo).await?; + let tunnel = client + .create_tunnel(&CreateTunnelRequest { + label, + endpoint, + wait_hostname, + timeout_secs: Some(Duration::from(timeout).as_secs()), + }) + .await?; + if json { + print_json(&tunnel)?; + } else { + println!("{} -> {}", tunnel.label, tunnel.endpoint); + match &tunnel.url { + Some(url) => println!("{url}"), + None => println!("(hostname pending; tunnel id {})", tunnel.id), + } + } + Ok(()) + } + TunnelCommands::List { json } => { + let client = AgentClient::connect(&repo)?; + let tunnels = client.list_tunnels().await?; + if json { + print_json(&tunnels)?; + } else if tunnels.is_empty() { + println!("No tunnels."); + } else { + for tunnel in tunnels { + let url = tunnel.url.as_deref().unwrap_or("(hostname pending)"); + println!( + "{} {} {} -> {}", + tunnel.id, tunnel.label, url, tunnel.endpoint + ); + } + } + Ok(()) + } + TunnelCommands::Get { id, json } => { + let client = AgentClient::connect(&repo)?; + let tunnel = client.get_tunnel(&id).await?; + if json { + print_json(&tunnel)?; + } else { + println!("{} ({})", tunnel.label, tunnel.id); + println!(" endpoint: {}", tunnel.endpoint); + match &tunnel.url { + Some(url) => println!(" url: {url}"), + None => println!(" url: (hostname pending)"), + } + } + Ok(()) + } + TunnelCommands::Delete { id, json } => { + let client = AgentClient::connect(&repo)?; + let outcome = client.delete_tunnel(&id).await?; + if json { + print_json(&outcome)?; + } else { + println!("Deleted tunnel {}", outcome.id); + } + Ok(()) + } + } +} + +async fn connect_or_start(repo: &Repo) -> Result { + if agent::running_agent_info(repo)?.is_none() { + eprintln!("Starting datum-connect agent…"); + } + agent::ensure_agent(repo).await +} + +async fn require_logged_in(repo: Repo) -> Result { + let datum = DatumCloudClient::with_repo(ApiEnv::default(), repo).await?; + if datum.auth_state().get().is_err() { + n0_error::bail_any!("Not logged in. Run `datum-connect login`."); + } + Ok(datum) +} + +fn login_state_name(state: LoginState) -> &'static str { + match state { + LoginState::Missing => "missing", + LoginState::NeedsRefresh => "needs_refresh", + LoginState::Valid => "valid", + } +} + +fn print_json(value: &T) -> Result<()> { + println!( + "{}", + serde_json::to_string_pretty(value).std_context("serializing json")? + ); + Ok(()) +} + +async fn shutdown_signal() { + let ctrl_c = tokio::signal::ctrl_c(); + #[cfg(unix)] + { + let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("install SIGTERM handler"); + tokio::select! { + _ = ctrl_c => {} + _ = sigterm.recv() => {} + } + } + #[cfg(not(unix))] + { + let _ = ctrl_c.await; + } +} + +#[derive(Serialize)] +struct LoginOutput { + email: String, + user_id: String, +} + +#[derive(Serialize)] +struct StatusOutput { + logged_in: bool, + login_state: String, + email: Option, + user_id: Option, + context: Option, + agent: AgentStatusOutput, +} + +#[derive(Serialize)] +struct ContextShowOutput { + context: Option, +} + +#[derive(Serialize)] +struct AgentStatusOutput { + running: bool, + pid: Option, + port: Option, + endpoint_id: Option, +} + +impl AgentStatusOutput { + fn from_running(status: AgentStatus) -> Self { + Self { + running: true, + pid: Some(status.pid), + port: Some(status.port), + endpoint_id: Some(status.endpoint_id), + } + } + + fn from_info(info: Option) -> Self { + match info { + Some(info) => Self { + running: true, + pid: Some(info.pid), + port: Some(info.port), + endpoint_id: Some(info.endpoint_id), + }, + None => Self { + running: false, + pid: None, + port: None, + endpoint_id: None, + }, + } + } +} diff --git a/cli/src/main.rs b/cli/src/main.rs index 4377cae..6de9992 100644 --- a/cli/src/main.rs +++ b/cli/src/main.rs @@ -1,5 +1,6 @@ //! Command line arguments. use clap::{Parser, Subcommand}; +mod cloud; mod dns_dev; mod tunnel_dev; @@ -9,7 +10,7 @@ use lib::{ }; use std::{net::SocketAddr, path::PathBuf}; use tracing::info; -use tracing_subscriber::prelude::*; +use tracing_subscriber::{EnvFilter, prelude::*}; /// Datum Connect Agent #[derive(Parser, Debug)] @@ -22,6 +23,34 @@ struct Args { #[derive(Subcommand, Debug)] enum Commands { + /// Log in to Datum Cloud via the browser. + Login { + /// Re-run the browser login even if a valid session exists. + #[clap(long)] + force: bool, + #[clap(long)] + json: bool, + }, + /// Clear the stored Datum Cloud session. + Logout { + #[clap(long)] + json: bool, + }, + /// Show login, selected project, and agent status. + Status { + #[clap(long)] + json: bool, + }, + /// Select the Datum Cloud organization and project used for tunnels. + #[clap(subcommand)] + Context(ContextCommands), + /// Run or inspect the headless listener that serves tunnels. + #[clap(subcommand)] + Agent(AgentCommands), + /// Create and manage public tunnels through the running agent. + #[clap(subcommand)] + Tunnel(TunnelCommands), + /// Start a tunnel server that exposes configured local services through the Datum gateway. Serve, @@ -43,6 +72,82 @@ enum Commands { Add(AddCommands), } +#[derive(Subcommand, Debug)] +pub(crate) enum ContextCommands { + /// Show the currently selected organization and project. + Show { + #[clap(long)] + json: bool, + }, + /// List organizations and projects. + List { + #[clap(long)] + json: bool, + }, + /// Select a project by id or name. + Set { + /// Project resource id or display name. + #[clap(long)] + project: String, + /// Organization resource id or display name (required when the project name is ambiguous). + #[clap(long)] + org: Option, + #[clap(long)] + json: bool, + }, +} + +#[derive(Subcommand, Debug)] +pub(crate) enum AgentCommands { + /// Start the agent in the foreground. + Start, + /// Stop a running agent. + Stop, + /// Show whether the agent is running. + Status { + #[clap(long)] + json: bool, + }, +} + +#[derive(Subcommand, Debug)] +pub(crate) enum TunnelCommands { + /// Create a tunnel to a local HTTP endpoint. + Create { + /// Display name for the tunnel. + #[clap(long)] + label: String, + /// Local host:port or URL, e.g. 127.0.0.1:4123. + #[clap(long)] + endpoint: String, + /// Block until Datum assigns a public hostname. + #[clap(long)] + wait_hostname: bool, + /// How long to wait for a hostname when --wait-hostname is set. + #[clap(long, default_value = "30s")] + timeout: humantime::Duration, + #[clap(long)] + json: bool, + }, + /// List tunnels in the selected project. + List { + #[clap(long)] + json: bool, + }, + /// Show one tunnel. + Get { + id: String, + #[clap(long)] + json: bool, + }, + /// Delete a tunnel. + Delete { + id: String, + #[clap(long)] + json: bool, + }, +} + #[derive(Debug, clap::Parser)] enum AddCommands { TcpProxy { @@ -134,6 +239,16 @@ pub struct ConnectArgs { #[tokio::main] async fn main() -> n0_error::Result<()> { + // Required before any TLS use (kube, reqwest, iroh). The GUI does the same + // in ui/src/main.rs; without it the agent panics in rustls when creating a tunnel. + rustls::crypto::ring::default_provider() + .install_default() + .expect("rustls default crypto provider"); + + if lib::agent::wants_headless_agent() { + return lib::agent::run_headless_agent_from_args().await; + } + // Load .env first so any process-env-driven config is visible to the rest // of init. We keep the load result so we can log it *after* tracing is up. let dotenv_path = dotenv::dotenv().ok(); @@ -164,8 +279,10 @@ async fn main() -> n0_error::Result<()> { } }); + let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); tracing_subscriber::registry() - .with(tracing_subscriber::fmt::layer()) + .with(filter) + .with(tracing_subscriber::fmt::layer().with_writer(std::io::stderr)) .with(sentry_layer) .init(); @@ -179,6 +296,12 @@ async fn main() -> n0_error::Result<()> { let repo = Repo::open_or_create(path).await?; match args.command { + Commands::Login { force, json } => cloud::login(repo, force, json).await?, + Commands::Logout { json } => cloud::logout(repo, json).await?, + Commands::Status { json } => cloud::status(repo, json).await?, + Commands::Context(command) => cloud::context(repo, command).await?, + Commands::Agent(command) => cloud::agent(repo, command).await?, + Commands::Tunnel(command) => cloud::tunnel(repo, command).await?, Commands::List => { let datum = DatumCloudClient::with_repo(ApiEnv::default(), repo.clone()).await?; let orgs = datum.orgs_and_projects().await?; diff --git a/lib/Cargo.toml b/lib/Cargo.toml index a2bfc4d..3d677bf 100644 --- a/lib/Cargo.toml +++ b/lib/Cargo.toml @@ -54,6 +54,9 @@ kube = { version = "2.0.1", default-features = false, features = ["client", "der gateway-api = "0.19.0" gethostname = "1.1.0" +[target.'cfg(unix)'.dependencies] +libc = "0.2" + [dev-dependencies] http-body-util = "0.1.3" hyper = { version = "1.8.1", features = ["full"] } diff --git a/lib/src/agent.rs b/lib/src/agent.rs new file mode 100644 index 0000000..fa2ed41 --- /dev/null +++ b/lib/src/agent.rs @@ -0,0 +1,1059 @@ +//! Headless agent: one ListenNode plus a local HTTP control API. +//! +//! Tunnel create/delete must run in the same process as the listener, so the CLI +//! talks to this agent over loopback HTTP instead of constructing its own node. + +use std::fs::OpenOptions; +use std::io::Write; +use std::path::PathBuf; +use std::process::Stdio; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use axum::extract::{Path, State}; +use axum::http::{HeaderMap, StatusCode}; +use axum::middleware::{self, Next}; +use axum::response::{IntoResponse, Response}; +use axum::routing::{get, put}; +use axum::{Json, Router}; +use n0_error::{Result, StdResultExt}; +use rand::Rng; +use serde::{Deserialize, Serialize}; +use tokio::net::TcpListener; +use tokio_util::sync::CancellationToken; +use tracing::{info, warn}; + +use crate::datum_cloud::{ApiEnv, DatumCloudClient, LoginState, NotLoggedIn, UserProfile}; +use crate::http_user_agent::datum_http_user_agent; +use crate::tunnel_activity::metrics_bytes_for_tunnel; +use crate::tunnels::{TunnelCreateQuota, TunnelService, TunnelSummary}; +use crate::{HeartbeatAgent, ListenNode, Repo, SelectedContext}; + +/// Hidden flag both the CLI and the desktop app honor so either binary can +/// spawn the same detached listener. +pub const HEADLESS_AGENT_FLAG: &str = "--headless-agent"; + +const DEFAULT_HOSTNAME_TIMEOUT: Duration = Duration::from_secs(30); +const HOSTNAME_POLL_INTERVAL: Duration = Duration::from_secs(1); +const AGENT_READY_POLL: Duration = Duration::from_millis(200); + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AgentInfo { + pub pid: u32, + pub port: u16, + pub token: String, + pub endpoint_id: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AgentStatus { + pub pid: u32, + pub port: u16, + pub endpoint_id: String, + pub logged_in: bool, + pub login_state: String, + pub email: Option, + pub context: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CreateTunnelRequest { + pub label: String, + pub endpoint: String, + #[serde(default)] + pub wait_hostname: bool, + #[serde(default)] + pub timeout_secs: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TunnelView { + pub id: String, + pub label: String, + pub endpoint: String, + pub url: Option, + pub hostnames: Vec, + pub enabled: bool, + pub accepted: bool, + pub programmed: bool, +} + +impl From<&TunnelSummary> for TunnelView { + fn from(summary: &TunnelSummary) -> Self { + Self { + id: summary.id.clone(), + label: summary.label.clone(), + endpoint: summary.endpoint.clone(), + url: summary.public_url(), + hostnames: summary.hostnames.clone(), + enabled: summary.enabled, + accepted: summary.accepted, + programmed: summary.programmed, + } + } +} + +impl From for TunnelSummary { + fn from(view: TunnelView) -> Self { + Self { + id: view.id, + label: view.label, + endpoint: view.endpoint, + hostnames: view.hostnames, + enabled: view.enabled, + accepted: view.accepted, + programmed: view.programmed, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct UpdateTunnelRequest { + pub label: String, + pub endpoint: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SetEnabledRequest { + pub enabled: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TunnelMetric { + pub id: String, + pub bytes_from_origin: u64, + pub bytes_to_origin: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DeleteTunnelResponse { + pub id: String, + pub project_id: String, + pub connector_deleted: bool, +} + +#[derive(Deserialize)] +struct ErrorBody { + error: String, +} + +struct ApiError { + status: StatusCode, + message: String, + tunnel: Option, +} + +impl ApiError { + fn new(status: StatusCode, message: impl Into) -> Self { + Self { + status, + message: message.into(), + tunnel: None, + } + } + + fn with_tunnel(mut self, tunnel: TunnelView) -> Self { + self.tunnel = Some(tunnel); + self + } +} + +impl IntoResponse for ApiError { + fn into_response(self) -> Response { + let mut body = serde_json::json!({ "error": self.message }); + if let Some(tunnel) = self.tunnel + && let Ok(value) = serde_json::to_value(tunnel) + { + body["tunnel"] = value; + } + (self.status, Json(body)).into_response() + } +} + +#[derive(Clone)] +struct AgentState { + token: String, + info: AgentInfo, + datum: DatumCloudClient, + listen: ListenNode, + tunnels: TunnelService, + heartbeat: HeartbeatAgent, +} + +pub fn is_pid_alive(pid: u32) -> bool { + if pid == 0 { + return false; + } + #[cfg(unix)] + { + // SAFETY: signal 0 does not deliver a signal; it only checks whether the pid exists. + let rc = unsafe { libc::kill(pid as i32, 0) }; + if rc == 0 { + return true; + } + std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) + } + #[cfg(windows)] + { + windows_is_pid_alive(pid) + } + #[cfg(not(any(unix, windows)))] + { + false + } +} + +#[cfg(windows)] +fn windows_is_pid_alive(pid: u32) -> bool { + const PROCESS_QUERY_LIMITED_INFORMATION: u32 = 0x1000; + const STILL_ACTIVE: u32 = 259; + extern "system" { + fn OpenProcess(access: u32, inherit: i32, pid: u32) -> isize; + fn CloseHandle(handle: isize) -> i32; + fn GetExitCodeProcess(handle: isize, code: *mut u32) -> i32; + } + unsafe { + // SAFETY: OpenProcess/GetExitCodeProcess/CloseHandle are used only to query + // whether a numeric pid still refers to a live process. + let handle = OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid); + if handle == 0 { + return false; + } + let mut code = 0u32; + let ok = GetExitCodeProcess(handle, &mut code); + CloseHandle(handle); + ok != 0 && code == STILL_ACTIVE + } +} + +fn kill_pid(pid: u32, force: bool) -> Result<()> { + #[cfg(unix)] + { + let signal = if force { libc::SIGKILL } else { libc::SIGTERM }; + // SAFETY: pid is a recorded agent process id, not the current process. + let rc = unsafe { libc::kill(pid as i32, signal) }; + if rc == 0 { + Ok(()) + } else { + Err(std::io::Error::last_os_error()).std_context("signaling agent process") + } + } + #[cfg(windows)] + { + let mut cmd = std::process::Command::new("taskkill"); + cmd.args(["/PID", &pid.to_string()]); + if force { + cmd.arg("/F"); + } + let status = cmd + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .status() + .std_context("taskkill")?; + if status.success() { + Ok(()) + } else { + n0_error::bail_any!("taskkill exited with {status}"); + } + } + #[cfg(not(any(unix, windows)))] + { + let _ = (pid, force); + n0_error::bail_any!("stopping the agent is not supported on this platform"); + } +} + +pub fn read_agent_info(repo: &Repo) -> Result> { + let path = repo.agent_info_path(); + if !path.exists() { + return Ok(None); + } + let data = std::fs::read_to_string(&path).std_context("failed to read agent.json")?; + let info: AgentInfo = serde_json::from_str(&data).std_context("failed to parse agent.json")?; + Ok(Some(info)) +} + +pub fn remove_agent_info(repo: &Repo) -> Result<()> { + let path = repo.agent_info_path(); + if path.exists() { + std::fs::remove_file(&path).std_context("failed to remove agent.json")?; + } + Ok(()) +} + +fn write_agent_info_exclusive(repo: &Repo, info: &AgentInfo) -> Result<()> { + let path = repo.agent_info_path(); + let json = serde_json::to_string_pretty(info).anyerr()?; + match OpenOptions::new().write(true).create_new(true).open(&path) { + Ok(mut file) => { + file.write_all(json.as_bytes()) + .std_context("failed to write agent.json")?; + Ok(()) + } + Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => { + if let Some(existing) = read_agent_info(repo)? + && is_pid_alive(existing.pid) + { + n0_error::bail_any!( + "Agent is already running (pid {}). Stop it with `datum-connect agent stop`.", + existing.pid + ); + } + std::fs::remove_file(&path).ok(); + std::fs::write(&path, json.as_bytes()).std_context("failed to write agent.json")?; + Ok(()) + } + Err(err) => Err(err).std_context("failed to create agent.json"), + } +} + +/// Load agent.json only if the recorded process is still alive. +pub fn running_agent_info(repo: &Repo) -> Result> { + let Some(info) = read_agent_info(repo)? else { + return Ok(None); + }; + if is_pid_alive(info.pid) { + Ok(Some(info)) + } else { + let _ = remove_agent_info(repo); + Ok(None) + } +} + +fn new_token() -> String { + rand::rng() + .sample_iter(&rand::distr::Alphanumeric) + .take(32) + .map(char::from) + .collect() +} + +fn login_state_name(state: LoginState) -> &'static str { + match state { + LoginState::Missing => "missing", + LoginState::NeedsRefresh => "needs_refresh", + LoginState::Valid => "valid", + } +} + +fn profile_email(profile: Option<&UserProfile>) -> Option { + profile.map(|p| p.email.clone()) +} + +/// Run the agent until `shutdown` is cancelled. +pub async fn run_agent(repo: Repo, shutdown: CancellationToken) -> Result<()> { + if let Some(existing) = running_agent_info(&repo)? { + n0_error::bail_any!( + "Agent is already running (pid {}). Stop it with `datum-connect agent stop`.", + existing.pid + ); + } + + let datum = DatumCloudClient::with_repo(ApiEnv::default(), repo.clone()).await?; + let listen = ListenNode::new(repo.clone()).await?; + let heartbeat = HeartbeatAgent::new(datum.clone(), listen.clone()); + heartbeat.start().await; + let tunnels = TunnelService::new(datum.clone(), listen.clone()); + + let listener = TcpListener::bind("127.0.0.1:0") + .await + .std_context("failed to bind agent control API")?; + let port = listener + .local_addr() + .std_context("failed to read agent bind address")? + .port(); + + let info = AgentInfo { + pid: std::process::id(), + port, + token: new_token(), + endpoint_id: listen.endpoint_id().to_string(), + }; + write_agent_info_exclusive(&repo, &info)?; + let _guard = AgentFileGuard { + path: repo.agent_info_path(), + }; + + let state = Arc::new(AgentState { + token: info.token.clone(), + info: info.clone(), + datum, + listen, + tunnels, + heartbeat, + }); + + let app = Router::new() + .route("/status", get(status)) + .route("/quota", get(quota)) + .route("/metrics", get(metrics)) + .route("/tunnels", get(list_tunnels).post(create_tunnel)) + .route( + "/tunnels/:id", + get(get_tunnel).patch(update_tunnel).delete(delete_tunnel), + ) + .route("/tunnels/:id/enabled", put(set_enabled)) + .layer(middleware::from_fn_with_state(state.clone(), require_token)) + .with_state(state); + + info!( + port, + endpoint_id = %info.endpoint_id, + "datum-connect agent listening" + ); + + axum::serve(listener, app) + .with_graceful_shutdown(shutdown.cancelled_owned()) + .await + .std_context("agent control API failed")?; + Ok(()) +} + +struct AgentFileGuard { + path: std::path::PathBuf, +} + +impl Drop for AgentFileGuard { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + } +} + +async fn require_token( + State(state): State>, + headers: HeaderMap, + request: axum::extract::Request, + next: Next, +) -> Response { + let expected = format!("Bearer {}", state.token); + let authorized = headers + .get(axum::http::header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value == expected) + || headers + .get("x-datum-agent-token") + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value == state.token); + if authorized { + next.run(request).await + } else { + ApiError::new(StatusCode::UNAUTHORIZED, "unauthorized").into_response() + } +} + +async fn status(State(state): State>) -> Json { + let _ = state.datum.reload_selected_context().await; + Json(agent_status(&state)) +} + +fn agent_status(state: &AgentState) -> AgentStatus { + let auth = state.datum.auth_state(); + let profile = auth.get().ok().map(|s| &s.profile); + AgentStatus { + pid: state.info.pid, + port: state.info.port, + endpoint_id: state.info.endpoint_id.clone(), + logged_in: auth.get().is_ok(), + login_state: login_state_name(state.datum.login_state()).to_string(), + email: profile_email(profile), + context: state.datum.selected_context(), + } +} + +async fn sync_selected_context(state: &AgentState) -> std::result::Result<(), ApiError> { + state + .datum + .reload_selected_context() + .await + .map_err(map_tunnel_err)?; + Ok(()) +} + +async fn list_tunnels( + State(state): State>, +) -> std::result::Result>, ApiError> { + sync_selected_context(&state).await?; + let tunnels = state.tunnels.list_active().await.map_err(map_tunnel_err)?; + Ok(Json(tunnels.iter().map(TunnelView::from).collect())) +} + +async fn get_tunnel( + State(state): State>, + Path(id): Path, +) -> std::result::Result, ApiError> { + sync_selected_context(&state).await?; + match state + .tunnels + .get_active(&id) + .await + .map_err(map_tunnel_err)? + { + Some(tunnel) => Ok(Json(TunnelView::from(&tunnel))), + None => Err(ApiError::new( + StatusCode::NOT_FOUND, + format!("Tunnel `{id}` not found"), + )), + } +} + +async fn create_tunnel( + State(state): State>, + Json(req): Json, +) -> std::result::Result, ApiError> { + if req.label.trim().is_empty() { + return Err(ApiError::new(StatusCode::BAD_REQUEST, "label is required")); + } + if req.endpoint.trim().is_empty() { + return Err(ApiError::new( + StatusCode::BAD_REQUEST, + "endpoint is required", + )); + } + + sync_selected_context(&state).await?; + let mut summary = state + .tunnels + .create_active(req.label.trim(), req.endpoint.trim()) + .await + .map_err(map_tunnel_err)?; + + if let Some(ctx) = state.datum.selected_context() { + state.heartbeat.register_project(ctx.project_id).await; + } + + if req.wait_hostname && summary.hostnames.is_empty() { + let timeout = req + .timeout_secs + .map(Duration::from_secs) + .unwrap_or(DEFAULT_HOSTNAME_TIMEOUT); + summary = wait_for_hostname(&state.tunnels, &summary.id, timeout) + .await + .map_err(map_tunnel_err)?; + if summary.hostnames.is_empty() { + return Err( + ApiError::new( + StatusCode::GATEWAY_TIMEOUT, + "Timed out waiting for a public hostname. The tunnel was created; delete it with `datum-connect tunnel delete` or poll `datum-connect tunnel get`.", + ) + .with_tunnel(TunnelView::from(&summary)), + ); + } + } + + Ok(Json(TunnelView::from(&summary))) +} + +async fn delete_tunnel( + State(state): State>, + Path(id): Path, +) -> std::result::Result, ApiError> { + sync_selected_context(&state).await?; + if state + .tunnels + .get_active(&id) + .await + .map_err(map_tunnel_err)? + .is_none() + { + return Err(ApiError::new( + StatusCode::NOT_FOUND, + format!("Tunnel `{id}` not found"), + )); + } + + let outcome = state + .tunnels + .delete_active(&id) + .await + .map_err(map_tunnel_err)?; + if outcome.connector_deleted { + state + .heartbeat + .deregister_project(&outcome.project_id) + .await; + } + Ok(Json(DeleteTunnelResponse { + id, + project_id: outcome.project_id, + connector_deleted: outcome.connector_deleted, + })) +} + +async fn update_tunnel( + State(state): State>, + Path(id): Path, + Json(req): Json, +) -> std::result::Result, ApiError> { + if req.label.trim().is_empty() { + return Err(ApiError::new(StatusCode::BAD_REQUEST, "label is required")); + } + if req.endpoint.trim().is_empty() { + return Err(ApiError::new( + StatusCode::BAD_REQUEST, + "endpoint is required", + )); + } + sync_selected_context(&state).await?; + let summary = state + .tunnels + .update_active(&id, req.label.trim(), req.endpoint.trim()) + .await + .map_err(map_tunnel_err)?; + Ok(Json(TunnelView::from(&summary))) +} + +async fn set_enabled( + State(state): State>, + Path(id): Path, + Json(req): Json, +) -> std::result::Result, ApiError> { + sync_selected_context(&state).await?; + let summary = state + .tunnels + .set_enabled_active(&id, req.enabled) + .await + .map_err(map_tunnel_err)?; + if req.enabled + && let Some(ctx) = state.datum.selected_context() + { + state.heartbeat.register_project(ctx.project_id).await; + } + Ok(Json(TunnelView::from(&summary))) +} + +async fn quota( + State(state): State>, +) -> std::result::Result>, ApiError> { + sync_selected_context(&state).await?; + let quota = state + .tunnels + .tunnel_create_quota_active() + .await + .map_err(map_tunnel_err)?; + Ok(Json(quota)) +} + +async fn metrics( + State(state): State>, +) -> std::result::Result>, ApiError> { + sync_selected_context(&state).await?; + let tunnels = state.tunnels.list_active().await.map_err(map_tunnel_err)?; + let metrics = state.listen.metrics(); + Ok(Json( + tunnels + .iter() + .map(|tunnel| { + let (bytes_from_origin, bytes_to_origin) = + metrics_bytes_for_tunnel(metrics.as_ref(), tunnel); + TunnelMetric { + id: tunnel.id.clone(), + bytes_from_origin, + bytes_to_origin, + } + }) + .collect(), + )) +} + +fn map_tunnel_err(err: n0_error::AnyError) -> ApiError { + if err.downcast_ref::().is_some() { + return ApiError::new( + StatusCode::UNAUTHORIZED, + "Not logged in. Run `datum-connect login`.", + ); + } + let message = format!("{err:#}"); + if message.contains("No project selected") { + return ApiError::new( + StatusCode::BAD_REQUEST, + "No project selected. Run `datum-connect context set --project `.", + ); + } + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, message) +} + +pub async fn wait_for_hostname( + tunnels: &TunnelService, + tunnel_id: &str, + timeout: Duration, +) -> Result { + let deadline = Instant::now() + timeout; + loop { + let Some(summary) = tunnels.get_active(tunnel_id).await? else { + n0_error::bail_any!("Tunnel `{tunnel_id}` disappeared while waiting for a hostname"); + }; + if !summary.hostnames.is_empty() || Instant::now() >= deadline { + return Ok(summary); + } + tokio::time::sleep(HOSTNAME_POLL_INTERVAL).await; + } +} + +pub async fn stop_agent(repo: &Repo) -> Result<()> { + let Some(info) = read_agent_info(repo)? else { + n0_error::bail_any!("Agent is not running."); + }; + if !is_pid_alive(info.pid) { + remove_agent_info(repo)?; + n0_error::bail_any!("Agent is not running."); + } + if info.pid == std::process::id() { + n0_error::bail_any!("Refusing to stop the current process via agent stop"); + } + + if let Err(err) = kill_pid(info.pid, false) { + warn!("failed to send SIGTERM to agent: {err:#}"); + } + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + if !is_pid_alive(info.pid) { + let _ = remove_agent_info(repo); + return Ok(()); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + + kill_pid(info.pid, true)?; + tokio::time::sleep(Duration::from_millis(200)).await; + let _ = remove_agent_info(repo); + if is_pid_alive(info.pid) { + n0_error::bail_any!("Failed to stop agent (pid {})", info.pid); + } + Ok(()) +} + +#[derive(Clone)] +pub struct AgentClient { + http: reqwest::Client, + base: String, + token: String, + pub info: AgentInfo, +} + +impl AgentClient { + pub fn from_info(info: AgentInfo) -> Result { + let http = reqwest::Client::builder() + .user_agent(datum_http_user_agent()) + .build() + .anyerr()?; + Ok(Self { + http, + base: format!("http://127.0.0.1:{}", info.port), + token: info.token.clone(), + info, + }) + } + + pub fn connect(repo: &Repo) -> Result { + let Some(info) = running_agent_info(repo)? else { + n0_error::bail_any!("Agent is not running. Run `datum-connect agent start`."); + }; + Self::from_info(info) + } + + async fn send( + &self, + method: reqwest::Method, + path: &str, + body: Option<&impl Serialize>, + ) -> Result<(StatusCode, Vec)> { + let mut req = self + .http + .request(method, format!("{}{path}", self.base)) + .header("Authorization", format!("Bearer {}", self.token)); + if let Some(body) = body { + req = req.json(body); + } + let response = req.send().await.std_context("agent request failed")?; + let status = response.status(); + let bytes = response + .bytes() + .await + .std_context("reading agent response")? + .to_vec(); + Ok((status, bytes)) + } + + fn decode_success Deserialize<'de>>(status: StatusCode, bytes: &[u8]) -> Result { + if status.is_success() { + return serde_json::from_slice(bytes).std_context("decoding agent response"); + } + let message = serde_json::from_slice::(bytes) + .ok() + .map(|body| body.error) + .unwrap_or_else(|| String::from_utf8_lossy(bytes).trim().to_string()); + n0_error::bail_any!("{message}") + } + + pub async fn status(&self) -> Result { + let (status, bytes) = self + .send(reqwest::Method::GET, "/status", None::<&()>) + .await?; + Self::decode_success(status, &bytes) + } + + pub async fn list_tunnels(&self) -> Result> { + let (status, bytes) = self + .send(reqwest::Method::GET, "/tunnels", None::<&()>) + .await?; + Self::decode_success(status, &bytes) + } + + pub async fn get_tunnel(&self, id: &str) -> Result { + match self.get_tunnel_optional(id).await? { + Some(tunnel) => Ok(tunnel), + None => n0_error::bail_any!("Tunnel `{id}` not found"), + } + } + + pub async fn get_tunnel_optional(&self, id: &str) -> Result> { + let (status, bytes) = self + .send(reqwest::Method::GET, &format!("/tunnels/{id}"), None::<&()>) + .await?; + if status == StatusCode::NOT_FOUND { + return Ok(None); + } + Ok(Some(Self::decode_success(status, &bytes)?)) + } + + pub async fn create_tunnel(&self, req: &CreateTunnelRequest) -> Result { + let (status, bytes) = self + .send(reqwest::Method::POST, "/tunnels", Some(req)) + .await?; + Self::decode_success(status, &bytes) + } + + pub async fn delete_tunnel(&self, id: &str) -> Result { + let (status, bytes) = self + .send( + reqwest::Method::DELETE, + &format!("/tunnels/{id}"), + None::<&()>, + ) + .await?; + Self::decode_success(status, &bytes) + } + + pub async fn update_tunnel(&self, id: &str, req: &UpdateTunnelRequest) -> Result { + let (status, bytes) = self + .send(reqwest::Method::PATCH, &format!("/tunnels/{id}"), Some(req)) + .await?; + Self::decode_success(status, &bytes) + } + + pub async fn set_enabled(&self, id: &str, enabled: bool) -> Result { + let (status, bytes) = self + .send( + reqwest::Method::PUT, + &format!("/tunnels/{id}/enabled"), + Some(&SetEnabledRequest { enabled }), + ) + .await?; + Self::decode_success(status, &bytes) + } + + pub async fn quota(&self) -> Result> { + let (status, bytes) = self + .send(reqwest::Method::GET, "/quota", None::<&()>) + .await?; + Self::decode_success(status, &bytes) + } + + pub async fn metrics(&self) -> Result> { + let (status, bytes) = self + .send(reqwest::Method::GET, "/metrics", None::<&()>) + .await?; + Self::decode_success(status, &bytes) + } +} + +/// True when this process was launched as the detached listener. +pub fn wants_headless_agent() -> bool { + std::env::args().any(|arg| arg == HEADLESS_AGENT_FLAG) +} + +/// `--repo` from argv, otherwise the default app-support location. +pub fn repo_path_from_cli_args() -> PathBuf { + let mut args = std::env::args().skip(1); + while let Some(arg) = args.next() { + if arg == "--repo" { + if let Some(path) = args.next() { + return PathBuf::from(path); + } + } else if let Some(path) = arg.strip_prefix("--repo=") { + return PathBuf::from(path); + } + } + Repo::default_location() +} + +/// Spawn a detached listener using this binary, then wait until it answers. +pub async fn ensure_agent(repo: &Repo) -> Result { + if let Ok(client) = AgentClient::connect(repo) + && client.status().await.is_ok() + { + return Ok(client); + } + spawn_detached_agent(repo)?; + match wait_until_ready(repo, Duration::from_secs(20)).await { + Ok(client) => Ok(client), + Err(err) => n0_error::bail_any!( + "{err:#}. Check {} for agent logs.", + repo.agent_log_path().display() + ), + } +} + +/// Fork this executable with [`HEADLESS_AGENT_FLAG`] so the listener outlives +/// the GUI or CLI that started it. +pub fn spawn_detached_agent(repo: &Repo) -> Result<()> { + let exe = std::env::current_exe().std_context("failed to resolve current executable")?; + let log_path = repo.agent_log_path(); + let log = OpenOptions::new() + .create(true) + .append(true) + .open(&log_path) + .std_context("failed to open agent.log")?; + let log_err = log.try_clone().std_context("failed to clone agent.log")?; + let mut cmd = std::process::Command::new(exe); + cmd.arg("--repo").arg(repo.path()); + cmd.arg(HEADLESS_AGENT_FLAG); + cmd.stdin(Stdio::null()); + cmd.stdout(Stdio::from(log)); + cmd.stderr(Stdio::from(log_err)); + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + cmd.process_group(0); + } + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + const DETACHED_PROCESS: u32 = 0x00000008; + const CREATE_NEW_PROCESS_GROUP: u32 = 0x00000200; + cmd.creation_flags(DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP); + } + cmd.spawn() + .std_context("failed to spawn datum-connect agent")?; + Ok(()) +} + +async fn shutdown_signal() { + let ctrl_c = tokio::signal::ctrl_c(); + #[cfg(unix)] + { + let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("install SIGTERM handler"); + tokio::select! { + _ = ctrl_c => {} + _ = sigterm.recv() => {} + } + } + #[cfg(not(unix))] + { + let _ = ctrl_c.await; + } +} + +/// Entry point for `--headless-agent` in the CLI and the desktop app. +pub async fn run_headless_agent_from_args() -> Result<()> { + let filter = tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")); + let _ = tracing_subscriber::fmt() + .with_env_filter(filter) + .with_writer(std::io::stderr) + .try_init(); + + let repo = Repo::open_or_create(repo_path_from_cli_args()).await?; + let shutdown = CancellationToken::new(); + let shutdown_for_signal = shutdown.clone(); + tokio::spawn(async move { + shutdown_signal().await; + shutdown_for_signal.cancel(); + }); + run_agent(repo, shutdown).await +} + +pub async fn wait_until_ready(repo: &Repo, timeout: Duration) -> Result { + let deadline = Instant::now() + timeout; + let mut last_err: Option = None; + while Instant::now() < deadline { + match AgentClient::connect(repo) { + Ok(client) => match client.status().await { + Ok(_) => return Ok(client), + Err(err) => last_err = Some(format!("{err:#}")), + }, + Err(err) => last_err = Some(format!("{err:#}")), + } + tokio::time::sleep(AGENT_READY_POLL).await; + } + match last_err { + Some(err) => n0_error::bail_any!("Timed out waiting for the agent to start: {err}"), + None => n0_error::bail_any!("Timed out waiting for the agent to start"), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::tunnels::TunnelSummary; + + #[test] + fn current_pid_is_alive() { + assert!(is_pid_alive(std::process::id())); + } + + #[test] + fn bogus_pid_is_not_alive() { + assert!(!is_pid_alive(u32::MAX - 1)); + } + + #[test] + fn agent_info_roundtrip() { + let info = AgentInfo { + pid: 42, + port: 1234, + token: "abc".into(), + endpoint_id: "ep".into(), + }; + let json = serde_json::to_string(&info).unwrap(); + let parsed: AgentInfo = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed.port, 1234); + assert_eq!(parsed.token, "abc"); + } + + #[test] + fn tunnel_view_prefers_named_hostname() { + let summary = TunnelSummary { + id: "t1".into(), + label: "app".into(), + endpoint: "http://127.0.0.1:4123".into(), + hostnames: vec![ + "v4.example.iroh.datum.net".into(), + "vast-gold-mine.iroh.datum.net".into(), + ], + enabled: true, + accepted: true, + programmed: true, + }; + let view = TunnelView::from(&summary); + assert_eq!( + view.url.as_deref(), + Some("https://vast-gold-mine.iroh.datum.net") + ); + } + + #[test] + fn tunnel_view_empty_hostnames() { + let summary = TunnelSummary { + id: "t1".into(), + label: "app".into(), + endpoint: "http://127.0.0.1:4123".into(), + hostnames: vec![], + enabled: true, + accepted: false, + programmed: false, + }; + assert!(TunnelView::from(&summary).url.is_none()); + } +} diff --git a/lib/src/datum_cloud.rs b/lib/src/datum_cloud.rs index 3d2ce6b..1499258 100644 --- a/lib/src/datum_cloud.rs +++ b/lib/src/datum_cloud.rs @@ -6,6 +6,7 @@ use chrono::{Duration, Utc}; use n0_error::{Result, StackResultExt, StdResultExt}; use n0_future::{BufferedStreamExt, TryStreamExt, task::AbortOnDropHandle}; use rand::Rng; +use serde::Serialize; use tokio::sync::{Mutex, watch}; use tracing::warn; @@ -111,6 +112,11 @@ impl DatumCloudClient { self.session.set_selected_context(selected_context).await } + /// Re-read `selected_context.yml` so a long-lived agent picks up CLI/GUI changes. + pub async fn reload_selected_context(&self) -> Result> { + self.session.reload_selected_context().await + } + fn project_control_plane_url(&self, project_id: &str) -> String { format!( "{}/apis/resourcemanager.miloapis.com/v1alpha1/projects/{project_id}/control-plane", @@ -563,6 +569,19 @@ impl SessionStateWrapper { Ok(()) } + async fn reload_selected_context(&self) -> Result> { + let Some(repo) = self.repo.as_ref() else { + return Ok(self.selected_context()); + }; + let selected = repo.read_selected_context().await?; + let current = self.selected_context(); + if current != selected { + self.selected_context.store(Arc::new(selected.clone())); + let _ = self.selected_context_tx.send(selected.clone()); + } + Ok(selected) + } + fn orgs_projects(&self) -> Vec { self.orgs_projects.load_full().as_ref().clone() } @@ -582,25 +601,107 @@ impl SessionStateWrapper { } } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct Organization { pub resource_id: String, pub display_name: String, pub r#type: String, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct OrganizationWithProjects { pub org: Organization, pub projects: Vec, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct Project { pub resource_id: String, pub display_name: String, } +/// Resolve a project (and optional org) name or id into a selected context. +/// +/// `project` and `org` match `resource_id` first, then display name +/// (case-insensitive). Ambiguous matches return an error listing candidates. +pub fn resolve_selected_context( + orgs: &[OrganizationWithProjects], + project: &str, + org: Option<&str>, +) -> Result { + let project_key = project.trim(); + if project_key.is_empty() { + n0_error::bail_any!("Project is required"); + } + + let filtered: Vec<&OrganizationWithProjects> = match org + .map(str::trim) + .filter(|s| !s.is_empty()) + { + Some(org_key) => { + let matches: Vec<&OrganizationWithProjects> = orgs + .iter() + .filter(|entry| org_matches(&entry.org, org_key)) + .collect(); + if matches.is_empty() { + n0_error::bail_any!("No organization matched `{org_key}`"); + } + if matches.len() > 1 { + let names = matches + .iter() + .map(|entry| format!("{} ({})", entry.org.display_name, entry.org.resource_id)) + .collect::>() + .join(", "); + n0_error::bail_any!("Multiple organizations matched `{org_key}`: {names}"); + } + matches + } + None => orgs.iter().collect(), + }; + + let mut hits: Vec<(&Organization, &Project)> = Vec::new(); + for entry in filtered { + for proj in &entry.projects { + if project_matches(proj, project_key) { + hits.push((&entry.org, proj)); + } + } + } + + match hits.as_slice() { + [] => n0_error::bail_any!("No project matched `{project_key}`"), + [(org, proj)] => Ok(SelectedContext { + org_id: org.resource_id.clone(), + org_name: org.display_name.clone(), + project_id: proj.resource_id.clone(), + project_name: proj.display_name.clone(), + }), + rest => { + let names = rest + .iter() + .map(|(org, proj)| { + format!( + "{} / {} ({} / {})", + org.display_name, proj.display_name, org.resource_id, proj.resource_id + ) + }) + .collect::>() + .join(", "); + n0_error::bail_any!( + "Multiple projects matched `{project_key}`: {names}. Pass --org to disambiguate." + ) + } + } +} + +fn org_matches(org: &Organization, key: &str) -> bool { + org.resource_id == key || org.display_name.eq_ignore_ascii_case(key) +} + +fn project_matches(project: &Project, key: &str) -> bool { + project.resource_id == key || project.display_name.eq_ignore_ascii_case(key) +} + /// Summary of an IAM Role for listing (e.g. in invite dialog). #[derive(Debug, Clone)] pub struct RoleSummary { @@ -953,3 +1054,76 @@ mod auth_failure_tests { ); } } + +#[cfg(test)] +mod resolve_context_tests { + use super::*; + + fn org(id: &str, name: &str, projects: Vec) -> OrganizationWithProjects { + OrganizationWithProjects { + org: Organization { + resource_id: id.to_string(), + display_name: name.to_string(), + r#type: "standard".to_string(), + }, + projects, + } + } + + fn project(id: &str, name: &str) -> Project { + Project { + resource_id: id.to_string(), + display_name: name.to_string(), + } + } + + #[test] + fn matches_project_id() { + let orgs = vec![org( + "org-1", + "Acme", + vec![project("proj-1", "Web"), project("proj-2", "API")], + )]; + let ctx = resolve_selected_context(&orgs, "proj-2", None).unwrap(); + assert_eq!(ctx.project_id, "proj-2"); + assert_eq!(ctx.project_name, "API"); + assert_eq!(ctx.org_id, "org-1"); + } + + #[test] + fn matches_project_display_name_case_insensitive() { + let orgs = vec![org("org-1", "Acme", vec![project("proj-1", "Web")])]; + let ctx = resolve_selected_context(&orgs, "web", None).unwrap(); + assert_eq!(ctx.project_id, "proj-1"); + } + + #[test] + fn disambiguates_with_org() { + let orgs = vec![ + org("org-1", "Acme", vec![project("p1", "Web")]), + org("org-2", "Beta", vec![project("p2", "Web")]), + ]; + let ctx = resolve_selected_context(&orgs, "Web", Some("Beta")).unwrap(); + assert_eq!(ctx.project_id, "p2"); + assert_eq!(ctx.org_id, "org-2"); + } + + #[test] + fn errors_on_ambiguous_project() { + let orgs = vec![ + org("org-1", "Acme", vec![project("p1", "Web")]), + org("org-2", "Beta", vec![project("p2", "Web")]), + ]; + let err = resolve_selected_context(&orgs, "Web", None).unwrap_err(); + let msg = format!("{err:#}"); + assert!(msg.contains("Multiple projects"), "{msg}"); + assert!(msg.contains("--org"), "{msg}"); + } + + #[test] + fn errors_when_missing() { + let orgs = vec![org("org-1", "Acme", vec![project("p1", "Web")])]; + let err = resolve_selected_context(&orgs, "missing", None).unwrap_err(); + assert!(format!("{err:#}").contains("No project matched")); + } +} diff --git a/lib/src/datum_cloud/auth.rs b/lib/src/datum_cloud/auth.rs index 9e11af5..04f366e 100644 --- a/lib/src/datum_cloud/auth.rs +++ b/lib/src/datum_cloud/auth.rs @@ -1416,10 +1416,12 @@ mod redirect_server { let hero_b64 = BASE64.encode(LOGIN_SUCCESS_PNG); let font_b64 = BASE64.encode(ALLIANCE_NO1_REGULAR_TTF); + let favicon_light_b64 = BASE64.encode(FAVICON_LIGHT_32); let favicon_dark_b64 = BASE64.encode(FAVICON_DARK_32); let html = OAUTH_REDIRECT_HTML .replace("{{HERO_B64}}", &hero_b64) .replace("{{FONT_B64}}", &font_b64) + .replace("{{FAVICON_LIGHT_B64}}", &favicon_light_b64) .replace("{{FAVICON_DARK_B64}}", &favicon_dark_b64); axum::response::Html(html) @@ -1435,6 +1437,14 @@ mod redirect_server { rel="icon" type="image/png" sizes="32x32" + media="(prefers-color-scheme: light)" + href="data:image/png;base64,{{FAVICON_LIGHT_B64}}" + /> +