mod config;
mod config_tui;
mod doctor;
mod extend;
mod init;
mod keepalive;
mod lease_create;
mod leases;
mod login;
mod picker;
mod pools;
mod purge;
mod release;
pub(crate) mod sandbox;
pub(crate) mod sandbox_transport;
mod select;
pub(crate) mod session;
mod ssh_proxy;
mod ssh_setup;
mod state;
mod status;
mod version;
pub(crate) mod vnc;
mod with_lease;
use clap::ValueEnum;
use serde::Serialize;
pub use config::{
SetTargetCommand, config_current_target, config_export, config_import, config_list_targets,
config_set_target, config_show, config_use_target,
};
pub use config_tui::run_config_tui as config_interactive;
pub use doctor::doctor;
pub use extend::extend;
pub use init::{InitCommand, init};
pub use lease_create::{LeaseCreateCommand, lease_create};
pub use login::{login, logout};
pub use purge::purge;
pub use release::release;
pub use ssh_proxy::{SshProxyCommand, ssh_config, ssh_proxy};
pub use status::status;
pub use version::version;
pub use with_lease::{WithLeaseCommand, with_lease};
pub async fn require_pool_capability(
pool: &str,
capability: &str,
target_override: Option<&str>,
endpoint_override: Option<&str>,
output: OutputFormat,
) -> anyhow::Result<()> {
let config = config::CliConfig::load()?.resolve(target_override, endpoint_override)?;
let pool = pools::fetch_pool_for_config_with_output(&config, pool, output).await?;
if !pool.supports(capability) {
anyhow::bail!(
"pool {} allocates {} resources, which do not support {capability}; available capabilities: {}",
pool.name,
pool.resource_kind,
pool.capabilities.join(", ")
);
}
Ok(())
}
pub async fn pick_lease_with_capability(
capability: &str,
target_override: Option<&str>,
endpoint_override: Option<&str>,
output: OutputFormat,
) -> anyhow::Result<String> {
let config = config::CliConfig::load()?.resolve(target_override, endpoint_override)?;
select::resolve_lease_for_capability(&config, capability, output).await
}
pub async fn require_lease_capability(
selector: &str,
capability: &str,
target_override: Option<&str>,
endpoint_override: Option<&str>,
output: OutputFormat,
) -> anyhow::Result<String> {
let config = config::CliConfig::load()?.resolve(target_override, endpoint_override)?;
let leases = leases::fetch_all_leases_with_output(&config, output).await?;
let by_id: Vec<_> = leases.iter().filter(|lease| lease.id == selector).collect();
let by_alias: Vec<_> = leases
.iter()
.filter(|lease| lease.alias.as_deref() == Some(selector))
.collect();
let by_pool: Vec<_> = leases
.iter()
.filter(|lease| lease.profile == selector)
.collect();
let matches = if !by_id.is_empty() {
by_id
} else if !by_alias.is_empty() {
by_alias
} else {
by_pool
};
let lease = match matches.as_slice() {
[lease] => lease,
[] => anyhow::bail!("no lease matches '{selector}'"),
many => anyhow::bail!(
"'{selector}' matches multiple leases: {}",
many.iter()
.map(|lease| lease.id.as_str())
.collect::<Vec<_>>()
.join(", ")
),
};
if !lease
.capabilities
.iter()
.any(|candidate| candidate == capability)
{
anyhow::bail!(
"lease {} is a {} resource and does not support {capability}; available capabilities: {}",
lease.id,
lease.resource_kind,
lease.capabilities.join(", ")
);
}
Ok(lease.id.clone())
}
use config::{AuthMode, ResolvedConfig};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum AuthInteraction {
Interactive,
NonInteractive,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
pub enum OutputFormat {
Text,
Json,
}
pub(crate) fn cli_version() -> &'static str {
option_env!("BUILD_VERSION").unwrap_or(env!("CARGO_PKG_VERSION"))
}
pub(crate) fn print_json<T: Serialize>(value: &T) -> anyhow::Result<()> {
println!("{}", serde_json::to_string_pretty(value)?);
Ok(())
}
pub(crate) async fn get_auth_header(
config: &ResolvedConfig,
method: &str,
path: &str,
body: &[u8],
) -> anyhow::Result<Option<String>> {
get_auth_header_with_interaction(config, method, path, body, AuthInteraction::Interactive).await
}
pub(crate) async fn get_auth_header_noninteractive(
config: &ResolvedConfig,
method: &str,
path: &str,
body: &[u8],
) -> anyhow::Result<Option<String>> {
get_auth_header_with_interaction(config, method, path, body, AuthInteraction::NonInteractive)
.await
}
pub(crate) async fn get_auth_header_for_output(
config: &ResolvedConfig,
method: &str,
path: &str,
body: &[u8],
output: OutputFormat,
) -> anyhow::Result<Option<String>> {
match output {
OutputFormat::Text => get_auth_header(config, method, path, body).await,
OutputFormat::Json => get_auth_header_noninteractive(config, method, path, body).await,
}
}
async fn get_auth_header_with_interaction(
config: &ResolvedConfig,
method: &str,
path: &str,
body: &[u8],
interaction: AuthInteraction,
) -> anyhow::Result<Option<String>> {
match &config.auth {
AuthMode::None => Ok(None),
AuthMode::Token => match &config.token {
Some(t) => Ok(Some(format!("Bearer {t}"))),
None => {
anyhow::bail!("Auth mode is 'token' but no token configured. Run: kobe config edit")
}
},
AuthMode::Oidc => {
let service_config = login::discover_pinned(&config.endpoint).await?;
let token = match interaction {
AuthInteraction::Interactive => {
kunobi_auth::client::AuthClient::new(service_config)?
.token()
.await?
}
AuthInteraction::NonInteractive => {
oidc_token_noninteractive(&service_config).await?
}
};
Ok(Some(format!("Bearer {token}")))
}
AuthMode::Ssh => {
let client = kunobi_auth::client::AuthClient::with_ssh(config.ssh_fingerprint.clone())?;
let audience = discover_ssh_audience(config).await?;
tofu_check(&config.endpoint, &audience, interaction).await?;
let header = client.authorize(&audience, method, path, body).await?;
Ok(Some(header))
}
}
}
async fn oidc_token_noninteractive(
config: &kunobi_auth::client::ServiceConfig,
) -> anyhow::Result<String> {
use kunobi_auth::client::TokenStore;
let store = TokenStore::new()?;
let Some(stored) = store.load(&config.issuer)? else {
anyhow::bail!("OIDC login is required; run `kobe login` before using --output json")
};
if !stored.is_expired() {
return Ok(stored.id_token);
}
let Some(refresh_token) = stored.refresh_token.as_deref() else {
anyhow::bail!(
"OIDC credentials expired without a refresh token; run `kobe login` before using --output json"
)
};
let mut refreshed = kunobi_auth::client::oidc::refresh(
&config.issuer,
&config.client_id,
&config.redirect_uri,
refresh_token,
)
.await
.map_err(|error| {
anyhow::anyhow!(
"OIDC refresh failed without interactive fallback ({error}); run `kobe login`"
)
})?;
refreshed.extra = stored.extra;
let token = refreshed.id_token.clone();
store.save(&refreshed)?;
Ok(token)
}
fn audience_cache() -> &'static std::sync::Mutex<std::collections::HashMap<String, String>> {
static CACHE: std::sync::OnceLock<std::sync::Mutex<std::collections::HashMap<String, String>>> =
std::sync::OnceLock::new();
CACHE.get_or_init(Default::default)
}
async fn discover_ssh_audience(config: &ResolvedConfig) -> anyhow::Result<String> {
let endpoint = config.endpoint.as_str();
if let Ok(cache) = audience_cache().lock()
&& let Some(hit) = cache.get(endpoint)
{
return Ok(hit.clone());
}
for attempt in 0..2 {
let resp: serde_json::Value = authed_client()
.get(format!("{endpoint}/v1/status"))
.send()
.await
.reaching(config)?
.json()
.await?;
if let Some(methods) = resp["auth"]["methods"].as_array() {
for method in methods {
if method["type"].as_str() == Some("ssh")
&& let Some(audience) = method["audience"].as_str()
{
if let Ok(mut cache) = audience_cache().lock() {
cache.insert(endpoint.to_string(), audience.to_string());
}
return Ok(audience.to_string());
}
}
}
if attempt == 0 {
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
}
}
anyhow::bail!(
"Server at {endpoint} has no SSH auth method configured. \
Check that an AccessPolicy with ssh auth exists in the cluster."
)
}
const SSH_PINNED_ISSUER: &str = "ssh";
async fn tofu_check(
endpoint: &str,
audience: &str,
interaction: AuthInteraction,
) -> anyhow::Result<()> {
let endpoint = endpoint.to_string();
let audience = audience.to_string();
tokio::task::spawn_blocking(move || tofu_check_blocking(&endpoint, &audience, interaction))
.await
.map_err(|error| anyhow::anyhow!("SSH trust check task failed: {error}"))?
}
fn tofu_check_blocking(
endpoint: &str,
audience: &str,
interaction: AuthInteraction,
) -> anyhow::Result<()> {
let store = kunobi_auth::client::TofuStore::new()?;
apply_tofu_result(
&store,
store.verify(endpoint, SSH_PINNED_ISSUER, audience)?,
interaction,
)
}
fn apply_tofu_result(
store: &kunobi_auth::client::TofuStore,
result: kunobi_auth::client::TofuResult,
interaction: AuthInteraction,
) -> anyhow::Result<()> {
match result {
kunobi_auth::client::TofuResult::Trusted => Ok(()),
kunobi_auth::client::TofuResult::FirstConnect { endpoint, audience }
if interaction == AuthInteraction::NonInteractive =>
{
anyhow::bail!(
"SSH trust is not established for {endpoint} (audience {audience}); rerun in text mode to review it"
)
}
kunobi_auth::client::TofuResult::FirstConnect { endpoint, audience } => {
eprintln!();
eprintln!("Connecting to {endpoint}");
eprintln!(" Audience: {audience}");
eprintln!();
eprint!("Trust this service? [y/N] ");
let mut input = String::new();
std::io::stdin().read_line(&mut input)?;
if input.trim().eq_ignore_ascii_case("y") {
store.trust(&endpoint, SSH_PINNED_ISSUER, &audience)?;
Ok(())
} else {
anyhow::bail!("Connection refused by user")
}
}
kunobi_auth::client::TofuResult::AudienceChanged {
endpoint,
previous,
current,
} if interaction == AuthInteraction::NonInteractive => {
anyhow::bail!(
"SSH audience changed for {endpoint} from {previous} to {current}; rerun in text mode to review it"
)
}
kunobi_auth::client::TofuResult::AudienceChanged {
endpoint,
previous,
current,
} => {
eprintln!();
eprintln!("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
eprintln!("@ WARNING: SERVICE AUDIENCE HAS CHANGED! @");
eprintln!("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
eprintln!("The audience for {endpoint} changed:");
eprintln!(" Previous: {previous}");
eprintln!(" Current: {current}");
eprintln!("This could mean the service was reconfigured, or it");
eprintln!("could indicate a man-in-the-middle attack.");
eprint!("Continue? [y/N] ");
let mut input = String::new();
std::io::stdin().read_line(&mut input)?;
if input.trim().eq_ignore_ascii_case("y") {
store.trust(&endpoint, SSH_PINNED_ISSUER, ¤t)?;
Ok(())
} else {
anyhow::bail!("Connection refused by user")
}
}
kunobi_auth::client::TofuResult::IssuerChanged {
endpoint,
previous,
current,
} => {
anyhow::bail!(
"SSH trust for {endpoint} is pinned to issuer {previous}, but the service now \
presents {current}. Remove the pin for {endpoint} from the trust store only if \
you know why it changed."
)
}
other => anyhow::bail!("Unhandled SSH trust result: {other:?}"),
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Unreachable {
Dns,
Refused,
TimedOut,
Tls,
Other,
}
impl Unreachable {
pub(crate) fn summary(self) -> &'static str {
match self {
Self::Dns => "the host name does not resolve",
Self::Refused => "nothing is listening there",
Self::TimedOut => "the connection timed out",
Self::Tls => "the TLS handshake failed",
Self::Other => "the connection failed",
}
}
pub(crate) fn hint(self) -> &'static str {
match self {
Self::Dns => {
"A private cluster resolves only from its network: connect the VPN, or check the endpoint for a typo."
}
Self::Refused => {
"The name resolves, so the address is reachable but nothing answered. Check that the kobe ingress is up."
}
Self::TimedOut => {
"The address is routable but silent, which usually means a firewall or a missing VPN route."
}
Self::Tls => {
"The server's certificate was rejected. Check that the endpoint host matches the certificate."
}
Self::Other => "Check the endpoint and that the kobe server is reachable from here.",
}
}
}
pub(crate) fn classify_unreachable(error: &reqwest::Error) -> Unreachable {
if error.is_timeout() {
return Unreachable::TimedOut;
}
let mut chain = String::new();
let mut source: Option<&(dyn std::error::Error + 'static)> = Some(error);
while let Some(current) = source {
chain.push_str(¤t.to_string().to_ascii_lowercase());
chain.push(' ');
source = current.source();
}
if chain.contains("dns error") || chain.contains("failed to lookup address") {
Unreachable::Dns
} else if chain.contains("connection refused") {
Unreachable::Refused
} else if chain.contains("timed out") {
Unreachable::TimedOut
} else if chain.contains("certificate") || chain.contains("tls") {
Unreachable::Tls
} else {
Unreachable::Other
}
}
pub(crate) fn unreachable_message(
endpoint: &str,
target: Option<&str>,
kind: Unreachable,
) -> String {
let host = endpoint
.split_once("://")
.map_or(endpoint, |(_, rest)| rest)
.split('/')
.next()
.unwrap_or(endpoint);
let origin = match target {
Some(target) => format!("target {target}"),
None => "target (none: endpoint given directly)".to_string(),
};
format!(
"cannot reach kobe at {host}: {summary}\n\n endpoint {endpoint}\n {origin}\n\n {hint}\n `kobe config view` shows where this endpoint came from.",
summary = kind.summary(),
hint = kind.hint(),
)
}
pub(crate) trait Reaching<T> {
fn reaching(self, config: &ResolvedConfig) -> anyhow::Result<T>;
}
impl<T> Reaching<T> for Result<T, reqwest::Error> {
fn reaching(self, config: &ResolvedConfig) -> anyhow::Result<T> {
self.map_err(|error| {
if error.is_builder() || error.is_body() || error.is_decode() {
return anyhow::Error::new(error);
}
anyhow::anyhow!(unreachable_message(
&config.endpoint,
config.target.as_deref(),
classify_unreachable(&error),
))
})
}
}
pub(crate) fn stdout_color() -> bool {
static COLOR: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*COLOR.get_or_init(|| {
std::io::IsTerminal::is_terminal(&std::io::stdout())
&& std::env::var_os("NO_COLOR").is_none()
})
}
pub(crate) fn styled(sgr: &str, text: impl std::fmt::Display) -> String {
if stdout_color() {
format!("\x1b[{sgr}m{text}\x1b[0m")
} else {
text.to_string()
}
}
pub(crate) fn authed_client() -> reqwest::Client {
static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
CLIENT.get_or_init(reqwest::Client::new).clone()
}
pub(crate) fn with_auth(
builder: reqwest::RequestBuilder,
auth_header: &Option<String>,
) -> reqwest::RequestBuilder {
match auth_header {
Some(h) => builder.header("Authorization", h),
None => builder,
}
}
pub(crate) fn lease_request_paths(lease_id: &str) -> Vec<String> {
let mut paths = vec![format!("/v1/leases/{lease_id}")];
if extend::is_sandbox_lease(lease_id) {
paths.push(format!("/v1/sandbox-leases/{lease_id}"));
}
paths
}
pub(crate) async fn send_lease_request(
config: &ResolvedConfig,
method: reqwest::Method,
lease_id: &str,
body: Option<&[u8]>,
) -> anyhow::Result<reqwest::Response> {
let paths = lease_request_paths(lease_id);
let client = authed_client();
let mut attempts = paths.iter().peekable();
loop {
let path = attempts.next().expect("at least one lease path");
let token = get_auth_header(config, method.as_str(), path, b"").await?;
let mut request = with_auth(
client.request(method.clone(), format!("{}{path}", config.endpoint)),
&token,
);
if let Some(body) = body {
request = request
.header("Content-Type", "application/json")
.body(body.to_vec());
}
let response = request.send().await.reaching(config)?;
if response.status() == reqwest::StatusCode::NOT_FOUND && attempts.peek().is_some() {
continue;
}
return Ok(response);
}
}
pub(crate) fn advertised_auth_methods(status: &serde_json::Value) -> Option<Vec<String>> {
let methods = status["auth"]["methods"].as_array()?;
let mut types: Vec<String> = Vec::new();
for method in methods {
if let Some(kind) = method["type"].as_str()
&& !types.iter().any(|seen| seen == kind)
{
types.push(kind.to_string());
}
}
Some(types)
}
#[cfg(test)]
mod auth_method_tests {
use super::advertised_auth_methods;
use serde_json::json;
#[test]
fn methods_are_read_from_each_entrys_type() {
let status = json!({"auth": {"methods": [
{"type": "oidc", "issuer": "https://token.actions.githubusercontent.com", "description": "github-actions"},
{"type": "oidc", "issuer": "https://clerk.kunobi.com", "clientId": "abc", "description": "kunobi-oauth"},
{"type": "ssh", "audience": "kobe"},
]}});
assert_eq!(
advertised_auth_methods(&status),
Some(vec!["oidc".to_string(), "ssh".to_string()])
);
}
#[test]
fn a_missing_list_is_none_and_an_empty_one_is_empty() {
assert_eq!(advertised_auth_methods(&json!({})), None);
assert_eq!(
advertised_auth_methods(&json!({"auth": {"methods": []}})),
Some(vec![])
);
}
#[test]
fn entries_without_a_type_are_skipped() {
let status =
json!({"auth": {"methods": [{"issuer": "x"}, "oidc", {"type": 3}, {"type": "token"}]}});
assert_eq!(
advertised_auth_methods(&status),
Some(vec!["token".to_string()])
);
}
}
#[cfg(test)]
mod reachability_tests {
use super::*;
#[test]
fn an_unreachable_message_names_the_host_target_and_remedy() {
let message = unreachable_message(
"https://kobe.zur1-worker1.int-pro.zondax.io",
Some("int-pro"),
Unreachable::Dns,
);
assert!(
message.starts_with("cannot reach kobe at kobe.zur1-worker1.int-pro.zondax.io: "),
"{message}"
);
assert!(
message.contains("the host name does not resolve"),
"{message}"
);
assert!(message.contains("int-pro"), "{message}");
assert!(message.contains("VPN"), "{message}");
assert!(!message.contains("nodename"), "{message}");
assert!(!message.contains("servname"), "{message}");
let direct = unreachable_message("http://127.0.0.1:8080", None, Unreachable::Refused);
assert!(direct.contains("endpoint given directly"), "{direct}");
assert!(direct.contains("nothing is listening there"), "{direct}");
assert!(direct.contains("at 127.0.0.1:8080:"), "{direct}");
}
#[test]
fn resolver_and_refusal_are_told_apart() {
assert_eq!(Unreachable::Dns.summary(), "the host name does not resolve");
assert_eq!(Unreachable::Refused.summary(), "nothing is listening there");
assert_ne!(Unreachable::Dns.hint(), Unreachable::Refused.hint());
for kind in [
Unreachable::Dns,
Unreachable::Refused,
Unreachable::TimedOut,
Unreachable::Tls,
Unreachable::Other,
] {
assert!(!kind.hint().is_empty(), "{kind:?} has no hint");
assert!(!kind.summary().is_empty(), "{kind:?} has no summary");
}
}
}
#[cfg(test)]
mod auth_tests {
use super::*;
#[test]
fn noninteractive_tofu_never_accepts_a_promptable_decision() {
let directory = tempfile::tempdir().unwrap();
let store = kunobi_auth::client::TofuStore::with_path(directory.path().join("known.json"));
let first = kunobi_auth::client::TofuResult::FirstConnect {
endpoint: "https://kobe.example".into(),
audience: "kobe".into(),
};
assert!(
apply_tofu_result(&store, first, AuthInteraction::NonInteractive)
.unwrap_err()
.to_string()
.contains("rerun in text mode")
);
let changed = kunobi_auth::client::TofuResult::AudienceChanged {
endpoint: "https://kobe.example".into(),
previous: "old".into(),
current: "new".into(),
};
assert!(
apply_tofu_result(&store, changed, AuthInteraction::NonInteractive)
.unwrap_err()
.to_string()
.contains("rerun in text mode")
);
apply_tofu_result(
&store,
kunobi_auth::client::TofuResult::Trusted,
AuthInteraction::NonInteractive,
)
.unwrap();
}
}